diff --git a/docs/architecture/decisions/049-channel-open-establishment-phase.md b/docs/architecture/decisions/049-channel-open-establishment-phase.md index 72b7495..3ac6299 100644 --- a/docs/architecture/decisions/049-channel-open-establishment-phase.md +++ b/docs/architecture/decisions/049-channel-open-establishment-phase.md @@ -6,7 +6,8 @@ Accepted — implemented in alkcall 0.5.0 (Unit 1: E-01 + N-1; see the "Amendment (Unit 1 implementation, 2026-09-06)" at the bottom). Amends ADR-047 §3 — the open-op wrapper gains an awaited establishment phase ahead of the spawned pump handler; resolves review -006 E-01 and N-1 +006 E-01 and N-1. Amendment 3 (2026-09-16, review 008 U-2) adds the +establisher reply projection — see "Amendment 3" below. ## Context @@ -379,8 +380,8 @@ is unfixable within that shape; alktty documented the same wall (backend `allocate` cannot cross, so failure classes stayed in-band — the phantom-channel shape ADR-049 removed, alive one layer down). -The plan is now real: `Establishment { plan: Option }` -with `ChannelPlan = Arc` — **typed-opaque, not +The plan is now real: `Establishment { plan: Option }` with +`ChannelPlan = Arc` — **typed-opaque, not `serde_json::Value`**. The review's `Option` sketch could not satisfy its own verification gate ("establisher dials, `plan` carries the handle"): the payloads establishers actually hand off are live @@ -414,4 +415,67 @@ and the registration entry points, plus a `debug!` telemetry line in `run_open_wrapper` when a handler exits without having accepted the channel's `BiStream` (the birth-teardown hint; the accept is observable in-process via the yield-once source). §6's pinned -EOF-shaped panic semantics are unchanged. \ No newline at end of file +EOF-shaped panic semantics are unchanged. + +## Amendment 3 (establisher reply projection, 2026-09-16 — review 008 U-2) + +Amendment 2 filled the establisher → handler direction (the plan); +the establisher → opener direction stayed empty — the wrapper +hardcodes the success reply to `channel_id` +(`run_open_wrapper`'s `ResponseEnvelope::ok`). alktunnels ADR-008's +bind-first listen establisher ends establishment at bind time, and the +observed OS-chosen bound address must ride the open-op reply as an +additive `"bound"` field (the SOCKS5 BIND reply#1 `BND.ADDR` fidelity +ask) — the alternative (an out-of-band query op) is a new wire +surface, strictly worse than an optional reply field. + +**The decision: `Establishment` gains optional reply fields the +wrapper merges into the success output.** + +- The carrier: `Establishment.reply_fields: Option>` (private, builder-constructed) — + `Establishment::new(plan).with_reply_field("bound", json!({...}))`, + plus `with_reply_fields(map)` (batch) and `reply_fields()` (read). + The map shape generalizes beyond `bound` without a fourth + amendment; `#[non_exhaustive]` (amendment 2) made the carrier + extension additive — no construction-site break. +- The merge: on establisher success the wrapper extends the success + output with the contributed fields AFTER reserving `channel_id`. + **The `channel_id` reservation:** the key is wrapper-owned; an + establisher supplying it is an establisher bug — the wrapper fails + the open loudly (`channel:open_failed`, reason `handler_error`, + message naming the reserved key) and tears the just-allocated + channel down, never silently shadowing its own value. Contributed + fields otherwise merge flat; the output schema remains the op's own + concern (the producing crate documents its optional fields — + alktunnels' listen-op spec documents `bound`). +- The client read path: `ChannelClient::open_channel_with_reply` + returns `(channel_id, reply, send, recv)` — the full success + output — so a consumer reads `bound` without dropping to + `call_open_op` + manual `adopt_channel`; + `open_channel` delegates and discards the extra fields, signature + unchanged. +- Compatibility: absent reply fields leave the reply byte-identical + to the pre-amendment shape (`{ channel_id }`) — verified by test. + Old consumers (`open_channel` extracting `channel_id`, ignoring + unknown fields) are unaffected by a NEW field; new consumers reading + `bound` against an OLD alkcall see the field absent — the additive + posture alktunnels ADR-008 §2 pins. + +Door type: the reply fields are an additive call-plane JSON surface +(like `channel:open_failed`'s `details` in §3 — consumers ignore +unknown fields by the envelope's own parse posture); the field set an +op may contribute is the producing crate's schema concern, not a +protocol vocabulary. No data-plane change. + +Implemented surface (alkcall 0.7.2): `Establishment::with_reply_field` +/ `with_reply_fields` / `reply_fields`, the wrapper's +reservation check (`RESERVED_REPLY_KEY` = `channel_id`) + +`merge_reply_fields`, `ChannelClient::open_channel_with_reply`. +Verification gates landed as tests: the projection produces +`{ channel_id, bound }` on the wire; no-fields (establisher or not) +produces exactly `{ channel_id }` (byte-identical); the reserved-key +establisher fails with reason `handler_error`, channel torn down, +ledger decremented, handler never spawned; the e2e gate carries +`bound` through a real channels connection to +`open_channel_with_reply` while `open_channel` stays unchanged. \ No newline at end of file diff --git a/src/channels/client.rs b/src/channels/client.rs index 03493b8..08c711b 100644 --- a/src/channels/client.rs +++ b/src/channels/client.rs @@ -346,6 +346,26 @@ impl ChannelClient { 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 @@ -359,7 +379,7 @@ impl ChannelClient { .adopt_channel(channel_id, alpn, None) .await .map_err(ChannelOpenError::AdoptFailed) - .map(|(send, recv)| (channel_id, send, recv)) + .map(|(send, recv)| (channel_id, out, send, recv)) } /// Take the `CallConnection` — used by the consumer to register @@ -1625,6 +1645,128 @@ mod tests { 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, + ®istry, + auth.clone(), + None, + ) + .expect("register_openable_with_establisher"); + let registry = Arc::new(registry); + let provider: Arc = 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 004 Unit 3 acceptance gates (F-04 serving half) ----------- /// F-04 gate 1: hub→consumer call over an existing `ChannelClient` diff --git a/src/channels/operations.rs b/src/channels/operations.rs index 807cc33..314320a 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -17,7 +17,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use futures::future::BoxFuture; -use serde_json::{json, Value}; +use serde_json::{json, Map, Value}; use crate::core::auth::{AuthContext, Identity}; use crate::core::types::{Capabilities, Connection}; @@ -374,14 +374,18 @@ pub type OpenHandler = Arc< pub type ChannelPlan = Arc; /// The establishment-phase result (ADR-049 §1, as filled by ADR-049 -/// amendment 2 — R-01). Carries the channel plan the wrapper threads -/// to the pump handler's `plan` parameter ([`OpenHandler`]). The -/// plan is optional: an establisher that only validates (TTY's -/// lookup/ownership shape) returns [`Establishment::default`]. +/// amendment 2 — R-01; extended by amendment 3 — review 008 U-2). +/// Carries the channel plan the wrapper threads to the pump handler's +/// `plan` parameter ([`OpenHandler`]) and optional reply fields the +/// wrapper merges into the open-op success reply (the establisher → +/// opener direction). The plan is optional: an establisher that only +/// validates (TTY's lookup/ownership shape) returns +/// [`Establishment::default`]. /// /// `#[non_exhaustive]` so a future carrier change is not another /// breaking release. Construct with [`Establishment::new`] (plan) or -/// [`Establishment::default`] (no plan). +/// [`Establishment::default`] (no plan); chain +/// [`Establishment::with_reply_field`] to contribute reply fields. #[derive(Debug, Clone, Default)] #[non_exhaustive] pub struct Establishment { @@ -389,12 +393,49 @@ pub struct Establishment { /// establisher and the handler agree on the concrete type /// ([`ChannelPlan`]); alkcall never inspects it. pub plan: Option, + pub(crate) reply_fields: Option>, } +/// The wrapper-owned success-reply key (ADR-049 amendment 3): the +/// wrapper reserves `channel_id` — an establisher contributing a reply +/// field with this key is an establisher bug and fails the open loudly +/// (`handler_error`) rather than shadowing the wrapper's value. +pub(crate) const RESERVED_REPLY_KEY: &str = "channel_id"; + impl Establishment { /// A result carrying `plan` to the pump handler. pub fn new(plan: ChannelPlan) -> Self { - Self { plan: Some(plan) } + Self { + plan: Some(plan), + reply_fields: None, + } + } + + /// Contribute one reply field (ADR-049 amendment 3 — the + /// establisher → opener direction). Merged into the open-op + /// success reply by the wrapper, after its `channel_id` + /// reservation. Establisher-supplied `channel_id` is rejected — + /// the check runs at merge time in the wrapper, not here, so the + /// failure carries the wire error path. Builder-style. + pub fn with_reply_field(mut self, key: impl Into, value: Value) -> Self { + self.reply_fields + .get_or_insert_with(Map::new) + .insert(key.into(), value); + self + } + + /// Contribute reply fields from a map (the batch form of + /// [`Establishment::with_reply_field`]). Builder-style. + pub fn with_reply_fields(mut self, fields: Map) -> Self { + self.reply_fields + .get_or_insert_with(Map::new) + .extend(fields); + self + } + + /// The contributed reply fields, for inspection. + pub fn reply_fields(&self) -> Option<&Map> { + self.reply_fields.as_ref() } } @@ -742,6 +783,36 @@ fn establishment_timeout_call_error() -> CallError { })) } +/// The `channel:open_failed` reply for an establisher that supplied +/// the wrapper-owned `channel_id` reply key (ADR-049 amendment 3) — +/// reason `handler_error`, the establisher-internal-failure class. +fn reserved_reply_key_call_error() -> CallError { + CallError::new( + CHANNEL_OPEN_FAILED, + "establisher supplied the reserved reply key `channel_id`", + false, + ) + .with_details(json!({ + "reason": "handler_error", + "message": "establisher supplied the reserved reply key `channel_id`", + })) +} + +/// Merge the establisher's reply fields into the success output +/// (ADR-049 amendment 3): called after the `channel_id` reservation +/// has been checked, so the merge is a flat extend — absent fields +/// leave the reply byte-identical to the pre-amendment shape. +fn merge_reply_fields(mut output: Value, reply_fields: Option>) -> Value { + if let Some(fields) = reply_fields { + if let Some(obj) = output.as_object_mut() { + for (key, value) in fields { + obj.insert(key, value); + } + } + } + output +} + /// Run the open-op wrapper's shared post-ACL steps: `check_open` → /// `open_channel` → **await the establisher bounded** (ADR-049 §1/§2 — /// no-op when no establisher is registered) → build the channel @@ -759,6 +830,12 @@ fn establishment_timeout_call_error() -> CallError { /// opener-side" property: the consumer's open resolves `Err` and no /// `channel_id` was ever returned. /// +/// On establishment success the wrapper merges the establisher's +/// contributed reply fields into the success output (ADR-049 amendment +/// 3) after reserving `channel_id` — an establisher supplying that key +/// fails the open with reason `handler_error`. Absent reply fields +/// leave the reply byte-identical to the pre-amendment shape. +/// /// `input` is the open op's input (params), passed through to the /// establisher (when registered) and the ALPN's `OpenHandler`. /// `call_identity` is the dispatch-resolved caller identity — `Some` @@ -802,6 +879,7 @@ async fn run_open_wrapper( return ResponseEnvelope::error(request_id, map_channel_error_to_call_error(&channel_err)); } + let mut reply_fields: Option> = None; let channel_id = match manager .open_channel(alpn, opener_identity.id.clone(), None) .await @@ -839,7 +917,31 @@ async fn run_open_wrapper( establishment_error_to_call_error(&e), ); } - Ok(Ok(establishment)) => establishment.plan, + Ok(Ok(establishment)) => { + // The `channel_id` reservation (ADR-049 + // amendment 3): the wrapper owns the key. + // An establisher supplying it is a bug — + // fail loudly (same open_failed path) so a + // reserved-key collision never silently + // shadows the wrapper's value. + if establishment + .reply_fields + .as_ref() + .is_some_and(|m| m.contains_key(RESERVED_REPLY_KEY)) + { + tracing::debug!( + channel_id = id, + "open wrapper: establisher supplied the reserved reply key; tearing down channel" + ); + teardown_failed_channel(manager, policy, id, &opener_identity); + return ResponseEnvelope::error( + request_id, + reserved_reply_key_call_error(), + ); + } + reply_fields = establishment.reply_fields; + establishment.plan + } } } None => None, @@ -942,7 +1044,10 @@ async fn run_open_wrapper( } }; - ResponseEnvelope::ok(request_id, json!({ "channel_id": channel_id })) + ResponseEnvelope::ok( + request_id, + merge_reply_fields(json!({ "channel_id": channel_id }), reply_fields), + ) } /// Map a `ChannelError` (from `ChannelLifecyclePolicy::check_open`) to @@ -2378,4 +2483,239 @@ mod tests { let h = plan.downcast_ref::().expect("typed"); assert_eq!(h.0, 7); } + + // --- ADR-049 amendment 3 (review 008 U-2): the reply projection ------- + + #[test] + fn establishment_builder_accumulates_reply_fields() { + let e = Establishment::default() + .with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 })) + .with_reply_field("extra", json!(true)); + let fields = e.reply_fields().expect("fields"); + assert_eq!(fields.len(), 2); + assert_eq!(fields["bound"]["port"], json!(42113)); + assert_eq!(fields["extra"], json!(true)); + + let batch: Map = [("a".to_string(), json!(1)), ("b".to_string(), json!(2))] + .into_iter() + .collect(); + let e = Establishment::new(Arc::new(TestHandle(1)) as ChannelPlan).with_reply_fields(batch); + assert_eq!(e.reply_fields().expect("fields").len(), 2); + assert!(e.plan.is_some(), "the plan rides alongside the fields"); + } + + #[tokio::test] + async fn run_open_wrapper_projects_establisher_reply_field_onto_success_output() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let establishing: OpenEstablisher = Arc::new(|_input, _auth| { + Box::pin(async { + Ok(Establishment::default() + .with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 }))) + }) + }); + let open_handler: OpenHandler = + Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); + let auth = AuthContext::anonymous(b"alk/call"); + let env = run_open_wrapper( + &manager, + &policy, + hook(establishing, None).as_ref(), + &open_handler, + &auth, + "alk/tty", + json!({}), + Some(identity("alice")), + "req-reply-projection".to_string(), + None, + ) + .await; + match env.result { + Ok(v) => { + let channel_id = v["channel_id"].as_u64().expect("channel_id present"); + assert!(channel_id > 0); + assert_eq!( + v["bound"], + json!({ "host": "203.0.113.9", "port": 42113 }), + "the establisher's reply field rides the success output" + ); + assert_eq!( + v.as_object().expect("object").len(), + 2, + "channel_id + bound, nothing else" + ); + } + Err(e) => panic!("projecting establisher should succeed, got: {e:?}"), + } + } + + #[tokio::test] + async fn run_open_wrapper_without_reply_fields_is_byte_identical_to_pre_amendment() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let open_handler: OpenHandler = + Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); + let auth = AuthContext::anonymous(b"alk/call"); + // With an establisher contributing nothing: + let env = run_open_wrapper( + &manager, + &policy, + hook(ok_establisher(), None).as_ref(), + &open_handler, + &auth, + "alk/tty", + json!({}), + Some(identity("alice")), + "req-no-fields-est".to_string(), + None, + ) + .await; + match env.result { + Ok(v) => assert_eq!( + v, + json!({ "channel_id": v["channel_id"] }), + "establisher, no fields: the reply is exactly channel_id" + ), + Err(e) => panic!("open should succeed, got: {e:?}"), + } + // And with no establisher at all: + let env = run_open_wrapper( + &manager, + &policy, + None, + &open_handler, + &auth, + "alk/tty", + json!({}), + Some(identity("alice")), + "req-no-fields-none".to_string(), + None, + ) + .await; + match env.result { + Ok(v) => assert_eq!( + v, + json!({ "channel_id": v["channel_id"] }), + "no establisher: the reply is exactly channel_id" + ), + Err(e) => panic!("open should succeed, got: {e:?}"), + } + } + + #[tokio::test] + async fn run_open_wrapper_establisher_channel_id_reply_key_fails_loudly() { + let manager = make_manager().await; + let concrete_policy = Arc::new(super::super::policy::PerIdentityChannelPolicy::new(8)); + let policy: Arc = + Arc::clone(&concrete_policy) as Arc; + let spawned = Arc::new(AtomicBool::new(false)); + let spawned_clone = Arc::clone(&spawned); + let open_handler: OpenHandler = Arc::new(move |_input, _plan, _conn, _auth| { + spawned_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let establishing: OpenEstablisher = Arc::new(|_input, _auth| { + Box::pin(async { + Ok(Establishment::default().with_reply_field("channel_id", json!(999_u64))) + }) + }); + let auth = AuthContext::anonymous(b"alk/call"); + let env = run_open_wrapper( + &manager, + &policy, + hook(establishing, None).as_ref(), + &open_handler, + &auth, + "alk/tty", + json!({}), + Some(identity("alice")), + "req-reserved-key".to_string(), + None, + ) + .await; + match env.result { + Err(e) => { + assert_eq!(e.code, "channel:open_failed"); + let details = e.details.expect("details carry the reason"); + assert_eq!(details["reason"], "handler_error"); + assert!( + details["message"] + .as_str() + .expect("message") + .contains("channel_id"), + "the message names the reserved key" + ); + } + Ok(v) => panic!("reserved-key establisher must fail the open, got: {v:?}"), + } + assert!( + manager.channel_ids().is_empty(), + "the just-allocated channel was torn down" + ); + assert_eq!( + concrete_policy.count_for(&identity("alice")), + 0, + "ledger take + policy.on_close restored the cap count" + ); + assert!( + !spawned.load(Ordering::SeqCst), + "pump handler must not spawn on the reserved-key rejection" + ); + } + + #[tokio::test] + async fn register_openable_with_establisher_replies_projected_fields_end_to_end() { + 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::Mutation, + Visibility::External, + json!({}), + json!({ + "type": "object", + "properties": { + "channel_id": { "type": "integer" }, + "bound": { "type": "object" } + } + }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alk/tty")); + let establishing: OpenEstablisher = Arc::new(|_input, _auth| { + Box::pin(async { + Ok(Establishment::default() + .with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 }))) + }) + }); + let open_handler: OpenHandler = + Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); + let registry = OperationRegistry::new(); + core.register_openable_with_establisher( + spec, + Some(establishing), + open_handler, + ®istry, + AuthContext::anonymous(b"alk/call"), + None, + ) + .expect("register"); + let env = registry + .invoke( + "channels/tty/sub", + json!({}), + test_context("reg-projection"), + ) + .await; + match env.result { + Ok(v) => { + assert!(v["channel_id"].as_u64().is_some()); + assert_eq!(v["bound"]["port"], json!(42113)); + } + Err(e) => panic!("projected open should succeed, got: {e:?}"), + } + } }