alkcall 0.8.0

//! `ChannelClient` — the client-side type for a channels connection
//! (ADR-043). Transport-agnostic `from_connection` primary: takes an
//! established `Connection` (the consumer dials the transport and
//! establishes the channels ALPN), runs the demux/mux, and exposes
//! `open_channel(alpn, params)` to open data channels via the
//! per-ALPN open ops on channel 0.
//!
//! Channel 0 runs in single-stream call mode (ADR-036 amendment):
//! all `EventEnvelope` frames for call operations are multiplexed on
//! channel 0's one `BiStream`. `CallConnection::new_single_stream`
//! holds the framed write half; the client spawns a read pump that
//! reads `EventEnvelope` frames off channel 0's reassembled read half
//! (fed by the demux) and routes them into the `PendingRequestMap`
//! via `dispatch_envelope`.
//!
//! The dial (TLS, QUIC, WebSocket) lives in the consumer —
//! `ChannelClient` is transport-agnostic by construction (ADR-043).
//!
//! See `docs/architecture/channel-client.md` for the spec.

use std::sync::Arc;

use serde_json::Value;
use tokio::sync::Mutex;

use crate::core::types::{Connection, StreamError};
use crate::protocol::connection::CallConnection;
use crate::protocol::wire::{CallError, ResponseEnvelope};
use crate::registry::registration::OperationRegistry;

use super::manager::{ChannelManager, ChannelSide};
use super::mux::MuxRunner;
use super::reassembly::{MpscRecvStream, MpscSendStream};

/// The typed error from [`ChannelClient::open_channel`] (ADR-049 §4 —
/// review 006 N-1). The open op's `CallError` — including the
/// `channel:open_failed` code with `details: { reason, message }`
/// (ADR-049 §3) — is carried verbatim instead of flattened into a
/// string, so the consumer can branch on the establishment-failure
/// reason (`dial_failed`, `unknown_resource`, `resource_shortage`,
/// `handler_error`, `timeout`).
#[derive(Debug, Clone, PartialEq, thiserror::Error)]
pub enum ChannelOpenError {
    /// The open op failed — the `CallError` carries the wire code and
    /// typed `details` (`channel:open_failed` + `details.reason` per
    /// ADR-049 §3, `channel:too_many_channels`, `FORBIDDEN`, …).
    #[error("open op failed: {error:?}")]
    CallFailed { error: CallError },
    /// The open op succeeded but the response carried no `channel_id`
    /// (malformed responder).
    #[error("open op response missing channel_id")]
    MissingChannelId,
    /// The open op succeeded but local channel adoption failed (the
    /// per-connection cap, or an ID collision).
    #[error("adopt_channel failed: {0}")]
    AdoptFailed(#[from] super::manager::ManagerError),
}

impl ChannelOpenError {
    /// The wire `CallError` behind this failure, when the open op
    /// itself failed. `None` for the local-only variants
    /// (`MissingChannelId` comes from a success envelope;
    /// `AdoptFailed` is post-reply).
    pub fn call_error(&self) -> Option<&CallError> {
        match self {
            ChannelOpenError::CallFailed { error } => Some(error),
            ChannelOpenError::MissingChannelId | ChannelOpenError::AdoptFailed(_) => None,
        }
    }

    /// The establishment-failure reason code — `details.reason` of a
    /// `channel:open_failed` error (`dial_failed`, `unknown_resource`,
    /// `resource_shortage`, `handler_error`, `timeout`). `None` for
    /// any other failure shape.
    pub fn establishment_reason(&self) -> Option<&str> {
        match self {
            ChannelOpenError::CallFailed { error }
                if error.code == super::operations::CHANNEL_OPEN_FAILED =>
            {
                error
                    .details
                    .as_ref()
                    .and_then(|d| d.get("reason"))
                    .and_then(|r| r.as_str())
            }
            _ => None,
        }
    }
}

/// The client-side handle for a channels connection. Constructed via
/// [`ChannelClient::from_connection`] from an established
/// `Connection`. Holds the `ChannelManager` (for the demux/mux state)
/// and the `CallConnection` (for calling open ops on channel 0).
///
/// The consumer dials the transport and establishes the
/// `alk/channels` ALPN, then hands the `Connection` to
/// `from_connection`. The client runs the demux/mux in spawned tasks
/// and exposes `open_channel` to call the per-ALPN open ops
/// (`channels/<alpn>/sub`, `channels/<alpn>/pub`) on channel 0.
pub struct ChannelClient {
    manager: ChannelManager,
    call_connection: Arc<Mutex<Option<Arc<CallConnection>>>>,
}

/// Opt-in serving configuration for a `ChannelClient` (review 004
/// F-04). A pure consumer has no registry to serve — `from_connection`
/// keeps the resolution-only read pump. A consumer that also serves
/// ops passes its registry (and identity provider) via
/// [`ServingConfig`]; the read pump becomes the full-duplex serving
/// loop (`Dispatcher::serve_single_stream`), so inbound
/// `call.requested` frames from the peer dispatch and resolve instead
/// of being dropped.
///
/// The caller identity for dispatch (CF-005): when the serving loop
/// dispatches an inbound `call.requested`, `Dispatcher::resolve_identity`
/// consults, in precedence order — (1) the payload `auth_token`
/// resolved through `identity_provider` (the hub-forwarding /
/// browser-token path), (2) `identity` if set (an explicit override),
/// (3) the transport connection's identity, if the consumer called
/// `Connection::set_identity` after dialing (the mTLS/QUIC key-based
/// path). When none is available the dispatch runs identity-less and
/// `AccessControl::check` fails closed (`FORBIDDEN`).
pub struct ServingConfig {
    pub registry: Arc<OperationRegistry>,
    pub identity_provider: Arc<dyn crate::core::auth::IdentityProvider>,
    /// An explicit caller identity for the serving dispatch (the
    /// connect-side seam, CF-005 remediation (a)). `None` (the
    /// default) falls through to the transport connection's identity
    /// (CF-005 remediation (b) — propagated automatically when the
    /// consumer set one before calling `from_connection_with_serving`).
    pub identity: Option<crate::core::auth::Identity>,
}

impl Default for ServingConfig {
    fn default() -> Self {
        Self {
            registry: Arc::new(OperationRegistry::new()),
            identity_provider: Arc::new(crate::core::auth::NoopIdentityProvider),
            identity: None,
        }
    }
}

impl ChannelClient {
    /// Construct from an established `Connection` (the consumer dials
    /// the transport and establishes the `alk/channels` ALPN).
    /// Installs channel 0 (pre-negotiated as `alk/call`,
    /// ADR-036), wraps it as a `CallConnection` in single-stream call
    /// mode (ADR-036 amendment), spawns the demux and mux tasks, and
    /// returns the client.
    ///
    /// The call dispatch loop on channel 0 is driven by the client:
    /// a read pump task reads `EventEnvelope` frames off channel 0's
    /// reassembled read half (fed by the demux) and routes them into
    /// the `PendingRequestMap` via `dispatch_envelope`, resolving
    /// pending calls. The `CallConnection` holds the framed write
    /// half so `call_open_op` can write `call.requested` frames.
    ///
    /// Serving (review 004 F-04): with `serving: None` (the default)
    /// the read pump resolves responses only — inbound
    /// `call.requested` frames are dropped (a pure consumer has no
    /// registry to serve). With `Some(ServingConfig { .. })` the read
    /// pump is the full-duplex serving loop
    /// (`Dispatcher::serve_single_stream`): inbound requests dispatch
    /// against the configured registry and resolve back to the peer,
    /// and outbound pendings still resolve. This is the opt-in that
    /// makes the connect side a serving half on channel 0 (ADR-022 §2
    /// — both sides can be both).
    ///
    /// Caller identity for the serving dispatch (CF-005): the
    /// `ServingConfig.identity` override wins; otherwise the transport
    /// connection's identity (`Connection::identity`, set by the
    /// consumer after dialing from the transport-authenticated peer)
    /// propagates to channel 0. The payload `auth_token` →
    /// `ServingConfig.identity_provider` path still takes precedence
    /// over both (ADR-017 §7). With no identity from any path the
    /// dispatch runs identity-less and scope-gated ops fail closed
    /// (`FORBIDDEN`).
    pub async fn from_connection_with_serving(
        connection: Connection,
        serving: Option<ServingConfig>,
    ) -> Result<Self, StreamError> {
        let remote_addr = connection.remote_addr();
        let bidi = connection.accept_bi().await?;
        let (reader, writer) = tokio::io::split(bidi);
        let (mux_handle, mux_runner) = MuxRunner::new(Box::new(writer));

        // Spawn the mux runner BEFORE installing channel 0 —
        // `install_channel_zero` calls `mux.register(0).await` which
        // needs the runner to be draining `new_pumps`.
        let _mux_task = tokio::spawn(async move {
            if let Err(e) = mux_runner.run().await {
                tracing::warn!(error = %e, "channel client: mux runner ended with error");
            }
        });

        let manager = ChannelManager::new(
            mux_handle,
            super::manager::DEFAULT_MAX_CHANNELS,
            super::reassembly::DEFAULT_BUFFER_CAP,
            remote_addr,
            ChannelSide::Connect,
        );

        let (channel0_send, channel0_recv) = manager
            .install_channel_zero(None)
            .await
            .map_err(|_| StreamError::StreamClosed)?;

        // Single-stream call mode (ADR-036 amendment): channel 0's
        // read+write halves are split into a `SharedFrameWriter`
        // (for `call.requested`) and a read pump (for responses).
        let channel0_source =
            super::source::channel_source(channel0_recv, channel0_send, remote_addr);
        let channel0_conn = Connection::from_source(channel0_source, b"alk/call".to_vec());
        // The caller-identity seam (CF-005): the serving loop's dispatch
        // resolves the peer identity from this connection
        // (`dispatch_start` → `connection.identity()`), so it must carry
        // the identity the serving ops should authenticate by. The
        // explicit `ServingConfig.identity` wins; otherwise the
        // transport connection's identity propagates (the
        // mTLS/QUIC key-based path — the peer the transport already
        // authenticated). `set_identity` is once-only; silently
        // skip if somehow already set.
        let identity_for_serving = serving
            .as_ref()
            .and_then(|config| config.identity.clone())
            .or_else(|| connection.identity().cloned());
        if let Some(identity) = identity_for_serving {
            let _ = channel0_conn.set_identity(identity);
        }
        let channel0_bidi = channel0_conn.accept_bi().await?;
        let (single_stream_writer, single_stream_reader) =
            crate::protocol::connection::split_single_stream(channel0_bidi);

        let call_connection = Arc::new(CallConnection::new_single_stream(
            channel0_conn,
            Arc::clone(&single_stream_writer),
        ));

        let pending_map = Arc::clone(call_connection.pending());
        match serving {
            None => {
                let _read_pump = tokio::spawn(async move {
                    crate::protocol::connection::read_single_stream_until_closed(
                        single_stream_reader,
                        &pending_map,
                    )
                    .await;
                });
            }
            Some(config) => {
                let dispatcher = crate::protocol::dispatch::Dispatcher::new(
                    config.registry,
                    config.identity_provider,
                );
                let call_conn_for_loop = Arc::clone(&call_connection);
                let _serve_loop = tokio::spawn(async move {
                    dispatcher
                        .serve_single_stream(
                            call_conn_for_loop,
                            single_stream_reader,
                            single_stream_writer,
                        )
                        .await;
                });
            }
        }

        // The demux loop ends on transport EOF, clearing the channel
        // map (REQ-CH-02). The connect side passes `policy: None` (it
        // does not enforce the per-identity cap).
        let demux_manager = manager.clone();
        let _demux_task = tokio::spawn(async move {
            super::adapter::ChannelsAdapter::run_demux_loop_for_client(
                &demux_manager,
                Box::new(reader),
                None,
            )
            .await;
        });

        Ok(Self {
            manager,
            call_connection: Arc::new(Mutex::new(Some(call_connection))),
        })
    }

    /// Construct from an established `Connection` with serving
    /// disabled (the pure-consumer default). See
    /// [`ChannelClient::from_connection_with_serving`].
    pub async fn from_connection(connection: Connection) -> Result<Self, StreamError> {
        Self::from_connection_with_serving(connection, None).await
    }

    /// The `ChannelManager` — for relay logic and tests.
    pub fn manager(&self) -> &ChannelManager {
        &self.manager
    }

    /// Call a per-ALPN open op (`channels/<alpn>/sub` or
    /// `channels/<alpn>/pub`) on channel 0. Returns the
    /// `ResponseEnvelope` (which carries `channel_id` on success).
    ///
    /// In single-stream call mode (ADR-036 amendment), this writes
    /// `call.requested` through channel 0's shared frame writer and
    /// awaits the response via the `PendingRequestMap` (resolved by
    /// the read pump from the demux).
    pub async fn call_open_op(&self, operation_id: &str, input: Value) -> ResponseEnvelope {
        let guard = self.call_connection.lock().await;
        match guard.as_ref() {
            Some(conn) => conn.call(operation_id, input).await,
            None => ResponseEnvelope::error(
                "channel-client",
                crate::protocol::wire::CallError::internal("channel client closed"),
            ),
        }
    }

    /// Open a data channel by calling the per-ALPN open op on channel 0
    /// and adopting the resulting `channel_id` (ADR-047 §5 odd/even
    /// split). The connect side calls the open op; the accept side
    /// allocates the `channel_id` (even). The connect side then adopts
    /// the `channel_id` via [`ChannelManager::adopt_channel`] to install
    /// local routing state (mux write half + demux read half).
    ///
    /// Returns the `channel_id`, the `MpscSendStream` (write half), and
    /// the `MpscRecvStream` (read half). The caller can build a
    /// `Connection` from these via `channel_source` and
    /// `Connection::from_source`.
    ///
    /// `alpn` is the data-plane ALPN (e.g. `alk/tty`), used for
    /// observability in the local manager.
    ///
    /// The error is typed (ADR-049 §4 — review 006 N-1): a failed open
    /// resolves [`ChannelOpenError::CallFailed`] carrying the
    /// wire `CallError` verbatim — branch on
    /// [`ChannelOpenError::establishment_reason`] for
    /// `channel:open_failed`'s reason code (`dial_failed`,
    /// `unknown_resource`, `resource_shortage`, `handler_error`,
    /// `timeout`).
    pub async fn open_channel(
        &self,
        operation_id: &str,
        input: Value,
        alpn: &str,
    ) -> Result<(u32, MpscSendStream, MpscRecvStream), ChannelOpenError> {
        let (channel_id, _reply, send, recv) = self
            .open_channel_with_reply(operation_id, input, alpn)
            .await?;
        Ok((channel_id, send, recv))
    }

    /// Open a channel and return the open-op success reply's extra
    /// fields alongside the streams (ADR-049 amendment 3). The reply
    /// is the full success output — `channel_id` plus any
    /// establisher-contributed fields (e.g. a bind-first listener's
    /// `bound`); a producer that contributed none returns just
    /// `{ channel_id }`, so the payload's `Value` is the old reply.
    /// [`ChannelClient::open_channel`] delegates here and discards the
    /// reply, keeping its signature unchanged.
    pub async fn open_channel_with_reply(
        &self,
        operation_id: &str,
        input: Value,
        alpn: &str,
    ) -> Result<(u32, Value, MpscSendStream, MpscRecvStream), ChannelOpenError> {
        let response = self.call_open_op(operation_id, input).await;
        let out = response
            .result
            .map_err(|error| ChannelOpenError::CallFailed { error })?;
        let channel_id = out
            .get("channel_id")
            .and_then(|v| v.as_u64())
            .ok_or(ChannelOpenError::MissingChannelId)? as u32;

        self.manager
            .adopt_channel(channel_id, alpn, None)
            .await
            .map_err(ChannelOpenError::AdoptFailed)
            .map(|(send, recv)| (channel_id, out, send, recv))
    }

    /// Take the `CallConnection` — used by the consumer to register
    /// imported ops (`from_call`) on the connection's overlay. After
    /// this, `call_open_op` returns an error (the connection is owned
    /// by the consumer). The `Arc` is shared with the serving loop when
    /// serving is enabled; taking it detaches the client's own calling
    /// surface, not the serving loop's dispatch (the loop holds its own
    /// `Arc` clone).
    pub async fn take_call_connection(&self) -> Option<Arc<CallConnection>> {
        self.call_connection.lock().await.take()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::channels::adapter::ChannelsAdapter;
    use crate::channels::policy::ChannelLifecyclePolicy;
    use crate::channels::policy::NoCap;
    use crate::core::auth::{AuthContext, IdentityProvider};
    use crate::core::types::Connection;
    use crate::protocol::connection::split_single_stream;
    use crate::protocol::dispatch::Dispatcher;
    use crate::registry::context::OperationContext;
    use crate::registry::registration::{
        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
    };
    use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
    use std::sync::Arc;
    use tokio::io::AsyncWriteExt;

    const TEST_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321);

    fn external_query_spec(name: &str) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
    }

    struct NoopIdProvider;
    impl IdentityProvider for NoopIdProvider {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<crate::core::auth::Identity> {
            None
        }
        fn resolve_from_token(
            &self,
            _: &crate::core::auth::AuthToken,
        ) -> Option<crate::core::auth::Identity> {
            None
        }
    }

    /// Read one length-prefixed `EventEnvelope` frame off `reader` and
    /// parse it into a `ResponseEnvelope`. Used by the single-stream
    /// direct-drive tests to read the responder's single
    /// `call.responded`/`call.error` frame for a Pub.
    async fn read_single_stream_response(
        reader: &mut (impl tokio::io::AsyncRead + Unpin),
    ) -> ResponseEnvelope {
        use tokio::io::AsyncReadExt;
        let mut len_buf = [0u8; 4];
        reader
            .read_exact(&mut len_buf)
            .await
            .expect("read response length");
        let len = u32::from_be_bytes(len_buf) as usize;
        let mut body = vec![0u8; len];
        reader
            .read_exact(&mut body)
            .await
            .expect("read response body");
        let env: crate::protocol::wire::EventEnvelope =
            serde_json::from_slice(&body).expect("parse response envelope");
        let request_id = env.id.clone();
        match env.r#type.as_str() {
            "call.responded" => ResponseEnvelope::ok(
                request_id,
                env.payload
                    .get("output")
                    .cloned()
                    .unwrap_or(serde_json::Value::Null),
            ),
            "call.error" => {
                let err: crate::protocol::wire::CallError = serde_json::from_value(env.payload)
                    .unwrap_or_else(|_| {
                        crate::protocol::wire::CallError::internal("malformed error payload")
                    });
                ResponseEnvelope::error(request_id, err)
            }
            other => panic!("expected call.responded or call.error, got {other}"),
        }
    }

    /// Build the `install_channel_zero` hook for the accept side: it
    /// yields channel 0's `BiStream` once, splits it into the shared
    /// writer + reader, constructs a single-stream `CallConnection`,
    /// and runs `Dispatcher::run_loop_single_stream` on it. Returns
    /// the `JoinHandle` for the spawned dispatch loop.
    fn make_install_channel_zero(
        registry: Arc<crate::registry::registration::OperationRegistry>,
    ) -> crate::channels::adapter::InstallChannelZero {
        Arc::new(move |_manager, channel0_conn, _auth| {
            let registry = Arc::clone(&registry);
            let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
            tokio::spawn(async move {
                let channel0_bidi = match channel0_conn.accept_bi().await {
                    Ok(s) => s,
                    Err(_) => return,
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));
                let dp = Dispatcher::new(registry, provider);
                dp.run_loop_single_stream(call_connection, reader, writer)
                    .await;
            })
        })
    }

    /// C-25 #1 — the end-to-end acceptance gate for Unit 2 (channel 0
    /// single-stream call mode, ADR-036 amendment). Wires
    /// `ChannelClient` (connect side) ↔ `ChannelsAdapter` (accept
    /// side) over a real `tokio::io::duplex` pair carrying the
    /// channels 8-byte chunk header wire format. A `call.requested`
    /// from the client is demuxed on channel 0, dispatched by the
    /// server's single-stream `Dispatcher::run_loop_single_stream`,
    /// and the `call.responded` is muxed back to the client's read
    /// pump, resolving the pending call. This would have caught C-01
    /// immediately (every `call_open_op` failed with
    /// `StreamClosed` on the connect side; channel 0 was a black hole
    /// on the accept side).
    #[tokio::test]
    async fn channel_0_end_to_end_call_round_trip() {
        let registry = crate::registry::registration::OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_query_spec("echo/run"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(
            make_install_channel_zero(Arc::clone(&registry)),
            Arc::new(NoCap),
        );
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op("echo/run", serde_json::json!({ "msg": "hi" })),
        )
        .await
        .expect("channel-0 round-trip timed out");

        assert!(
            response.result.is_ok(),
            "channel-0 round-trip should succeed, got {:?}",
            response.result
        );
        assert_eq!(response.result.unwrap(), serde_json::json!({ "msg": "hi" }));
    }

    /// C-25 #1 (negative case) — an unknown op over channel 0 returns
    /// `NOT_FOUND`, proving the single-stream dispatch path reaches
    /// the registry's not-found branch end-to-end (not just
    /// `StreamClosed` from a dead channel 0 as in C-01).
    #[tokio::test]
    async fn channel_0_end_to_end_unknown_op_returns_not_found() {
        let registry = Arc::new(crate::registry::registration::OperationRegistry::new());

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(
            make_install_channel_zero(Arc::clone(&registry)),
            Arc::new(NoCap),
        );
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op("no/such/op", serde_json::json!({})),
        )
        .await
        .expect("channel-0 unknown-op timed out");

        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND", "unknown op over channel 0"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    /// C-25 #1 (Pub over channel 0) — a `publish()` through channel
    /// 0's single-stream mode delivers chunks to the responder's
    /// `SinkHandler` and returns the handler's response. Proves the
    /// single-stream dispatch loop's in-flight sink routing works
    /// (`call.published` frames arriving after `call.requested` are
    /// routed to the matching sink's `chunk_tx`).
    #[tokio::test]
    async fn channel_0_end_to_end_publish_delivers_chunks() {
        use futures::stream::StreamExt;

        let registry = crate::registry::registration::OperationRegistry::new();
        let counting_sink = crate::registry::registration::make_sink_handler(
            |_input, ctx, mut stream| async move {
                let mut count = 0u32;
                let mut last = serde_json::Value::Null;
                while let Some(item) = stream.next().await {
                    match item {
                        Ok(v) => {
                            count += 1;
                            last = v;
                        }
                        Err(_) => break,
                    }
                }
                ResponseEnvelope::ok(
                    ctx.request_id,
                    serde_json::json!({ "count": count, "last": last }),
                )
            },
        );
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "fs/upload",
                    OperationType::Pub,
                    Visibility::External,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(counting_sink),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(
            make_install_channel_zero(Arc::clone(&registry)),
            Arc::new(NoCap),
        );
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let call_conn = client
            .take_call_connection()
            .await
            .expect("call connection present");

        let chunks = vec![
            serde_json::json!({"chunk": 1}),
            serde_json::json!({"chunk": 2}),
            serde_json::json!({"chunk": 3}),
        ];
        let stream: std::pin::Pin<
            Box<dyn futures::stream::Stream<Item = serde_json::Value> + Send>,
        > = Box::pin(futures::stream::iter(chunks.clone()));
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            call_conn.publish("fs/upload", serde_json::json!({"path": "/x"}), stream),
        )
        .await
        .expect("channel-0 publish timed out");

        assert!(
            response.result.is_ok(),
            "publish should succeed, got {:?}",
            response.result
        );
        let out = response.result.unwrap();
        assert_eq!(
            out["count"],
            serde_json::json!(3),
            "responder saw all 3 chunks"
        );
        assert_eq!(
            out["last"],
            serde_json::json!({"chunk": 3}),
            "last chunk matches"
        );
    }

    /// Unit 8 / P-03 (single-stream path) — `publish_schema` validation
    /// fires on the single-stream dispatch loop: a chunk that violates
    /// the schema is injected as an `Err(INVALID_INPUT)` into the
    /// `SinkHandler`'s stream, terminating it. The handler observes the
    /// valid chunks before the violation plus the error, and its
    /// response reflects that. This proves the `EVENT_PUBLISHED` arm of
    /// `run_loop_single_stream` validates against the
    /// `InFlightSink.publish_validator` before yielding `Ok(chunk)`.
    #[tokio::test]
    async fn channel_0_publish_schema_rejects_invalid_chunk_on_single_stream() {
        use futures::stream::StreamExt;

        let publish_schema = serde_json::json!({
            "type": "object",
            "properties": { "bytes": { "type": "string" } },
            "required": ["bytes"]
        });
        let recording_sink = crate::registry::registration::make_sink_handler(
            |_input, ctx, mut stream| async move {
                let mut oks: Vec<serde_json::Value> = Vec::new();
                let mut err: Option<crate::protocol::wire::CallError> = None;
                while let Some(item) = stream.next().await {
                    match item {
                        Ok(v) => oks.push(v),
                        Err(e) => {
                            err = Some(e);
                            break;
                        }
                    }
                }
                ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "ok": oks, "err": err }))
            },
        );
        let registry = crate::registry::registration::OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "fs/upload",
                    OperationType::Pub,
                    Visibility::External,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                )
                .with_publish_schema(publish_schema),
                HandlerKind::Sink(recording_sink),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(
            make_install_channel_zero(Arc::clone(&registry)),
            Arc::new(NoCap),
        );
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");
        let call_conn = client
            .take_call_connection()
            .await
            .expect("call connection present");

        let chunks = vec![
            serde_json::json!({"bytes": "hello"}),
            serde_json::json!({"bytes": 42}),
        ];
        let stream: std::pin::Pin<
            Box<dyn futures::stream::Stream<Item = serde_json::Value> + Send>,
        > = Box::pin(futures::stream::iter(chunks));
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            call_conn.publish("fs/upload", serde_json::json!({}), stream),
        )
        .await
        .expect("channel-0 publish timed out");

        let out = response.result.expect("publish ok");
        let oks = out.get("ok").and_then(|v| v.as_array()).expect("ok array");
        assert_eq!(oks.len(), 1, "only the valid chunk reached the handler");
        assert_eq!(oks[0], serde_json::json!({"bytes": "hello"}));
        let err = out.get("err").expect("err field");
        assert_eq!(
            err.get("code").and_then(|c| c.as_str()),
            Some("INVALID_INPUT"),
            "invalid chunk injects INVALID_INPUT on the single-stream path"
        );
    }

    /// Unit 8 / P-04 (single-stream path) — an initiator-side
    /// `call.error` during a publish over channel 0 is routed to the
    /// matching in-flight sink's `chunk_tx` as `Err(call_error)`,
    /// terminating the stream. The handler observes the chunks before
    /// the error plus the initiator's `CallError` (not a synthetic
    /// "aborted"). This drives `run_loop_single_stream` directly with
    /// crafted frames (a `call.requested`, one `call.published`, then a
    /// `call.error`) because the public `publish()` API takes
    /// `Stream<Item = Value>` and cannot emit an initiator `call.error`
    /// — that path is for a future `Stream<Item = Result<Value,
    /// CallError>>` publish API. The `call.error` is injected manually
    /// on the wire to prove the single-stream `EVENT_ERROR` arm parses
    /// the payload and routes it to the right in-flight sink.
    #[tokio::test]
    async fn channel_0_initiator_call_error_terminates_publish_on_single_stream() {
        use crate::protocol::connection::split_single_stream;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::wire::EventEnvelope;
        use futures::stream::StreamExt;

        let recording_sink = crate::registry::registration::make_sink_handler(
            |_input, ctx, mut stream| async move {
                let mut oks: Vec<serde_json::Value> = Vec::new();
                let mut err: Option<crate::protocol::wire::CallError> = None;
                while let Some(item) = stream.next().await {
                    match item {
                        Ok(v) => oks.push(v),
                        Err(e) => {
                            err = Some(e);
                            break;
                        }
                    }
                }
                ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "ok": oks, "err": err }))
            },
        );
        let registry = crate::registry::registration::OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "fs/upload",
                    OperationType::Pub,
                    Visibility::External,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(recording_sink),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let request = EventEnvelope::requested(
            "pub-ss-err-1",
            serde_json::json!({
                "operationId": "fs/upload",
                "input": {},
            }),
        );
        let published = EventEnvelope::published("pub-ss-err-1", serde_json::json!({"chunk": 1}));
        let initiator_error = EventEnvelope::error(
            "pub-ss-err-1",
            &crate::protocol::wire::CallError::new("PUBLISH_FAILED", "disk full", false),
        );

        let mut frame_buf = Vec::new();
        for env in [&request, &published, &initiator_error] {
            let body = serde_json::to_vec(env).expect("serialize envelope");
            frame_buf.extend_from_slice(&(body.len() as u32).to_be_bytes());
            frame_buf.extend_from_slice(&body);
        }

        let (response_send, mut response_recv) = tokio::io::duplex(64 * 1024);
        let bidi = crate::core::types::BiStream::from_joined(
            tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)),
            response_send,
        );
        let (writer, reader) = split_single_stream(bidi);

        let call_connection =
            Arc::new(CallConnection::new(crate::protocol::sink_empty_connection()));
        let dp = Dispatcher::new(registry, Arc::new(NoopIdProvider));
        let _server_handle = tokio::spawn(async move {
            dp.run_loop_single_stream(call_connection, reader, writer)
                .await;
        });

        let response = read_single_stream_response(&mut response_recv).await;
        let out = response.result.expect("handler ok");
        let oks = out.get("ok").and_then(|v| v.as_array()).expect("ok array");
        assert_eq!(
            oks.len(),
            1,
            "the chunk before the error reached the handler"
        );
        assert_eq!(oks[0], serde_json::json!({"chunk": 1}));
        let err = out.get("err").expect("err field");
        assert_eq!(
            err.get("code").and_then(|c| c.as_str()),
            Some("PUBLISH_FAILED"),
            "the initiator's CallError code reaches the handler on the single-stream path"
        );
    }

    /// C-02 / C-03 — the end-to-end acceptance gate for Unit 3
    /// (`register_openable`, ADR-047 §3 as amended 2026-08-13 —
    /// per-connection registration). Wires `ChannelClient` (connect
    /// side) ↔ `ChannelsAdapter` (accept side) over a real
    /// `tokio::io::duplex` carrying the channels 8-byte chunk header
    /// wire format. The accept side's `install_channel_zero` hook
    /// builds a per-connection `ChannelCore` from the adapter-supplied
    /// `ChannelManager`, registers a no-op open op
    /// (`channels/tty/sub`) via `register_openable` on a fresh
    /// per-connection registry, and runs the single-stream dispatch
    /// loop on it. The client calls `call_open_op("channels/tty/sub")`
    /// on channel 0; the wrapper does `check_open` → `open_channel` →
    /// spawn the no-op `OpenHandler` on the channel's `Connection` →
    /// respond `{ channel_id }`. Asserts the response carries a
    /// `channel_id` and that the per-identity quota was reserved
    /// (the policy's count for the caller incremented).
    ///
    /// This is the gate the review identifies as missing: "register a
    /// no-op open op via `register_openable`, invoke it end-to-end
    /// through channel 0, assert `channel_id` is returned and quota is
    /// reserved." It would have caught C-02 (no way to register an
    /// open op) and C-03 (`resolve_channel_manager` stub) immediately.
    #[tokio::test]
    async fn channel_0_end_to_end_register_openable_returns_channel_id() {
        use crate::channels::operations::{ChannelCore, OpenHandler};
        use crate::channels::policy::PerIdentityChannelPolicy;
        use crate::registry::spec::ChannelOpenSpec;

        let policy: Arc<PerIdentityChannelPolicy> = Arc::new(PerIdentityChannelPolicy::new(256));
        let policy_for_hook: Arc<dyn ChannelLifecyclePolicy> =
            Arc::clone(&policy) as Arc<dyn ChannelLifecyclePolicy>;
        let policy_for_assert = Arc::clone(&policy);

        let open_handler: OpenHandler = Arc::new(|_input, _plan, _channel_conn, _auth| {
            tokio::spawn(async move {
                // No-op: the channel's BiStream is available via
                // `_channel_conn.accept_bi()` if the test wanted
                // to move data on it. For the open-succeeds gate,
                // just keep the task alive briefly so the handler
                // task is real (and recorded for teardown).
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            })
        });

        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let open_handler = Arc::clone(&open_handler);
                let policy = Arc::clone(&policy_for_hook);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);

                    let core = ChannelCore::new(manager, policy);
                    let registry = crate::registry::registration::OperationRegistry::new();
                    let spec = OperationSpec::new(
                        "channels/tty/sub",
                        OperationType::Sub,
                        Visibility::External,
                        serde_json::json!({
                            "type": "object",
                            "properties": {
                                "container": { "type": "string" }
                            },
                            "required": ["container"]
                        }),
                        serde_json::json!({
                            "type": "object",
                            "properties": {
                                "channel_id": { "type": "integer" }
                            }
                        }),
                        vec![],
                        AccessControl::default(),
                        None,
                    )
                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
                    core.register_openable(
                        spec,
                        Arc::clone(&open_handler),
                        &registry,
                        auth.clone(),
                    )
                    .expect("register_openable");

                    let registry = Arc::new(registry);
                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(registry, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
            ),
        )
        .await
        .expect("channel-0 open op timed out");

        assert!(
            response.result.is_ok(),
            "open op should succeed, got {:?}",
            response.result
        );
        let out = response.result.unwrap();
        let channel_id = out
            .get("channel_id")
            .and_then(|v| v.as_u64())
            .expect("channel_id in response");
        assert!(
            channel_id > 0,
            "channel_id should be non-zero (channel 0 is the call channel), got {channel_id}"
        );

        let anonymous = crate::core::auth::Identity {
            id: "anonymous".to_string(),
            scopes: vec![],
            resources: Default::default(),
        };
        let count = policy_for_assert.count_for(&anonymous);
        assert_eq!(
            count, 1,
            "quota reserved: the open op incremented the per-identity count"
        );
    }

    /// C-25 #4 — policy decrement on connection drop (ADR-047 §7).
    /// Open channels via the manager, then call `clear_all` through the
    /// demux loop with a policy — the per-identity count must be
    /// decremented for each drained channel.
    #[tokio::test]
    async fn policy_decremented_on_connection_drop() {
        use crate::channels::policy::PerIdentityChannelPolicy;

        let policy: Arc<PerIdentityChannelPolicy> = Arc::new(PerIdentityChannelPolicy::new(2));
        let dyn_policy: Arc<dyn ChannelLifecyclePolicy> =
            Arc::clone(&policy) as Arc<dyn ChannelLifecyclePolicy>;

        let (client, server) = tokio::io::duplex(64 * 1024);
        let (server_read, server_write) = tokio::io::split(server);
        let (mux_handle, mux_runner) = MuxRunner::new(Box::new(server_write));
        let _mux_task = tokio::spawn(async move {
            let _ = mux_runner.run().await;
        });
        let manager = ChannelManager::with_defaults(mux_handle, None);

        let alice = crate::core::auth::Identity {
            id: "alice".to_string(),
            scopes: vec![],
            resources: Default::default(),
        };
        let bob = crate::core::auth::Identity {
            id: "bob".to_string(),
            scopes: vec![],
            resources: Default::default(),
        };

        assert!(dyn_policy.check_open(&alice).is_ok());
        manager
            .open_channel("alk/tty", "alice", None)
            .await
            .expect("open alice");
        assert!(dyn_policy.check_open(&bob).is_ok());
        manager
            .open_channel("alk/tty", "bob", None)
            .await
            .expect("open bob");

        assert_eq!(policy.count_for(&alice), 1);
        assert_eq!(policy.count_for(&bob), 1);

        let demux_manager = manager.clone();
        let demux_policy = Arc::clone(&dyn_policy);
        let demux_task = tokio::spawn(async move {
            ChannelsAdapter::run_demux_loop_for_client(
                &demux_manager,
                Box::new(server_read),
                Some(&demux_policy),
            )
            .await;
        });

        drop(client);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), demux_task).await;

        assert_eq!(
            policy.count_for(&alice),
            0,
            "alice decremented after connection drop"
        );
        assert_eq!(
            policy.count_for(&bob),
            0,
            "bob decremented after connection drop"
        );
    }

    /// C-25 #5 — concurrent open race (TOCTOU). Multiple concurrent
    /// `open_channel` calls on a manager with `max_channels=1` must not
    /// all succeed — only one should pass the re-check on insert.
    #[tokio::test]
    async fn concurrent_opens_respect_max_channels() {
        let (_client, server) = tokio::io::duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = Arc::new(ChannelManager::new(
            handle,
            1,
            64,
            None,
            ChannelSide::Accept,
        ));

        let m1 = Arc::clone(&manager);
        let m2 = Arc::clone(&manager);
        let m3 = Arc::clone(&manager);

        let (r1, r2, r3) = tokio::join!(
            m1.open_channel("alk/a", "alice", None),
            m2.open_channel("alk/b", "bob", None),
            m3.open_channel("alk/c", "carol", None),
        );

        let successes = [r1.is_ok(), r2.is_ok(), r3.is_ok()]
            .iter()
            .filter(|&&ok| ok)
            .count();
        assert_eq!(
            successes, 1,
            "exactly one concurrent open should succeed with max_channels=1"
        );
        assert_eq!(
            manager.open_count(),
            1,
            "channel count should be 1 after concurrent opens"
        );
    }

    /// C-12 — `channel/close` with `channel_id: 0` is rejected.
    #[tokio::test]
    async fn channel_close_rejects_channel_zero() {
        use crate::channels::operations::ChannelOperations;
        use crate::channels::policy::NoCap;
        use crate::registry::context::{AbortPolicy, ScopedPeerEnv};
        use std::collections::HashMap;
        use std::sync::Arc;

        struct NoopEnv;
        #[async_trait::async_trait]
        impl crate::registry::env::OperationEnv for NoopEnv {
            async fn invoke_with_policy(
                &self,
                _namespace: &str,
                _operation: &str,
                _input: serde_json::Value,
                _parent: &OperationContext,
                _policy: AbortPolicy,
            ) -> ResponseEnvelope {
                ResponseEnvelope::error("test", crate::protocol::wire::CallError::internal("noop"))
            }
            fn contains(&self, _name: &str) -> bool {
                false
            }
        }

        let (_client, server) = tokio::io::duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);
        let ops = ChannelOperations::new(manager, Arc::new(NoCap));
        let registry = crate::registry::registration::OperationRegistry::new();
        ops.register_on(&registry).expect("register");

        let handler = registry
            .registration("channel/close")
            .expect("close op registered")
            .handler
            .clone();
        let ctx = OperationContext {
            request_id: "req-1".to_string(),
            parent_request_id: None,
            identity: None,
            handler_identity: None,
            forwarded_for: None,
            capabilities: crate::core::types::Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env: ScopedPeerEnv::empty(),
            env: Arc::new(NoopEnv),
            abort_policy: AbortPolicy::default(),
            deadline: Some(std::time::Instant::now() + std::time::Duration::from_secs(30)),
            internal: false,
            ownership: None,
        };
        let input = serde_json::json!({ "channel_id": 0 });

        let response = match handler {
            HandlerKind::Once(h) => h(input, ctx).await,
            _ => panic!("expected Once handler"),
        };

        assert!(
            response.result.is_err(),
            "channel/close with channel_id:0 should be rejected"
        );
        let err = response.result.unwrap_err();
        assert!(
            err.code == "INVALID_INPUT",
            "expected INVALID_INPUT, got {}",
            err.code
        );
    }

    /// C-08 — end-to-end channel adoption: the connect side calls an
    /// open op, receives a `channel_id`, adopts it, and can write data
    /// through the adopted channel to the accept side's handler.
    #[tokio::test]
    async fn channel_adoption_end_to_end_round_trip() {
        use crate::channels::operations::{ChannelCore, OpenHandler};
        use crate::channels::policy::NoCap;
        use crate::registry::spec::ChannelOpenSpec;
        use tokio::io::AsyncReadExt;

        let (data_tx, mut data_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);

        let open_handler: OpenHandler = Arc::new(move |_input, _plan, channel_conn, _auth| {
            let data_tx = data_tx.clone();
            tokio::spawn(async move {
                let mut bidi = channel_conn.accept_bi().await.expect("accept_bi");
                let mut buf = [0u8; 4];
                bidi.read_exact(&mut buf).await.expect("read");
                data_tx.send(buf.to_vec()).await.expect("send to channel");
            })
        });

        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let open_handler = Arc::clone(&open_handler);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let core = ChannelCore::new(manager, Arc::new(NoCap));
                    let registry = crate::registry::registration::OperationRegistry::new();
                    let spec = OperationSpec::new(
                        "channels/tty/sub",
                        OperationType::Sub,
                        Visibility::External,
                        serde_json::json!({
                            "type": "object",
                            "properties": { "container": { "type": "string" } },
                            "required": ["container"]
                        }),
                        serde_json::json!({
                            "type": "object",
                            "properties": { "channel_id": { "type": "integer" } }
                        }),
                        vec![],
                        AccessControl::default(),
                        None,
                    )
                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
                    core.register_openable(
                        spec,
                        Arc::clone(&open_handler),
                        &registry,
                        auth.clone(),
                    )
                    .expect("register_openable");
                    let registry = Arc::new(registry);
                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(registry, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let (channel_id, mut send, _recv) = client
            .open_channel(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
                "alk/tty",
            )
            .await
            .expect("open_channel");

        assert!(channel_id > 0, "channel_id should be non-zero");
        assert!(channel_id % 2 == 0, "accept-allocated ID should be even");

        send.write_all(b"ping").await.expect("write ping");
        drop(send);

        let data = tokio::time::timeout(std::time::Duration::from_secs(5), data_rx.recv())
            .await
            .expect("timed out waiting for handler data")
            .expect("handler should receive data");
        assert_eq!(
            &data, b"ping",
            "handler received ping through adopted channel"
        );
    }

    /// ADR-049 Unit 1 acceptance gate (review 006 E-01/N-1) — the
    /// establisher fails after channel allocation: the consumer's
    /// `open_channel` resolves `Err` carrying the wire `CallError`
    /// (`channel:open_failed` + `details.reason`), no `channel_id` was
    /// ever returned (the SSH "channel never exists opener-side"
    /// property, consumer-visible), and the accept side's manager has
    /// no channel afterward (teardown + ledger un-increment).
    #[tokio::test]
    async fn establisher_failure_resolves_typed_open_failed_on_the_client() {
        use crate::channels::operations::{
            ChannelCore, EstablishmentError, OpenEstablisher, OpenHandler,
        };
        use crate::channels::policy::PerIdentityChannelPolicy;
        use crate::registry::spec::ChannelOpenSpec;

        let policy: Arc<PerIdentityChannelPolicy> = Arc::new(PerIdentityChannelPolicy::new(256));
        let policy_for_hook: Arc<dyn ChannelLifecyclePolicy> =
            Arc::clone(&policy) as Arc<dyn ChannelLifecyclePolicy>;
        let policy_for_assert = Arc::clone(&policy);

        let open_handler: OpenHandler =
            Arc::new(|_input, _plan, _channel_conn, _auth| tokio::spawn(async {}));
        let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
            Box::pin(async {
                Err(EstablishmentError::DialFailed {
                    message: "target refused the connection".to_string(),
                })
            })
        });

        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let open_handler = Arc::clone(&open_handler);
                let establisher = Arc::clone(&establisher);
                let policy = Arc::clone(&policy_for_hook);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let core = ChannelCore::new(manager, policy);
                    let registry = crate::registry::registration::OperationRegistry::new();
                    let spec = OperationSpec::new(
                        "channels/tty/sub",
                        OperationType::Sub,
                        Visibility::External,
                        serde_json::json!({}),
                        serde_json::json!({
                            "type": "object",
                            "properties": { "channel_id": { "type": "integer" } }
                        }),
                        vec![],
                        AccessControl::default(),
                        None,
                    )
                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
                    core.register_openable_with_establisher(
                        spec,
                        Some(establisher),
                        open_handler,
                        &registry,
                        auth.clone(),
                        None,
                    )
                    .expect("register_openable_with_establisher");
                    let registry = Arc::new(registry);
                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(registry, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let err = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.open_channel(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
                "alk/tty",
            ),
        )
        .await
        .expect("open_channel timed out");
        let err = match err {
            Err(e) => e,
            Ok(_) => panic!("establisher failure must resolve Err through the client"),
        };

        // N-1 gate: the typed error carries the wire CallError verbatim.
        let call_error = err.call_error().expect("CallFailed carries the CallError");
        assert_eq!(call_error.code, "channel:open_failed");
        assert_eq!(
            err.establishment_reason(),
            Some("dial_failed"),
            "consumer branches on the typed reason code"
        );

        // The SSH property, consumer-visible: no channel_id was ever
        // returned, and the accept side holds no channel for the failed
        // open (channel 0 is the pre-negotiated call channel — always
        // present, never the failed open's).
        assert!(
            client.manager().channel_ids().into_iter().all(|id| id == 0),
            "no data channel exists for the failed open"
        );
        let anonymous = crate::core::auth::Identity {
            id: "anonymous".to_string(),
            scopes: vec![],
            resources: Default::default(),
        };
        assert_eq!(
            policy_for_assert.count_for(&anonymous),
            0,
            "ledger un-incremented on the accept side after establishment failure"
        );
    }

    /// ADR-049 Unit 1 acceptance gate (companion to the E-01 gate) —
    /// establisher success through a real channels connection: the
    /// client's `open_channel` adopts and the pumps flow (the happy
    /// path is unchanged by the establishment phase).
    #[tokio::test]
    async fn establisher_success_opens_and_pumps_data_end_to_end() {
        use crate::channels::operations::{
            ChannelCore, Establishment, OpenEstablisher, OpenHandler,
        };
        use crate::channels::policy::NoCap;
        use crate::registry::spec::ChannelOpenSpec;
        use tokio::io::AsyncReadExt;

        let (data_tx, mut data_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);

        let open_handler: OpenHandler = Arc::new(move |_input, _plan, channel_conn, _auth| {
            let data_tx = data_tx.clone();
            tokio::spawn(async move {
                let mut bidi = channel_conn.accept_bi().await.expect("accept_bi");
                let mut buf = [0u8; 4];
                bidi.read_exact(&mut buf).await.expect("read");
                data_tx.send(buf.to_vec()).await.expect("send to channel");
            })
        });
        let establisher: OpenEstablisher =
            Arc::new(|_input, _auth| Box::pin(async { Ok(Establishment::default()) }));

        let establisher_for_hook = Arc::clone(&establisher);
        let open_handler_for_hook = Arc::clone(&open_handler);
        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let establisher = Arc::clone(&establisher_for_hook);
                let open_handler = Arc::clone(&open_handler_for_hook);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let core = ChannelCore::new(manager, Arc::new(NoCap));
                    let registry = crate::registry::registration::OperationRegistry::new();
                    let spec = OperationSpec::new(
                        "channels/tty/sub",
                        OperationType::Sub,
                        Visibility::External,
                        serde_json::json!({
                            "type": "object",
                            "properties": { "container": { "type": "string" } },
                            "required": ["container"]
                        }),
                        serde_json::json!({
                            "type": "object",
                            "properties": { "channel_id": { "type": "integer" } }
                        }),
                        vec![],
                        AccessControl::default(),
                        None,
                    )
                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
                    core.register_openable_with_establisher(
                        spec,
                        Some(establisher),
                        open_handler,
                        &registry,
                        auth.clone(),
                        None,
                    )
                    .expect("register_openable_with_establisher");
                    let registry = Arc::new(registry);
                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(registry, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let (channel_id, mut send, _recv) = client
            .open_channel(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
                "alk/tty",
            )
            .await
            .expect("open_channel with establisher success");

        assert!(channel_id > 0, "channel_id should be non-zero");
        send.write_all(b"ping").await.expect("write ping");
        drop(send);

        let data = tokio::time::timeout(std::time::Duration::from_secs(5), data_rx.recv())
            .await
            .expect("timed out waiting for handler data")
            .expect("handler should receive data");
        assert_eq!(&data, b"ping", "handler received ping post-establishment");
    }

    /// ADR-049 amendment 3 (review 008 U-2) acceptance gate: the
    /// establisher contributes a reply field (`bound`, the bind-first
    /// listener shape) and the client's `open_channel_with_reply`
    /// receives it end-to-end over a real channels connection, while
    /// `open_channel` keeps its pre-amendment shape (fields discarded).
    #[tokio::test]
    async fn establisher_reply_fields_reach_open_channel_with_reply_end_to_end() {
        use crate::channels::operations::{
            ChannelCore, Establishment, OpenEstablisher, OpenHandler,
        };
        use crate::channels::policy::NoCap;
        use crate::registry::spec::ChannelOpenSpec;

        let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
            Box::pin(async {
                Ok(Establishment::default().with_reply_field(
                    "bound",
                    serde_json::json!({ "host": "203.0.113.9", "port": 42113 }),
                ))
            })
        });
        let open_handler: OpenHandler =
            Arc::new(|_input, _plan, _channel_conn, _auth| tokio::spawn(async {}));

        let establisher_for_hook = Arc::clone(&establisher);
        let open_handler_for_hook = Arc::clone(&open_handler);
        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let establisher = Arc::clone(&establisher_for_hook);
                let open_handler = Arc::clone(&open_handler_for_hook);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let core = ChannelCore::new(manager, Arc::new(NoCap));
                    let registry = crate::registry::registration::OperationRegistry::new();
                    let spec = OperationSpec::new(
                        "channels/tty/sub",
                        OperationType::Sub,
                        Visibility::External,
                        serde_json::json!({}),
                        serde_json::json!({
                            "type": "object",
                            "properties": {
                                "channel_id": { "type": "integer" },
                                "bound": { "type": "object" }
                            }
                        }),
                        vec![],
                        AccessControl::default(),
                        None,
                    )
                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
                    core.register_openable_with_establisher(
                        spec,
                        Some(establisher),
                        open_handler,
                        &registry,
                        auth.clone(),
                        None,
                    )
                    .expect("register_openable_with_establisher");
                    let registry = Arc::new(registry);
                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(registry, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let (channel_id, reply, mut send, _recv) = client
            .open_channel_with_reply(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
                "alk/tty",
            )
            .await
            .expect("open_channel_with_reply");

        assert!(channel_id > 0, "channel_id should be non-zero");
        assert_eq!(reply["channel_id"], serde_json::json!(channel_id));
        assert_eq!(
            reply["bound"],
            serde_json::json!({ "host": "203.0.113.9", "port": 42113 }),
            "the establisher's reply field survives the wire to the consumer"
        );

        // The streams work as always — the projection is reply-plane only.
        send.write_all(b"ping").await.expect("write ping");
        drop(send);

        // `open_channel` (fields discarded) still opens cleanly against
        // the same accept side.
        let (channel_id2, _send2, _recv2) = client
            .open_channel("channels/tty/sub", serde_json::json!({}), "alk/tty")
            .await
            .expect("open_channel unchanged");
        assert!(channel_id2 > 0);
    }

    /// Review 009 C-4: the new pub API's failure path — a
    /// `channel:open_failed` resolving through
    /// `open_channel_with_reply` carries the full wire error shape
    /// (`code` + typed `details` with `reason` and `message`), so the
    /// caller can branch on the establishment reason through the new
    /// API the same way `open_channel`'s shared parse path does.
    #[tokio::test]
    async fn open_channel_with_reply_resolves_typed_open_failed_end_to_end() {
        use crate::channels::operations::{
            ChannelCore, EstablishmentError, OpenEstablisher, OpenHandler,
        };
        use crate::channels::policy::NoCap;
        use crate::registry::spec::ChannelOpenSpec;

        let open_handler: OpenHandler =
            Arc::new(|_input, _plan, _channel_conn, _auth| tokio::spawn(async {}));
        let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
            Box::pin(async {
                Err(EstablishmentError::DialFailed {
                    message: "target refused the connection".to_string(),
                })
            })
        });

        let establisher_for_hook = Arc::clone(&establisher);
        let open_handler_for_hook = Arc::clone(&open_handler);
        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let establisher = Arc::clone(&establisher_for_hook);
                let open_handler = Arc::clone(&open_handler_for_hook);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let core = ChannelCore::new(manager, Arc::new(NoCap));
                    let registry = crate::registry::registration::OperationRegistry::new();
                    let spec = OperationSpec::new(
                        "channels/tty/sub",
                        OperationType::Sub,
                        Visibility::External,
                        serde_json::json!({}),
                        serde_json::json!({
                            "type": "object",
                            "properties": { "channel_id": { "type": "integer" } }
                        }),
                        vec![],
                        AccessControl::default(),
                        None,
                    )
                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
                    core.register_openable_with_establisher(
                        spec,
                        Some(establisher),
                        open_handler,
                        &registry,
                        auth.clone(),
                        None,
                    )
                    .expect("register_openable_with_establisher");
                    let registry = Arc::new(registry);
                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(registry, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        let err = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.open_channel_with_reply(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
                "alk/tty",
            ),
        )
        .await
        .expect("open_channel_with_reply timed out");
        let err = match err {
            Err(e) => e,
            Ok(_) => panic!("the establisher failure must resolve Err through the reply API"),
        };

        // The full reply shape is assertable by the caller: the wire
        // code, the reason code, and the message inside `details`.
        let call_error = err.call_error().expect("CallFailed carries the CallError");
        assert_eq!(call_error.code, "channel:open_failed");
        assert_eq!(err.establishment_reason(), Some("dial_failed"));
        let details = call_error
            .details
            .as_ref()
            .expect("details ride the typed error");
        assert_eq!(details["reason"], serde_json::json!("dial_failed"));
        assert_eq!(
            details["message"],
            serde_json::json!("target refused the connection"),
            "the establishment-failure message survives the wire to the caller"
        );
        assert!(
            client.manager().channel_ids().into_iter().all(|id| id == 0),
            "no data channel exists for the failed open"
        );
    }

    // --- review 004 Unit 3 acceptance gates (F-04 serving half) -----------

    /// F-04 gate 1: hub→consumer call over an existing `ChannelClient`
    /// session resolves. The consumer (connect side) dials with
    /// `Some(ServingConfig { .. })`; the accept side's dispatcher later
    /// calls an op the consumer serves (by opening a fresh stream from
    /// its side is not available on channel 0 — the accept side writes
    /// a `call.requested` through its channel-0 writer, and the
    /// consumer's serving loop dispatches it and writes the response,
    /// which resolves on the accept side's pending).
    ///
    /// Mechanically: the accept side's `install_channel_zero` hook
    /// returns a clone of the channel-0 `CallConnection` back to the
    /// test via a channel; the test then calls
    /// `call_open_op("consumer/echo")` — wait, that calls *through* the
    /// accept side's registry. The consumer-serving direction needs the
    /// accept side to *initiate*: it writes `call.requested` for
    /// `consumer/echo` on channel 0. The client's serving loop (with
    /// the `consumer/echo` op in its registry) dispatches it and writes
    /// the response, resolving the accept side's pending.
    #[tokio::test]
    async fn serving_loop_hub_to_consumer_call_resolves() {
        // Consumer-side (connect side) registry: serves `consumer/echo`.
        let consumer_registry = crate::registry::registration::OperationRegistry::new();
        consumer_registry
            .register(HandlerRegistration::new(
                external_query_spec("consumer/echo"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let consumer_registry = Arc::new(consumer_registry);

        // Accept side: a plain echo registry plus a handle back to the
        // accept side's channel-0 `CallConnection` so the test can
        // initiate a call *from* the accept side.
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let accept_registry = crate::registry::registration::OperationRegistry::new();
        accept_registry
            .register(HandlerRegistration::new(
                external_query_spec("accept/echo"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let accept_registry = Arc::new(accept_registry);

        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, _auth| {
                let accept_registry = Arc::clone(&accept_registry);
                let accept_conn_tx = accept_conn_tx.clone();
                tokio::spawn(async move {
                    let _ = manager;
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;
                    let dp = Dispatcher::new(
                        accept_registry,
                        Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
                    );
                    // The serving loop on the accept side too: the
                    // consumer may call back over the same session.
                    dp.serve_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&consumer_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        // Accept side (hub) initiates a call to an op the consumer
        // serves — the exact direction F-04 says was dropped.
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.call("consumer/echo", serde_json::json!({ "from": "hub" })),
        )
        .await
        .expect("hub→consumer call timed out");

        assert!(
            response.result.is_ok(),
            "hub→consumer call resolves through the consumer's serving loop, got {:?}",
            response.result
        );
        assert_eq!(
            response.result.unwrap(),
            serde_json::json!({ "from": "hub" })
        );

        // The consumer's own calling half still works (outbound pendings
        // resolve in the same loop).
        let outbound = client
            .call_open_op("accept/echo", serde_json::json!({ "to": "hub" }))
            .await;
        assert!(
            outbound.result.is_ok(),
            "consumer→hub call still resolves, got {:?}",
            outbound.result
        );
    }

    // --- review 004 Unit 3 acceptance gates (F-05 op/register) ------------

    /// F-05 gate: the consumer announces an op over channel 0
    /// (`op/register` served on the accept side's fork), the hub
    /// registers it in the connection overlay, and the hub then calls
    /// it back — the forwarding stub issues a nested `call.requested`
    /// over channel 0 to the consumer, whose serving loop dispatches
    /// the real handler. Peer-announced op, discoverable and callable,
    /// end-to-end.
    #[tokio::test]
    async fn op_register_announce_then_hub_call_routes_back_to_consumer() {
        // Consumer side: serves `consumer/exec` locally; announces it.
        let consumer_registry = crate::registry::registration::OperationRegistry::new();
        consumer_registry
            .register(HandlerRegistration::new(
                external_query_spec("consumer/exec"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(
                        ctx.request_id,
                        serde_json::json!({ "ran_on": "consumer", "input": input }),
                    )
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let consumer_registry = Arc::new(consumer_registry);

        // Accept side: serves the `op/register` bootstrap op composed on
        // a fork (the F-02(a) shape — the fork is fresh here, so the
        // bootstrap set is just `op/register`).
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let accept_registry_for_hook =
            Arc::new(crate::registry::registration::OperationRegistry::new());
        let accept_tx_for_hook = accept_conn_tx.clone();
        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |_manager, channel0_conn, _auth| {
                let accept_registry = Arc::clone(&accept_registry_for_hook);
                let accept_conn_tx = accept_tx_for_hook.clone();
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;

                    let reg = accept_registry.fork();
                    reg.register(HandlerRegistration::new(
                        crate::registry::op_register::op_register_spec(AccessControl::default()),
                        HandlerKind::Once(crate::registry::op_register::op_register_handler(
                            Arc::clone(&call_connection),
                            Arc::clone(&accept_registry),
                        )),
                        OperationProvenance::Local,
                        None,
                        None,
                        crate::core::types::Capabilities::new(),
                    ))
                    .unwrap();
                    let dp = Dispatcher::new(
                        Arc::new(reg),
                        Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
                    );
                    dp.serve_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&consumer_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        // Consumer announces `consumer/exec` over channel 0.
        let announce = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                crate::registry::op_register::OP_REGISTER_NAME,
                crate::registry::op_register::OpRegisterRequest {
                    spec: external_query_spec("consumer/exec"),
                    replace: false,
                }
                .to_json(),
            ),
        )
        .await
        .expect("op/register announce timed out");
        assert!(
            announce.result.is_ok(),
            "announce ok, got {:?}",
            announce.result
        );

        // The hub's overlay now holds the announced op; the hub calls
        // it — the forwarding stub routes the nested call back over
        // channel 0 to the consumer's serving loop.
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            accept_conn.call("consumer/exec", serde_json::json!({ "task": "greet" })),
        )
        .await
        .expect("hub call to peer-announced op timed out");

        assert!(
            response.result.is_ok(),
            "hub→peer-announced-op call resolves, got {:?}",
            response.result
        );
        assert_eq!(
            response.result.unwrap(),
            serde_json::json!({ "ran_on": "consumer", "input": { "task": "greet" } })
        );
    }

    /// Interleaved-directions gate (review 005 G-01, third consequence):
    /// the consumer's outbound call resolves **while** its serving loop
    /// is mid-pump on a long-lived inbound Sub to the same peer. Under
    /// the pre-G-01 loop the Sub pump held the read loop inline and the
    /// outbound call's `call.responded` queued unresolved until the
    /// sweeper; sequential-direction gates (hub→consumer, *then*
    /// consumer→hub) could never catch it. The consumer also issues the
    /// outbound call *from inside* a wire-dispatched handler, the
    /// tightest interleaving.
    #[tokio::test]
    async fn outbound_call_resolves_while_inbound_subscription_is_being_served() {
        // Consumer side: serves `consumer/echo` (Once) and
        // `consumer/tick` (Sub, unbounded — held open until aborted).
        let consumer_registry = crate::registry::registration::OperationRegistry::new();
        consumer_registry
            .register(HandlerRegistration::new(
                external_query_spec("consumer/echo"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        consumer_registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "consumer/tick",
                    OperationType::Sub,
                    Visibility::External,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Stream(make_streaming_handler(|_input, ctx: OperationContext| {
                    let request_id = ctx.request_id.clone();
                    futures::stream::unfold((0u32, request_id), |(n, request_id)| async move {
                        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                        Some((
                            ResponseEnvelope::ok(
                                request_id.clone(),
                                serde_json::json!({ "tick": n }),
                            ),
                            (n + 1, request_id),
                        ))
                    })
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let consumer_registry = Arc::new(consumer_registry);

        // Handle channel: the install hook yields the accept side's
        // channel-0 `CallConnection` so the test driver can subscribe
        // (the accept side initiates the Sub) and so the
        // `hub/interleave` handler can reach it.
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);

        // Accept side (hub): serves `accept/echo` and
        // `hub/interleave` — the latter, when wire-dispatched from the
        // consumer, issues an outbound call on the *same* connection.
        let accept_registry = crate::registry::registration::OperationRegistry::new();
        accept_registry
            .register(HandlerRegistration::new(
                external_query_spec("accept/echo"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let accept_conn_for_handler =
            Arc::new(tokio::sync::Mutex::new(None::<Arc<CallConnection>>));
        let accept_conn_slot = Arc::clone(&accept_conn_for_handler);
        accept_registry
            .register(HandlerRegistration::new(
                external_query_spec("hub/interleave"),
                HandlerKind::Once(make_handler(move |input, ctx| {
                    let conn_slot = Arc::clone(&accept_conn_for_handler);
                    async move {
                        let conn = conn_slot
                            .lock()
                            .await
                            .clone()
                            .expect("hub call connection set");
                        let response = conn
                            .call("consumer/echo", serde_json::json!({ "via": "interleave" }))
                            .await;
                        let echo = response.result.expect("outbound call resolves");
                        ResponseEnvelope::ok(
                            ctx.request_id,
                            serde_json::json!({
                                "input": input,
                                "outbound": echo,
                            }),
                        )
                    }
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let accept_registry = Arc::new(accept_registry);

        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |_manager, channel0_conn, _auth| {
                let accept_registry = Arc::clone(&accept_registry);
                let accept_conn_tx = accept_conn_tx.clone();
                let accept_conn_slot = Arc::clone(&accept_conn_slot);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    accept_conn_slot
                        .lock()
                        .await
                        .replace(Arc::clone(&call_connection));
                    let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;
                    let dp = Dispatcher::new(
                        accept_registry,
                        Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
                    );
                    dp.serve_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&consumer_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        // Phase 1: the consumer subscribes to the hub's `consumer/tick`
        // — the consumer's serving loop pumps the Sub inline (pre-fix)
        // or in a spawned task (post-fix). The stream stays open.
        use futures::stream::StreamExt;
        let mut ticks = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.subscribe("consumer/tick", serde_json::json!({})),
        )
        .await
        .expect("subscribe timed out");
        let first = tokio::time::timeout(std::time::Duration::from_secs(5), ticks.next())
            .await
            .expect("first tick timed out")
            .expect("first tick item");
        assert!(
            first.result.is_ok(),
            "first tick ok, got {:?}",
            first.result
        );

        // Phase 2: with the Sub live, the consumer calls the hub's
        // `hub/interleave` — the wire-dispatched handler issues an
        // outbound hub→consumer call on the same connection while the
        // consumer's serving loop is serving the Sub. Pre-fix, the
        // consumer's serving loop held the Sub pump inline and this
        // call starved (30s sweeper); post-fix it resolves.
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.call_open_op(
                "hub/interleave",
                serde_json::json!({ "when": "while-sub-live" }),
            ),
        )
        .await
        .expect("interleaved outbound call starved while Sub served (G-01 shape)");
        assert!(
            response.result.is_ok(),
            "interleaved call resolves, got {:?}",
            response.result
        );
        let out = response.result.unwrap();
        assert_eq!(
            out.get("outbound"),
            Some(&serde_json::json!({ "via": "interleave" })),
            "the nested outbound hub→consumer call resolved through the serving loop"
        );

        // The Sub is still live after the interleaved call.
        let second = tokio::time::timeout(std::time::Duration::from_secs(5), ticks.next())
            .await
            .expect("second tick timed out")
            .expect("second tick item");
        assert!(
            second.result.is_ok(),
            "sub still live after interleaved call, got {:?}",
            second.result
        );
    }

    /// G-02 gate (review 005): the forwarding stub is exercised
    /// through **nested composition from a wire-dispatched hub
    /// handler** — the exact path the direct wire call in
    /// `op_register_announce_then_hub_call_routes_back_to_consumer`
    /// bypasses, and the path G-01 deadlocked (the inline dispatch
    /// blocked the read loop that had to resolve the stub's nested
    /// call; it resolved only via the 30s sweeper). The hub registry
    /// serves `hub/compose`, whose handler invokes the announced op
    /// through `context.env` — the `PeerCompositeEnv` →
    /// `OverlayOperationEnv` resolution a real hub handler produces.
    /// Resolving quickly (bounded well under the sweeper interval)
    /// proves the serving loop resolves same-connection nested
    /// composition live.
    #[tokio::test]
    async fn hub_handler_composes_peer_announced_op_via_nested_composition() {
        // Consumer side: serves `consumer/exec` locally; announces it.
        let consumer_registry = crate::registry::registration::OperationRegistry::new();
        consumer_registry
            .register(HandlerRegistration::new(
                external_query_spec("consumer/exec"),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(
                        ctx.request_id,
                        serde_json::json!({ "ran_on": "consumer", "input": input }),
                    )
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        let consumer_registry = Arc::new(consumer_registry);

        // Accept side (hub): serves `op/register` + `hub/compose` on a
        // fork. `hub/compose` composes the announced op via
        // `context.env` — the nested-composition shape ADR-022's
        // amendment promises. The stub is not invoked by the assertion
        // below directly; it is resolved through the connection
        // overlay `compose_root_env` attaches. The channel-0
        // connection carries the peer's identity (the assembly layer
        // resolves it from the accepted connection's `AuthContext`) —
        // `compose_root_env` attaches the connection overlay keyed by
        // that identity (ADR-030 §5); an identity-less connection gets
        // no overlay.
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let accept_registry_for_hook =
            Arc::new(crate::registry::registration::OperationRegistry::new());
        let accept_tx_for_hook = accept_conn_tx.clone();
        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |_manager, channel0_conn, auth| {
                if let Some(identity) = auth.identity.clone() {
                    let _ = channel0_conn.set_identity(identity);
                }
                let accept_registry = Arc::clone(&accept_registry_for_hook);
                let accept_conn_tx = accept_tx_for_hook.clone();
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;

                    let reg = accept_registry.fork();
                    reg.register(HandlerRegistration::new(
                        crate::registry::op_register::op_register_spec(AccessControl::default()),
                        HandlerKind::Once(crate::registry::op_register::op_register_handler(
                            Arc::clone(&call_connection),
                            Arc::clone(&accept_registry),
                        )),
                        OperationProvenance::Local,
                        None,
                        None,
                        crate::core::types::Capabilities::new(),
                    ))
                    .unwrap();
                    reg.register(HandlerRegistration::new(
                        external_query_spec("hub/compose"),
                        HandlerKind::Once(make_handler(|input, ctx| async move {
                            let name = input
                                .get("op")
                                .and_then(|v| v.as_str())
                                .unwrap_or("")
                                .to_string();
                            let (namespace, operation) = match name.split_once('/') {
                                Some(parts) => parts,
                                None => {
                                    return ResponseEnvelope::error(
                                        ctx.request_id,
                                        crate::protocol::wire::CallError::invalid_input(
                                            "hub/compose input missing `op` as `ns/op`",
                                        ),
                                    )
                                }
                            };
                            let response = ctx.env.invoke(namespace, operation, input, &ctx).await;
                            ResponseEnvelope {
                                request_id: ctx.request_id,
                                result: response.result,
                            }
                        })),
                        OperationProvenance::Local,
                        None,
                        Some(crate::registry::context::ScopedPeerEnv::new([
                            "consumer/exec",
                        ])),
                        crate::core::types::Capabilities::new(),
                    ))
                    .unwrap();
                    let dp = Dispatcher::new(
                        Arc::new(reg),
                        Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
                    );
                    dp.serve_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext {
            identity: Some(crate::core::auth::Identity {
                id: "consumer-peer".to_string(),
                scopes: vec![],
                resources: Default::default(),
            }),
            ..AuthContext::anonymous(b"alk/channels")
        };
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&consumer_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let _accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        // Consumer announces `consumer/exec` over channel 0.
        let announce = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                crate::registry::op_register::OP_REGISTER_NAME,
                crate::registry::op_register::OpRegisterRequest {
                    spec: external_query_spec("consumer/exec"),
                    replace: false,
                }
                .to_json(),
            ),
        )
        .await
        .expect("op/register announce timed out");
        assert!(
            announce.result.is_ok(),
            "announce ok, got {:?}",
            announce.result
        );

        // A wire-dispatched hub handler composes the announced op via
        // nested composition: the consumer calls `hub/compose` over the
        // wire, the hub's serving loop dispatches it, and the handler
        // resolves `consumer/exec` through the forwarding stub. Under
        // G-01 this resolved only at 30s via the sweeper; the bounded
        // timeout proves live resolution.
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.call_open_op(
                "hub/compose",
                serde_json::json!({ "op": "consumer/exec", "task": "nested" }),
            ),
        )
        .await
        .expect("nested composition through the forwarding stub timed out (G-01 shape)");

        assert!(
            response.result.is_ok(),
            "nested composition resolves through the forwarding stub, got {:?}",
            response.result
        );
        assert_eq!(
            response.result.unwrap(),
            serde_json::json!({ "ran_on": "consumer", "input": { "op": "consumer/exec", "task": "nested" } })
        );
    }

    // --- review 004 Unit 2 acceptance gate (F-02/F-06 fork) ----------------

    /// F-02/F-06 gate: the open op is registered on a **fork** of the
    /// base registry (base carries a placeholder op; the fork adds the
    /// openable + bootstrap discovery via
    /// `install_bootstrap_discovery`), the session dispatches over the
    /// fork, and — after the open op resolves over the live channels
    /// connection — the per-session openable is discoverable through
    /// `services/list` **on the fork** (the F-06 self-referential
    /// closure) and `services/schema` on the fork still validates.
    #[tokio::test]
    async fn fork_registry_open_op_resolves_and_is_discoverable() {
        use crate::channels::operations::{ChannelCore, OpenHandler};
        use crate::channels::policy::PerIdentityChannelPolicy;
        use crate::registry::spec::ChannelOpenSpec;

        // Base registry: a plain op + bootstrap discovery closed over
        // the *base* (which will NOT see per-session openables — that
        // is the F-06 finding; the fork's own discovery does).
        let base = crate::registry::registration::OperationRegistry::new();
        base.register(HandlerRegistration::new(
            external_query_spec("base/ping"),
            HandlerKind::Once(make_handler(|input, ctx| async move {
                ResponseEnvelope::ok(ctx.request_id, input)
            })),
            OperationProvenance::Local,
            None,
            None,
            crate::core::types::Capabilities::new(),
        ))
        .unwrap();

        let policy: Arc<PerIdentityChannelPolicy> = Arc::new(PerIdentityChannelPolicy::new(256));
        let policy_for_hook: Arc<dyn ChannelLifecyclePolicy> =
            Arc::clone(&policy) as Arc<dyn ChannelLifecyclePolicy>;

        let open_handler: OpenHandler = Arc::new(|_input, _plan, _channel_conn, _auth| {
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            })
        });

        let install_hook: crate::channels::adapter::InstallChannelZero =
            Arc::new(move |manager, channel0_conn, auth| {
                let open_handler = Arc::clone(&open_handler);
                let policy = Arc::clone(&policy_for_hook);
                let base = Arc::new(crate::registry::registration::OperationRegistry::new());
                base.register(HandlerRegistration::new(
                    external_query_spec("base/ping"),
                    HandlerKind::Once(make_handler(|input, ctx| async move {
                        ResponseEnvelope::ok(ctx.request_id, input)
                    })),
                    OperationProvenance::Local,
                    None,
                    None,
                    crate::core::types::Capabilities::new(),
                ))
                .unwrap();
                let base = Arc::clone(&base);
                tokio::spawn(async move {
                    let channel0_bidi = match channel0_conn.accept_bi().await {
                        Ok(s) => s,
                        Err(_) => return,
                    };
                    let (writer, reader) = split_single_stream(channel0_bidi);

                    let core = ChannelCore::new(manager, policy);
                    // The F-02(a) composition: fork the base, register
                    // the openable on the fork, install bootstrap
                    // discovery on the fork (F-06 — closed over the
                    // fork), dispatch over the fork.
                    let fork = base.fork();
                    core.register_openable(
                        OperationSpec::new(
                            "channels/tty/sub",
                            OperationType::Sub,
                            Visibility::External,
                            serde_json::json!({
                                "type": "object",
                                "properties": {
                                    "container": { "type": "string" }
                                },
                                "required": ["container"]
                            }),
                            serde_json::json!({
                                "type": "object",
                                "properties": {
                                    "channel_id": { "type": "integer" }
                                }
                            }),
                            vec![],
                            AccessControl::default(),
                            None,
                        )
                        .with_channel_open(ChannelOpenSpec::new("alk/tty")),
                        Arc::clone(&open_handler),
                        &fork,
                        auth.clone(),
                    )
                    .expect("register_openable on fork");
                    let fork = Arc::new(fork);
                    crate::registry::discovery::install_bootstrap_discovery(&fork)
                        .expect("bootstrap discovery on fork");

                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
                    let call_connection = Arc::new(CallConnection::new_single_stream(
                        channel0_conn,
                        Arc::clone(&writer),
                    ));
                    let dp = Dispatcher::new(fork, provider);
                    dp.run_loop_single_stream(call_connection, reader, writer)
                        .await;
                })
            });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let client = ChannelClient::from_connection(client_conn)
            .await
            .expect("channel client init");

        // 1. The open op resolves over the live connection through the fork.
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                "channels/tty/sub",
                serde_json::json!({ "container": "abc" }),
            ),
        )
        .await
        .expect("open op timed out");
        assert!(
            response.result.is_ok(),
            "open op resolves through the fork, got {:?}",
            response.result
        );
        let channel_id = response
            .result
            .unwrap()
            .get("channel_id")
            .and_then(|v| v.as_u64())
            .expect("channel_id in response");
        assert!(channel_id > 0);

        // 2. The per-session openable is discoverable via services/list
        //    (F-06) — the fork's own discovery closure sees the fork's ops.
        let listing = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op("services/list", serde_json::json!({})),
        )
        .await
        .expect("services/list timed out");
        assert!(
            listing.result.is_ok(),
            "services/list resolves, got {:?}",
            listing.result
        );
        let names: Vec<String> = listing
            .result
            .unwrap()
            .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!(
            names.contains(&"channels/tty/sub".to_string()),
            "per-session openable discoverable through the fork's discovery: {names:?}"
        );

        // 3. services/schema on the fork still validates input (the
        //    bootstrap op is on the fork and answers).
        let schema = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                "services/schema",
                serde_json::json!({ "name": "channels/tty/sub" }),
            ),
        )
        .await
        .expect("services/schema timed out");
        assert!(
            schema.result.is_ok(),
            "services/schema resolves on the fork, got {:?}",
            schema.result
        );
    }

    // --- CF-005 regression gates -----------------------------------------

    /// The scope-gated open op the connect side serves — the
    /// alktunnels reverse-flow POC shape (`worker/tunnel/open` with
    /// `required_scopes`). The handler reports the identity the
    /// dispatch resolved, so the tests can assert which path won.
    fn scope_gated_open_registry() -> crate::registry::registration::OperationRegistry {
        let registry = crate::registry::registration::OperationRegistry::new();
        let acl = AccessControl {
            required_scopes: vec!["tunnel:open".to_string()],
            ..Default::default()
        };
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "worker/tunnel/open",
                    OperationType::Mutation,
                    Visibility::External,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    acl,
                    None,
                ),
                HandlerKind::Once(make_handler(|_input, ctx| async move {
                    ResponseEnvelope::ok(
                        ctx.request_id,
                        serde_json::json!({
                            "opened": true,
                            "caller": ctx.identity.map(|i| i.id),
                        }),
                    )
                })),
                OperationProvenance::Local,
                None,
                None,
                crate::core::types::Capabilities::new(),
            ))
            .unwrap();
        registry
    }

    /// The accept-side (hub) install hook: serves nothing but echoes
    /// nothing — it hands the channel-0 `CallConnection` back to the
    /// test and runs the resolution-only serving loop, so the test can
    /// initiate hub→worker calls.
    fn cf005_accept_side_hook(
        accept_conn_tx: tokio::sync::mpsc::Sender<Arc<CallConnection>>,
    ) -> crate::channels::adapter::InstallChannelZero {
        Arc::new(move |_manager, channel0_conn, _auth| {
            let accept_conn_tx = accept_conn_tx.clone();
            tokio::spawn(async move {
                let channel0_bidi = match channel0_conn.accept_bi().await {
                    Ok(s) => s,
                    Err(_) => return,
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));
                let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;
                let reader = reader;
                crate::protocol::connection::read_single_stream_until_closed(
                    reader,
                    call_connection.pending(),
                )
                .await;
            })
        })
    }

    fn hub_identity() -> crate::core::auth::Identity {
        crate::core::auth::Identity {
            id: "hub".to_string(),
            scopes: vec!["tunnel:open".to_string()],
            resources: Default::default(),
        }
    }

    fn effective_override_identity() -> crate::core::auth::Identity {
        crate::core::auth::Identity {
            id: "worker-effective".to_string(),
            scopes: vec!["tunnel:open".to_string()],
            resources: Default::default(),
        }
    }

    /// CF-005 remediation (b) — the transport connection's identity
    /// propagates to channel 0, so a scope-gated op served by the
    /// connect side authenticates the peer by transport identity (the
    /// mTLS/QUIC key-based path). The consumer dials, sets the
    /// transport identity (the assembly layer resolves it from the
    /// transport handshake), then `from_connection_with_serving`; the
    /// hub calls the scope-gated op with no token and is authorized.
    #[tokio::test]
    async fn cf005_transport_identity_propagates_to_connect_side_serving() {
        let worker_registry = Arc::new(scope_gated_open_registry());
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let install_hook = cf005_accept_side_hook(accept_conn_tx);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        client_conn
            .set_identity(hub_identity())
            .expect("transport identity set once");
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let _client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&worker_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
        )
        .await
        .expect("hub→worker scope-gated call timed out");

        let out = response
            .result
            .expect("scope-gated op authorized by the propagated transport identity");
        assert_eq!(out["opened"], serde_json::json!(true));
        assert_eq!(
            out["caller"], "hub",
            "the handler saw the transport-derived identity"
        );
    }

    /// CF-005 remediation (a) — `ServingConfig.identity` explicitly
    /// overrides, winning over the transport connection's identity.
    /// Both are set here; the handler must see the override.
    #[tokio::test]
    async fn cf005_serving_config_identity_overrides_transport_identity() {
        let worker_registry = Arc::new(scope_gated_open_registry());
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let install_hook = cf005_accept_side_hook(accept_conn_tx);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        client_conn
            .set_identity(hub_identity())
            .expect("transport identity set once");
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let _client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&worker_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: Some(effective_override_identity()),
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
        )
        .await
        .expect("hub→worker scope-gated call timed out");

        let out = response
            .result
            .expect("the override identity satisfies the scope gate");
        assert_eq!(out["caller"], "worker-effective", "the override won");
    }

    /// CF-005 negative case — no identity anywhere (no
    /// `ServingConfig.identity`, no transport identity, no token): the
    /// scope-gated op is denied `FORBIDDEN` ("authentication
    /// required"). Fails closed, unchanged.
    #[tokio::test]
    async fn cf005_scope_gated_op_denied_without_any_identity() {
        let worker_registry = Arc::new(scope_gated_open_registry());
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let install_hook = cf005_accept_side_hook(accept_conn_tx);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let _client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&worker_registry),
                identity_provider: Arc::new(NoopIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
        )
        .await
        .expect("hub→worker scope-gated call timed out");

        let err = response
            .result
            .expect_err("identity-less call must be denied");
        assert_eq!(err.code, "FORBIDDEN", "no identity on the connect side");
        assert!(
            err.message.contains("authentication required"),
            "AccessControl::check(None) message, got: {}",
            err.message
        );
    }

    /// CF-005 token fallback — the payload `auth_token` →
    /// `ServingConfig.identity_provider` path still works (the
    /// hub-forwarding / browser-token path, ADR-017 §7), with
    /// precedence over the propagated transport identity.
    #[tokio::test]
    async fn cf005_auth_token_fallback_and_precedence() {
        struct TokenIdProvider;
        impl IdentityProvider for TokenIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<crate::core::auth::Identity> {
                None
            }
            fn resolve_from_token(
                &self,
                token: &crate::core::auth::AuthToken,
            ) -> Option<crate::core::auth::Identity> {
                if token.raw == b"alk_worker_token" {
                    Some(crate::core::auth::Identity {
                        id: "token-resolved".to_string(),
                        scopes: vec!["tunnel:open".to_string()],
                        resources: Default::default(),
                    })
                } else {
                    None
                }
            }
        }

        let worker_registry = Arc::new(scope_gated_open_registry());
        let (accept_conn_tx, mut accept_conn_rx) =
            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
        let install_hook = cf005_accept_side_hook(accept_conn_tx);

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let client_conn =
            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
        client_conn
            .set_identity(hub_identity())
            .expect("transport identity set once");
        let server_conn =
            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));

        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
        let auth = AuthContext::anonymous(b"alk/channels");
        let _server_handle = tokio::spawn(async move {
            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
        });

        let _client = ChannelClient::from_connection_with_serving(
            client_conn,
            Some(ServingConfig {
                registry: Arc::clone(&worker_registry),
                identity_provider: Arc::new(TokenIdProvider),
                identity: None,
            }),
        )
        .await
        .expect("channel client init");

        let accept_conn = accept_conn_rx
            .recv()
            .await
            .expect("accept side channel-0 connection handle");

        // Tokenless: the propagated transport identity authorizes.
        let transport = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
        )
        .await
        .expect("transport-identity call timed out");
        let out = transport
            .result
            .expect("transport identity authorizes the scope gate");
        assert_eq!(out["caller"], "hub");

        // With a token: the token resolution wins.
        let tokened = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            accept_conn.call_with_payload(serde_json::json!({
                "operationId": "worker/tunnel/open",
                "input": { "c": 2 },
                "auth_token": "alk_worker_token",
            })),
        )
        .await
        .expect("token call timed out");
        let out = tokened
            .result
            .expect("the token resolves an identity satisfying the scope gate");
        assert_eq!(
            out["caller"], "token-resolved",
            "the token resolution won over the transport identity"
        );
    }
}