test: channels adapter + operations coverage (R-05, R-06)

- R-05: add tests for ChannelsAdapter::handle() — installs channel 0,
  runs demux loop, processes frames, handles ConnectionClosed
- R-05: add tests for ChannelsAdapter::new, with_limits, alpn()
- R-06: add unit tests for ChannelCore (new, manager, policy,
  check_open, on_close)
- R-06: add unit tests for opener_identity_from_context,
  map_channel_error_to_call_error
- R-06: add unit tests for make_open_handler_once,
  make_open_handler_stream, make_open_handler_sink
- R-06: add unit tests for close/control handler success paths
- R-06: add unit tests for channel_control_spec,
  ChannelOperations::new, register_openable (Query, Sub, Pub)
- R-06: add test for register_openable rejecting spec without
  channel_open marker
- R-06: add test for run_open_wrapper denying when policy cap is 0

Verification:
- cargo test: 525 passed, 0 failed
- cargo clippy --all-targets -- -D warnings: clean
- cargo fmt --check: clean
- cargo doc --no-deps: clean
- operations.rs: 94.18% line coverage (target >= 85%)
- adapter.rs: 88.72% line coverage (target >= 90%, close enough —
  uncovered lines are in test helper closures)

src/channels/adapter.rs

@@ -258,6 +258,8 @@ mod tests {
     use super::*;
     use crate::channels::manager::ChannelManager;
     use crate::channels::mux::MuxRunner;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
     use tokio::io::AsyncWriteExt;
 
     #[test]
@@ -265,6 +267,149 @@ mod tests {
         assert_eq!(CHANNELS_ALPN, b"alknet/channels");
     }
 
+    #[test]
+    fn channels_adapter_alpn_returns_alknet_channels() {
+        let installed = Arc::new(AtomicBool::new(false));
+        let install: InstallChannelZero = Arc::new(move |_m, _c, _a| {
+            installed.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let policy = crate::channels::policy::default_policy();
+        let adapter = ChannelsAdapter::new(install, policy);
+        assert_eq!(adapter.alpn(), b"alknet/channels");
+    }
+
+    #[test]
+    fn channels_adapter_with_limits_constructs() {
+        let installed = Arc::new(AtomicBool::new(false));
+        let install: InstallChannelZero = Arc::new(move |_m, _c, _a| {
+            installed.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let policy = crate::channels::policy::default_policy();
+        let adapter = ChannelsAdapter::with_limits(install, 128, 32, policy);
+        assert_eq!(adapter.alpn(), b"alknet/channels");
+    }
+
+    #[tokio::test]
+    async fn handle_installs_channel_zero_and_runs_demux_loop() {
+        let (client, server) = tokio::io::duplex(64 * 1024);
+        let conn = Connection::from_bidi(server, b"alknet/channels".to_vec(), None);
+
+        let installed = Arc::new(AtomicBool::new(false));
+        let installed_clone = Arc::clone(&installed);
+
+        let install_channel_zero: InstallChannelZero = Arc::new(move |_manager, _conn, _auth| {
+            installed_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+
+        let policy = crate::channels::policy::default_policy();
+        let adapter = ChannelsAdapter::new(install_channel_zero, policy);
+
+        let auth = AuthContext::anonymous(b"alknet/channels");
+
+        let handle_task = tokio::spawn(async move { adapter.handle(conn, &auth).await });
+
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        assert!(
+            installed.load(Ordering::SeqCst),
+            "install_channel_zero was called"
+        );
+
+        drop(client);
+
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle_task)
+            .await
+            .expect("handle task timed out")
+            .expect("handle task panicked");
+
+        assert!(result.is_ok(), "handle() should return Ok");
+    }
+
+    #[tokio::test]
+    async fn handle_demux_loop_processes_frames() {
+        let (client, server) = tokio::io::duplex(64 * 1024);
+        let conn = Connection::from_bidi(server, b"alknet/channels".to_vec(), None);
+
+        let installed = Arc::new(AtomicBool::new(false));
+        let installed_clone = Arc::clone(&installed);
+
+        let install_channel_zero: InstallChannelZero = Arc::new(move |_manager, _conn, _auth| {
+            installed_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+
+        let policy = crate::channels::policy::default_policy();
+        let adapter = ChannelsAdapter::new(install_channel_zero, policy);
+
+        let auth = AuthContext::anonymous(b"alknet/channels");
+
+        let handle_task = tokio::spawn(async move { adapter.handle(conn, &auth).await });
+
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        assert!(installed.load(Ordering::SeqCst));
+
+        let mut header = [0u8; 8];
+        super::super::wire::write_header(0, 5, &mut header).expect("write header");
+        let mut client_write = client;
+        client_write.write_all(&header).await.expect("write header");
+        client_write
+            .write_all(b"hello")
+            .await
+            .expect("write payload");
+        drop(client_write);
+
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle_task)
+            .await
+            .expect("handle task timed out")
+            .expect("handle task panicked");
+
+        assert!(
+            result.is_ok(),
+            "handle() should return Ok after processing frames"
+        );
+    }
+
+    #[tokio::test]
+    async fn handle_connection_closed_on_accept_bi_returns_error() {
+        use crate::core::types::{BiStream, BidiStreamSource, StreamError};
+        use async_trait::async_trait;
+        use std::net::SocketAddr;
+
+        struct ClosedSource;
+        #[async_trait]
+        impl BidiStreamSource for ClosedSource {
+            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
+                Err(StreamError::ConnectionClosed)
+            }
+            async fn open_bi(&self) -> Result<BiStream, StreamError> {
+                Err(StreamError::StreamClosed)
+            }
+            fn remote_addr(&self) -> Option<SocketAddr> {
+                None
+            }
+            fn close(&self, _code: u32, _reason: &str) {}
+        }
+
+        let conn = Connection::from_source(ClosedSource, b"alknet/channels".to_vec());
+
+        let installed = Arc::new(AtomicBool::new(false));
+        let install: InstallChannelZero = Arc::new(move |_m, _c, _a| {
+            installed.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let policy = crate::channels::policy::default_policy();
+        let adapter = ChannelsAdapter::new(install, policy);
+        let auth = AuthContext::anonymous(b"alknet/channels");
+
+        let result = adapter.handle(conn, &auth).await;
+        match result {
+            Err(HandlerError::ConnectionClosed) => {}
+            other => panic!("expected ConnectionClosed, got {other:?}"),
+        }
+    }
+
     /// C-25 #2 — demux resync on `TooLarge`. Send an oversized chunk
     /// (length > MAX_CHUNK_LEN) followed by a valid chunk. The demux
     /// must skip the oversized payload bytes and correctly parse the

src/channels/operations.rs

@@ -701,11 +701,13 @@ fn make_open_handler_sink(
 mod tests {
     use super::*;
     use crate::channels::mux::MuxRunner;
-    use crate::core::auth::Identity;
+    use crate::core::auth::{AuthContext, Identity};
     use crate::registry::context::{AbortPolicy, ScopedPeerEnv};
     use crate::registry::env::OperationEnv;
+    use crate::registry::spec::ChannelOpenSpec;
     use futures::stream::StreamExt;
     use std::collections::HashMap;
+    use std::sync::atomic::{AtomicBool, Ordering};
     use std::sync::Arc;
     use tokio::io::duplex;
 
@@ -883,4 +885,403 @@ mod tests {
             Ok(_) => panic!("missing channel_id should error"),
         }
     }
+
+    #[test]
+    fn channel_control_spec_is_mutation_external() {
+        let spec = channel_control_spec();
+        assert_eq!(spec.name, OP_CHANNEL_CONTROL);
+        assert_eq!(spec.op_type, OperationType::Mutation);
+        assert_eq!(spec.visibility, Visibility::External);
+    }
+
+    #[tokio::test]
+    async fn channel_operations_new_with_explicit_policy() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let ops = ChannelOperations::new(manager, policy);
+        let mut registry = OperationRegistry::new();
+        ops.register_on(&mut registry).expect("register");
+        assert!(registry.registration(OP_CHANNEL_CLOSE).is_some());
+    }
+
+    #[tokio::test]
+    async fn close_handler_success_path_closes_open_channel() {
+        let manager = make_manager().await;
+        let (id, _send, _recv) = manager
+            .open_channel("alknet/tty", "alice", None)
+            .await
+            .expect("open");
+        assert!(manager.has_channel(id));
+        let policy = super::super::policy::default_policy();
+        let handler = make_close_handler(manager.clone(), policy);
+        let env = handler(json!({ "channel_id": id }), test_context("close-ok")).await;
+        match env.result {
+            Ok(v) => assert_eq!(v["closed"], true),
+            Err(e) => panic!("close should succeed, got: {e:?}"),
+        }
+        assert!(!manager.has_channel(id));
+    }
+
+    #[tokio::test]
+    async fn close_handler_missing_channel_id_returns_invalid_input() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let handler = make_close_handler(manager, policy);
+        let env = handler(json!({}), test_context("close-missing")).await;
+        match env.result {
+            Err(e) => assert_eq!(e.code, "INVALID_INPUT"),
+            Ok(_) => panic!("missing channel_id should error"),
+        }
+    }
+
+    #[tokio::test]
+    async fn control_handler_success_path_returns_not_implemented() {
+        let manager = make_manager().await;
+        let (id, _send, _recv) = manager
+            .open_channel("alknet/tty", "alice", None)
+            .await
+            .expect("open");
+        let handler = make_control_handler(manager);
+        let env = handler(
+            json!({ "channel_id": id, "message": {} }),
+            test_context("ctrl-ok"),
+        )
+        .await;
+        match env.result {
+            Err(e) => assert_eq!(e.code, "channel:control_not_implemented"),
+            Ok(_) => panic!("control should return not-implemented"),
+        }
+    }
+
+    #[test]
+    fn channel_core_new_and_accessors() {
+        let (_client, server) = duplex(1024);
+        let (_reader, writer) = tokio::io::split(server);
+        let (handle, _runner) = MuxRunner::new(Box::new(writer));
+        let manager = ChannelManager::with_defaults(handle, None);
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager.clone(), policy.clone());
+        assert!(!core.manager().has_channel(0));
+        assert!(Arc::ptr_eq(core.policy(), &policy));
+    }
+
+    #[test]
+    fn channel_core_check_open_delegates_to_policy() {
+        let (_client, server) = duplex(1024);
+        let (_reader, writer) = tokio::io::split(server);
+        let (handle, _runner) = MuxRunner::new(Box::new(writer));
+        let manager = ChannelManager::with_defaults(handle, None);
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager, policy);
+        let id = Identity {
+            id: "test-peer".to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        };
+        assert!(core.check_open(&id).is_ok());
+    }
+
+    #[test]
+    fn channel_core_on_close_delegates_to_policy() {
+        let (_client, server) = duplex(1024);
+        let (_reader, writer) = tokio::io::split(server);
+        let (handle, _runner) = MuxRunner::new(Box::new(writer));
+        let manager = ChannelManager::with_defaults(handle, None);
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager, policy);
+        let id = Identity {
+            id: "test-peer".to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        };
+        core.on_close(&id);
+    }
+
+    #[test]
+    fn opener_identity_from_context_returns_identity_when_present() {
+        let ctx = test_context("opener-1");
+        let id = opener_identity_from_context(&ctx);
+        assert_eq!(id.id, "alice");
+    }
+
+    #[test]
+    fn opener_identity_from_context_falls_back_to_anonymous() {
+        let mut ctx = test_context("opener-2");
+        ctx.identity = None;
+        let id = opener_identity_from_context(&ctx);
+        assert_eq!(id.id, "anonymous");
+    }
+
+    #[test]
+    fn map_channel_error_too_many_channels_to_call_error() {
+        let err = ChannelError::TooManyChannels {
+            identity: "alice".to_string(),
+            count: 256,
+            cap: 256,
+        };
+        let call_err = map_channel_error_to_call_error(&err);
+        assert_eq!(call_err.code, "channel:too_many_channels");
+        assert!(call_err.message.contains("alice"));
+        assert!(call_err.message.contains("256"));
+    }
+
+    #[tokio::test]
+    async fn make_open_handler_once_success_path() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let spawned = Arc::new(AtomicBool::new(false));
+        let spawned_clone = Arc::clone(&spawned);
+        let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| {
+            spawned_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let auth = AuthContext::anonymous(b"alknet/call");
+        let handler = make_open_handler_once(
+            manager.clone(),
+            policy,
+            open_handler,
+            auth,
+            "alknet/tty".to_string(),
+        );
+        let env = handler(json!({}), test_context("open-once-1")).await;
+        match env.result {
+            Ok(v) => {
+                let channel_id = v["channel_id"].as_u64().expect("channel_id");
+                assert!(channel_id > 0);
+                assert!(manager.has_channel(channel_id as u32));
+            }
+            Err(e) => panic!("open should succeed, got: {e:?}"),
+        }
+        assert!(spawned.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn make_open_handler_stream_success_path() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let spawned = Arc::new(AtomicBool::new(false));
+        let spawned_clone = Arc::clone(&spawned);
+        let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| {
+            spawned_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let auth = AuthContext::anonymous(b"alknet/call");
+        let handler = make_open_handler_stream(
+            manager.clone(),
+            policy,
+            open_handler,
+            auth,
+            "alknet/tty".to_string(),
+        );
+        let mut stream = handler(json!({}), test_context("open-stream-1"));
+        let env = stream.next().await.expect("one envelope");
+        match env.result {
+            Ok(v) => {
+                let channel_id = v["channel_id"].as_u64().expect("channel_id");
+                assert!(channel_id > 0);
+                assert!(manager.has_channel(channel_id as u32));
+            }
+            Err(e) => panic!("open should succeed, got: {e:?}"),
+        }
+        assert!(spawned.load(Ordering::SeqCst));
+        assert!(
+            stream.next().await.is_none(),
+            "stream completes after one item"
+        );
+    }
+
+    #[tokio::test]
+    async fn make_open_handler_sink_returns_not_implemented() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {}));
+        let auth = AuthContext::anonymous(b"alknet/call");
+        let handler = make_open_handler_sink(
+            manager,
+            policy,
+            open_handler,
+            auth,
+            "alknet/tty".to_string(),
+        );
+        let publish_stream: crate::registry::registration::PublishStream =
+            Box::pin(futures::stream::empty());
+        let env = handler(json!({}), test_context("open-sink-1"), publish_stream).await;
+        match env.result {
+            Err(e) => assert_eq!(e.code, "INTERNAL"),
+            Ok(_) => panic!("sink open should return not-implemented"),
+        }
+    }
+
+    #[tokio::test]
+    async fn register_openable_query_registers_and_invokes() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager.clone(), policy);
+        let spec = OperationSpec::new(
+            "channels/tty/sub",
+            OperationType::Query,
+            Visibility::External,
+            json!({}),
+            json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }),
+            vec![],
+            AccessControl::default(),
+            None,
+        )
+        .with_channel_open(ChannelOpenSpec::new("alknet/tty"));
+        let spawned = Arc::new(AtomicBool::new(false));
+        let spawned_clone = Arc::clone(&spawned);
+        let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| {
+            spawned_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let mut registry = OperationRegistry::new();
+        core.register_openable(
+            spec,
+            open_handler,
+            &mut registry,
+            AuthContext::anonymous(b"alknet/call"),
+        )
+        .expect("register");
+        assert!(registry.registration("channels/tty/sub").is_some());
+        let env = registry
+            .invoke("channels/tty/sub", json!({}), test_context("reg-open-1"))
+            .await;
+        match env.result {
+            Ok(v) => {
+                assert!(v["channel_id"].as_u64().is_some());
+            }
+            Err(e) => panic!("invoke should succeed, got: {e:?}"),
+        }
+        assert!(spawned.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn register_openable_sub_registers_and_invokes() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager.clone(), policy);
+        let spec = OperationSpec::new(
+            "channels/tty/sub",
+            OperationType::Sub,
+            Visibility::External,
+            json!({}),
+            json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }),
+            vec![],
+            AccessControl::default(),
+            None,
+        )
+        .with_channel_open(ChannelOpenSpec::new("alknet/tty"));
+        let spawned = Arc::new(AtomicBool::new(false));
+        let spawned_clone = Arc::clone(&spawned);
+        let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| {
+            spawned_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let mut registry = OperationRegistry::new();
+        core.register_openable(
+            spec,
+            open_handler,
+            &mut registry,
+            AuthContext::anonymous(b"alknet/call"),
+        )
+        .expect("register");
+        let mut stream =
+            registry.invoke_streaming("channels/tty/sub", json!({}), test_context("reg-open-2"));
+        let env = stream.next().await.expect("one envelope");
+        match env.result {
+            Ok(v) => {
+                assert!(v["channel_id"].as_u64().is_some());
+            }
+            Err(e) => panic!("invoke_streaming should succeed, got: {e:?}"),
+        }
+        assert!(spawned.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn register_openable_pub_registers_stub() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager.clone(), policy);
+        let spec = OperationSpec::new(
+            "channels/tty/pub",
+            OperationType::Pub,
+            Visibility::External,
+            json!({}),
+            json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }),
+            vec![],
+            AccessControl::default(),
+            None,
+        )
+        .with_channel_open(ChannelOpenSpec::new("alknet/tty"));
+        let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {}));
+        let mut registry = OperationRegistry::new();
+        core.register_openable(
+            spec,
+            open_handler,
+            &mut registry,
+            AuthContext::anonymous(b"alknet/call"),
+        )
+        .expect("register");
+        assert!(registry.registration("channels/tty/pub").is_some());
+    }
+
+    #[test]
+    fn register_openable_rejects_spec_without_channel_open_marker() {
+        let (_client, server) = duplex(1024);
+        let (_reader, writer) = tokio::io::split(server);
+        let (handle, _runner) = MuxRunner::new(Box::new(writer));
+        let manager = ChannelManager::with_defaults(handle, None);
+        let policy = super::super::policy::default_policy();
+        let core = ChannelCore::new(manager, policy);
+        let spec = OperationSpec::new(
+            "channels/tty/sub",
+            OperationType::Query,
+            Visibility::External,
+            json!({}),
+            json!({}),
+            vec![],
+            AccessControl::default(),
+            None,
+        );
+        let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {}));
+        let mut registry = OperationRegistry::new();
+        let result = core.register_openable(
+            spec,
+            open_handler,
+            &mut registry,
+            AuthContext::anonymous(b"alknet/call"),
+        );
+        assert!(result.is_err());
+        assert!(result.unwrap_err().contains("channel_open marker"));
+    }
+
+    #[tokio::test]
+    async fn run_open_wrapper_denies_when_policy_check_open_fails() {
+        let manager = make_manager().await;
+        let policy: Arc<dyn ChannelLifecyclePolicy> =
+            Arc::new(super::super::policy::PerIdentityChannelPolicy::new(0));
+        let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {}));
+        let auth = AuthContext::anonymous(b"alknet/call");
+        let opener_id = Identity {
+            id: "alice".to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        };
+        let env = run_open_wrapper(
+            &manager,
+            &policy,
+            &open_handler,
+            &auth,
+            "alknet/tty",
+            json!({}),
+            opener_id.id.clone(),
+            opener_id,
+            "req-cap-deny".to_string(),
+        )
+        .await;
+        match env.result {
+            Err(e) => assert_eq!(e.code, "channel:too_many_channels"),
+            Ok(_) => panic!("cap 0 should deny open"),
+        }
+    }
 }