diff --git a/docs/architecture/channel-client.md b/docs/architecture/channel-client.md index dddb45f..fac7bf7 100644 --- a/docs/architecture/channel-client.md +++ b/docs/architecture/channel-client.md @@ -48,56 +48,55 @@ impl ChannelClient { pub async fn from_connection(connection: Connection) -> Result; - /// Open a data channel with the given ALPN and params. Sends - /// `channel/open` on channel 0, waits for the response, and returns - /// the channel. + /// Call a per-ALPN open op (`channels//sub` or + /// `channels//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`. + pub async fn call_open_op(&self, operation_id: &str, input: Value) + -> ResponseEnvelope; + + /// 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. + /// + /// 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`. pub async fn open_channel( &self, + operation_id: &str, + input: Value, alpn: &str, - params: Value, - direction: ChannelDirection, - ) -> Result; + ) -> Result<(u32, MpscSendStream, MpscRecvStream), String>; - /// Subscribe to the peer's resource updates. Returns a stream of - /// resource-set events (ADR-037 channel/resources/subscribe). Each - /// event carries the JSON `output.resources` array from ADR-037's - /// `channel/resources/subscribe` response shape. - pub async fn subscribe_resources(&self) - -> Result, ChannelError>; - - /// The call-protocol connection on channel 0, for invoking channel - /// lifecycle operations and any other call ops the peer exposes. - pub fn call(&self) -> &CallConnection; -} - -pub enum ChannelDirection { - InitiatorToResponder, - ResponderToInitiator, -} - -pub struct Channel { - pub channel_id: u32, - /// The channel's BiStream, accessible via the BidiStreamSource - /// (accept_bi — ADR-038 as amended by ADR-035). - pub source: ChannelBidiStreamSource, -} - -/// One event from `channel/resources/subscribe`. Wraps the JSON `output` -/// object from ADR-037's subscribe response — the `resources` array -/// describing what ALPNs the peer exposes and with what `access` preview. -/// The channels crate maps the JSON to this typed struct; the fields mirror -/// ADR-037's response shape. -pub struct ResourceEvent { - pub resources: Vec, -} - -pub struct ResourceEntry { - pub alpn: String, - pub backends_or_targets: Vec, // ALPN-specific enumeration - pub access: Value, // preview of AccessControl (advisory) + /// 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). + pub async fn take_call_connection(&self) -> Option; } ``` +### Post-047: per-ALPN open ops, not a generic `channel/open` + +ADR-047 dissolved the generic `channel/open` operation into per-ALPN +open ops (`channels//sub`, `channels//pub`). Each ALPN +crate registers its own open op via `ChannelCore::register_openable` +(ADR-047 §3, as amended 2026-08-13 — per-connection registration). +The `ChannelClient` calls these ops by name on channel 0 via +`call_open_op`; `open_channel` wraps `call_open_op` + `adopt_channel`. + +The pre-047 `ChannelDirection`, `Channel { channel_id, source }` struct, +and `subscribe_resources` method are deferred (OQ-40, OQ-41). The +`ResourceEntry.access` preview was dropped by ADR-047 §6 — it is +available via `services/schema` on the op spec. + ## Transport-agnostic by construction `ChannelClient` is the client side of the channels protocol. The channels @@ -131,13 +130,12 @@ an already-established, already-authenticated `Connection`, exactly as ## Bidirectionality preserved The channels protocol is bidirectional — either side can open a channel -(ADR-037 §direction semantics). `ChannelClient::open_channel` supports both -`ChannelDirection::InitiatorToResponder` and -`ChannelDirection::ResponderToInitiator`. The client is not "the client -side" in the request/response sense — it can also receive `channel/open` -requests from the peer (the peer initiates, the client's `ChannelManager` -responds). This mirrors the call protocol's operation overlay (each side -populates what operations they expose). +(ADR-037 §direction semantics). The connection owner (the side that holds +the `ChannelManager`) allocates `channel_id`s (ADR-047 §5). The connect +side calls per-ALPN open ops on channel 0; the accept side allocates the +`channel_id` and responds. Both sides can initiate — the connect side +calls the open op, the accept side could also call open ops on the +connect side's channel 0 (if the connect side registers any). `ChannelClient` is one endpoint of a bidirectional channels connection. The name follows the `CallClient` convention (the side that dialed), not a @@ -167,24 +165,26 @@ All design decisions are documented as ADRs in [decisions/](decisions/). | ADR | Decision | Summary | |-----|----------|---------| -| [080](decisions/080-channelclient.md) | ChannelClient | Client side; transport-agnostic `from_connection` primary; dial lives in `AlknetClient` (ADR-045, resolves OQ-55) | -| [093](decisions/093-channels-pure-channel-multiplexing.md) | channels Pure Channel Multiplexing | No `stream_types` on `open_channel`/`Channel`; handler owns sub-stream multiplexing | +| [043](decisions/043-channelclient.md) | ChannelClient | Client side; transport-agnostic `from_connection` primary; dial lives in `AlknetClient` (ADR-045, resolves OQ-55) | +| [035](decisions/035-channels-pure-channel-multiplexing.md) | channels Pure Channel Multiplexing | No `stream_types` on `open_channel`/`Channel`; handler owns sub-stream multiplexing | ## Open Questions - **OQ-55** (resolved by ADR-045): `AlknetClient` core **dial+TLS seam** — `alknet-client` with three dial methods. `ChannelClient`'s API is transport-agnostic (`from_connection`); the dial is the shared seam. - See [ADR-045](decisions/089-alknetclient-native-dial-seam.md). + See [ADR-045](decisions/045-alknetclient-native-dial-seam.md). ## References - ADR-043: ChannelClient (the decision) - ADR-035: channels pure channel multiplexing (no `stream_types`) -- ADR-037: channel lifecycle operations (`open_channel` sends `channel/open`) +- ADR-037: channel lifecycle operations (`open_channel` sends per-ALPN open ops) - ADR-038: ChannelBidiStreamSource (what `Channel.source` wraps, as amended by ADR-035 — `accept_bi` yields a `BiStream`) - ADR-039: ChannelManager (the shared state `ChannelClient` holds) +- ADR-047: openable ALPNs are operations (per-ALPN open ops dissolve + the generic `channel/open`) - OQ-55: AlknetClient / client establishment extraction - `docs/architecture/crates/call/client-and-adapters.md` — `CallClient` (the shape `ChannelClient` mirrors) \ No newline at end of file diff --git a/docs/architecture/channels-adapter.md b/docs/architecture/channels-adapter.md index fbd369e..0beef46 100644 --- a/docs/architecture/channels-adapter.md +++ b/docs/architecture/channels-adapter.md @@ -139,14 +139,23 @@ This is what makes the channels layer WASM-compatible and transport-agnostic — the `ChannelManager` is pure byte routing with no platform or protocol dependencies. -## The `channel/open` handler +## The `channel/open` handler (post-047: per-ALPN open ops) -The `channel/open` (and `channel/close`, `channel/control`, -`channel/resources/subscribe`) operations are registered on the call -protocol's `OperationRegistry` at registration time. The -`ChannelOperations` constructor takes a `ChannelLifecyclePolicy` -(ADR-041) — the default is `PerIdentityChannelPolicy::new(256)` (a -real per-identity cap, not NoOp): +ADR-047 dissolved the generic `channel/open` operation into per-ALPN +open ops (`channels//sub`, `channels//pub`). Each ALPN +crate registers its own open op via `ChannelCore::register_openable` +(ADR-047 §3, as amended 2026-08-13 — per-connection registration). +The `ChannelCore` wrapper does the channel machinery: ACL check (run by +the registry before the wrapper) → `check_open(identity)` → +`manager.open_channel(alpn, opener)` → spawn the ALPN handler on the +channel's `Connection` → respond with `{ "channel_id": }`. + +The `channel/close`, `channel/control`, and +`channel/resources/subscribe` operations are registered on the call +protocol's `OperationRegistry` at registration time via +`ChannelOperations::register_on`. The `ChannelOperations` constructor +takes a `ChannelLifecyclePolicy` (ADR-041) — the default is +`PerIdentityChannelPolicy::new(256)` (a real per-identity cap, not NoOp): ```rust let policy = Arc::new(PerIdentityChannelPolicy::new(256)); @@ -162,12 +171,10 @@ direct channels constructs one policy and shares it across whatever connections it accepts. See ADR-041 for the policy trait and the default/opt-out variants. -The `channel/open` handler (ADR-037): +The per-ALPN open op wrapper (ADR-047 §3): 1. ACL is already checked by `OperationRegistry::invoke` before this handler runs. -2. Looks up the ALPN in `HandlerRegistry` → `channel:unknown_alpn` if - missing. -3. **Per-identity cap check (ADR-041):** +2. **Per-identity cap check (ADR-041):** `policy.check_open(&op_ctx.identity)?` — deny with `channel:too_many_channels` if the identity is over its cap. The identity is the direct caller (the peer on this channels @@ -175,18 +182,18 @@ The `channel/open` handler (ADR-037): (ADR-026). For the hub-relay path, the spoke sees the hub as the direct caller — the hub's quota on the spoke reflects the aggregate of all relayed channels (ADR-041 §5). -4. Allocates the `channel_id` via `next_id.fetch_add(1, Relaxed)` (DP-1: - server-assigned). The per-connection `max_channels` (ADR-040) is +3. Allocates the `channel_id` via `next_id.fetch_add(1, Relaxed)` (DP-1: + connection-owner-assigned). The per-connection `max_channels` (ADR-040) is checked here too — the per-connection memory bound; if hit, the same `channel:too_many_channels` error is returned (which cap fired first is an implementation detail — ADR-041 §4). -5. Constructs the `ChannelBidiStreamSource` (ADR-038, as amended by +4. Constructs the `ChannelBidiStreamSource` (ADR-038, as amended by ADR-035) — one reassembly buffer, yielding a `BiStream`. -6. Spawns the handler task — `tokio::spawn(handler.handle(conn, &auth))`. +5. Spawns the handler task — `tokio::spawn(handler.handle(conn, &auth))`. Identical to what `TtyAdapter::handle` does today, but on a channels-backed `Connection`. -7. Records the `ChannelState`. -8. Returns the `channel_id`. +6. Records the `ChannelState`. +7. Returns the `channel_id`. The `channel/close` handler (ADR-037) gains a symmetric `policy.on_close(&op_ctx.identity)` call after the drain completes @@ -291,12 +298,11 @@ All design decisions are documented as ADRs in [decisions/](decisions/). | ADR | Decision | Summary | |-----|----------|---------| -| [075](decisions/075-channelsadapter-and-channelmanager.md) | ChannelsAdapter and ChannelManager | The split; the contracts | -| [093](decisions/093-channels-pure-channel-multiplexing.md) | channels Pure Channel Multiplexing | The umbrella decision: 8-byte header, no `stream_type`, one reassembly buffer per channel | -| [076](decisions/076-backpressure-channel-limits-id-reuse.md) | Backpressure, Limits, ID Reuse | Bounded-buffer, 256-channel per-connection memory bound, monotonic IDs (DoS defense reframed by ADR-041) | -| [094](decisions/094-per-identity-channel-cap.md) | Per-Identity Channel Cap | 256 per `PeerId`, enforced via `ChannelLifecyclePolicy` in `channels-call`; per-connection `max_channels` reframed as a memory bound | -| [078](decisions/078-two-pump-shutdown-on-completion.md) | Two-Pump Pattern | Shutdown-on-completion contract | -| [079](decisions/079-hub-relay-translate-not-forward.md) | Hub Relay | Translate channel 0, byte-forward data channels | +| [039](decisions/039-channelsadapter-and-channelmanager.md) | ChannelsAdapter and ChannelManager | The split; the contracts | +| [035](decisions/035-channels-pure-channel-multiplexing.md) | channels Pure Channel Multiplexing | The umbrella decision: 8-byte header, no `stream_type`, one reassembly buffer per channel | +| [040](decisions/040-backpressure-channel-limits-id-reuse.md) | Backpressure, Limits, ID Reuse | Bounded-buffer, 256-channel per-connection memory bound, monotonic IDs (DoS defense reframed by ADR-041) | +| [041](decisions/041-per-identity-channel-cap.md) | Per-Identity Channel Cap | 256 per `PeerId`, enforced via `ChannelLifecyclePolicy` in `channels-call`; per-connection `max_channels` reframed as a memory bound | +| [042](decisions/042-hub-relay-translate-not-forward.md) | Hub Relay | Translate channel 0, byte-forward data channels | ## References diff --git a/docs/architecture/channels-wire.md b/docs/architecture/channels-wire.md index f50c3b1..5665ff0 100644 --- a/docs/architecture/channels-wire.md +++ b/docs/architecture/channels-wire.md @@ -130,6 +130,13 @@ yields repeatedly — each stream is a channel, and the header provides `channel_id` correlation. Same code path, same wire format, same handler experience. See ADR-034 §substrate modes (as amended by ADR-035), ADR-039. +**Implementation status:** the in-line substrate mode is implemented +(single bidi stream, header-demuxed N channels). The QUIC-native +multi-stream substrate (accept remaining bidi streams, read headers off +each) is deferred to the downstream alknet crate (OQ-41). The wire +format and demux loop are correct for both substrates; only the outer +`accept_bi()` loop is missing. + ## Wire-level invariants (REQ-CH-01, 02, 04, 05) The de-risk POC (`docs/research/alknet-channels/poc-summary.md` §Issues @@ -238,9 +245,9 @@ payload — is decided; the *function surface* is not. | Phase | Mechanism | Reference | |-------|-----------|-----------| -| Open | `channel/open` call operation on channel 0; responder allocates `channel_id`, returns it | ADR-037 | +| Open | Per-ALPN open op (`channels//sub` or `channels//pub`) on channel 0; connection owner allocates `channel_id`, returns it | ADR-047 §3, §5 | | Data | chunks with `channel_id` routed to reassembly buffers; handler sees a `BiStream` | this doc, [channels-connection.md](channels-connection.md) | -| Control (out-of-band) | `channel/control` call operation on channel 0 | ADR-037 | +| Control (out-of-band) | `channel/control` call operation on channel 0 (deferred — OQ-39) | ADR-037 | | Close | `channel/close` call operation on channel 0; data chunks flushed before close | ADR-037, REQ-CH-06 | ### REQ-CH-06: exit-chunk-before-close ordering (generalizes ADR-055) @@ -270,8 +277,8 @@ All design decisions are documented as ADRs in [decisions/](decisions/). | ADR | Decision | Summary | |-----|----------|---------| -| [071](decisions/071-channels-wire-format.md) | channels Wire Format | 8-byte chunk header (amended by ADR-035); channels layer has no `stream_type` concept; one-way door | -| [093](decisions/093-channels-pure-channel-multiplexing.md) | channels Pure Channel Multiplexing | The umbrella decision: 8-byte header, no `stream_type`, `into_sub_streams` removed, `BiStream`-only, TTY always 5-byte | +| [034](decisions/034-channels-wire-format.md) | channels Wire Format | 8-byte chunk header (amended by ADR-035); channels layer has no `stream_type` concept; one-way door | +| [035](decisions/035-channels-pure-channel-multiplexing.md) | channels Pure Channel Multiplexing | The umbrella decision: 8-byte header, no `stream_type`, `into_sub_streams` removed, `BiStream`-only, TTY always 5-byte | ## Open Questions @@ -289,14 +296,12 @@ Key questions affecting this doc: header, no `stream_type`) - ADR-035: channels pure channel multiplexing (the umbrella decision that amends ADR-034/074/077) -- ADR-052: alknet-tty wire format (the 5-byte format carried - transparently in the channels payload) - ADR-036: channel 0 pre-negotiated - ADR-037: channel lifecycle operations - ADR-040: backpressure, channel limits, ID reuse +- ADR-047: openable ALPNs are operations (per-ALPN open ops dissolve + the generic `channel/open`) - `docs/research/alknet-channels/poc-summary.md` §POC Target 1, §Issues Surfaced #4-#6 (REQ-CH-01, 02, 04) - `docs/research/stream-unification/findings.md` — the research that - surfaced the 8-byte format decision -- `crates/alknet-tty/src/wire.rs` — the 5-byte format implementation - (carried transparently in the channels payload) \ No newline at end of file + surfaced the 8-byte format decision \ No newline at end of file diff --git a/docs/architecture/open-questions.md b/docs/architecture/open-questions.md index fe360cc..d93ff58 100644 --- a/docs/architecture/open-questions.md +++ b/docs/architecture/open-questions.md @@ -82,6 +82,9 @@ is the load-bearing piece the broker composes on. | OQ-36 | `channel_id` allocation in Pub case | resolved | medium | ADR-047 §5 — "connection owner allocates" (the side that holds the `ChannelManager`); amends "responder allocates" | | OQ-37 | `from_call` relay wrapper for marked ops | open | medium | ADR-047 §1 names it as a consumer (hub) concern; alkcall's `from_call` reconstructs the marker (Gap F resolved) so the consumer can branch on it | | OQ-38 | ALPN→path-segment mapping | resolved | low | ADR-047 §"Negative" — strip the `alknet/` prefix; ALPNs without that prefix use the full ALPN string (rare, two-way-door) | +| OQ-39 | `channel/control` control-handle surface | open | medium | The `channel/control` handler currently returns `channel:control_not_implemented`. The control-handle surface (per-channel control callbacks registered by ALPN crates, routing `message` to the handler's control handle for `channel_id`) is real design work — each ALPN crate needs a way to register a control callback, and the channels layer needs a control-handle registry keyed by `channel_id`. Deferred until an ALPN crate (TTY, tunnel) needs out-of-band control. | +| OQ-40 | `channel/resources/subscribe` live subscription | open | medium | The `channel/resources/subscribe` handler currently returns `channel:resources_not_implemented`. The live subscription aggregated from ALPN-crate resource enumerators (ADR-047 §6) requires each ALPN crate to provide a resource enumerator, and the channels layer to aggregate them into a live `Stream` that emits on any change. The current stub is a one-shot error; the real implementation is deferred until a consumer (hub, dashboard) needs live resource discovery. | +| OQ-41 | QUIC-native multi-stream substrate | open | medium | Only the in-line substrate mode is implemented (single bidi stream, header-demuxed N channels). The QUIC-native multi-stream substrate (accept remaining bidi streams, read headers off each — ADR-034 §substrate modes) is deferred to the downstream alknet crate. The wire format and demux loop are correct for both substrates; only the outer `accept_bi()` loop is missing. The alknet crate owns the QUIC dial/accept loop and is the natural place for the multi-stream accept loop. This crate stays transport-agnostic (no QUIC dependency, WASM-compatible). | ## Core Types diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index a189cd1..0d18e5b 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -1,9 +1,16 @@ //! `ChannelsAdapter` — implements `ProtocolHandler` for //! `alknet/channels` (ADR-039). The accept path: receive one //! `Connection`, install channel 0, then run the demux loop — read -//! 8-byte chunk headers off every bidi stream the transport yields, -//! route each chunk's payload to the matching `channel_id`'s -//! reassembly buffer. +//! 8-byte chunk headers off the bidi stream and route each chunk's +//! payload to the matching `channel_id`'s reassembly buffer. +//! +//! The current implementation handles the in-line substrate mode +//! (single bidi stream, header-demuxed N channels — TCP+TLS, +//! WebTransport session, SSH `direct-tcpip`). The QUIC-native +//! multi-stream substrate (accept remaining bidi streams, read headers +//! off each) is deferred to the downstream alknet crate (OQ-41). +//! The wire format and demux loop are correct for both substrates; +//! only the outer `accept_bi()` loop is missing. //! //! Channel 0 is pre-negotiated as `alknet/call` (ADR-036). The //! `install_channel_zero` hook (ADR-036 amendment — single-stream call @@ -92,12 +99,14 @@ impl ChannelsAdapter { } /// The demux loop — reads 8-byte chunk headers off the bidi stream - /// and routes payloads to the `ChannelManager`. On an in-line - /// transport (TCP+TLS, WebTransport session), `accept_bi()` yields - /// once and the header demuxes N channels from that stream. On - /// QUIC, `accept_bi()` yields repeatedly — each stream carries one - /// logical channel, and the header's `channel_id` correlates it. - /// Same code path, same wire format (ADR-034 §substrate modes). + /// and routes payloads to the `ChannelManager`. This handles the + /// in-line substrate mode (single bidi stream, header-demuxed N + /// channels — TCP+TLS, WebTransport session, SSH `direct-tcpip`). + /// The QUIC-native multi-stream substrate (accept remaining bidi + /// streams, read headers off each) is deferred to the downstream + /// alknet crate (OQ-41). The wire format and demux loop are correct + /// for both substrates; only the outer `accept_bi()` loop is missing. + /// See ADR-034 §substrate modes. async fn run_demux_loop( manager: &ChannelManager, reader: Box, diff --git a/src/channels/operations.rs b/src/channels/operations.rs index 58305b9..ad5e64c 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -232,6 +232,11 @@ fn make_close_handler(manager: ChannelManager, policy: Arc Handler { Arc::new(move |input: Value, ctx: OperationContext| { let manager = manager.clone(); @@ -252,12 +257,14 @@ fn make_control_handler(manager: ChannelManager) -> Handler { CallError::not_found("channel:unknown_channel"), ); } - // The control routing to the handler's control handle is - // ALPN-specific and not implemented in the generic layer — - // the ALPN crate registers a control callback. For now, - // return ok; the ALPN crate overrides this behavior via - // composition. - ResponseEnvelope::ok(ctx.request_id, json!({ "ok": true })) + ResponseEnvelope::error( + ctx.request_id, + CallError::new( + "channel:control_not_implemented", + "channel/control is not yet implemented (OQ-39)", + false, + ), + ) }) }) } @@ -266,25 +273,23 @@ fn make_control_handler(manager: ChannelManager) -> Handler { /// snapshot of the open channels, then subsequent events on any change. /// The `access` preview is dropped (ADR-047 §6 — it's on the op spec, /// available via `services/schema`). +/// +/// Not yet implemented — the live subscription aggregated from +/// ALPN-crate resource enumerators (ADR-047 §6) is deferred (OQ-40). +/// Returns `channel:resources_not_implemented` so callers fail loudly +/// rather than receiving a fake one-shot snapshot. fn make_resources_subscribe_handler(manager: ChannelManager) -> StreamingHandler { Arc::new(move |_input: Value, ctx: OperationContext| { - let manager = manager.clone(); + let _manager = manager.clone(); Box::pin(futures::stream::once(async move { - let mut resources: Vec = Vec::new(); - // The resource set is the set of open channels' ALPNs. - // A real implementation would aggregate across all - // registered openable ALPNs (each ALPN crate provides a - // resource enumerator, ADR-047 §6). For now, emit the - // currently-open channels. Iterating `channel_ids()` is - // O(open channels) — the previous `0..u32::MAX` loop was - // O(max id) and could spin millions of times on sparse ids - // after churn. - for channel_id in manager.channel_ids() { - if let Some(alpn) = manager.channel_alpn(channel_id) { - resources.push(json!({ "alpn": alpn })); - } - } - ResponseEnvelope::ok(ctx.request_id, json!({ "resources": resources })) + ResponseEnvelope::error( + ctx.request_id, + CallError::new( + "channel:resources_not_implemented", + "channel/resources/subscribe is not yet implemented (OQ-40)", + false, + ), + ) })) }) } @@ -789,55 +794,37 @@ mod tests { } #[tokio::test] - async fn resources_subscribe_handler_emits_alpns_for_open_channels() { + async fn resources_subscribe_handler_returns_not_implemented() { let manager = make_manager().await; manager .open_channel("alknet/tty", "alice", None) .await .expect("open tty"); - manager - .open_channel("alknet/tunnel", "alice", None) - .await - .expect("open tunnel"); let handler = make_resources_subscribe_handler(manager.clone()); let mut stream = handler(Value::Null, test_context("res-1")); - let env = stream.next().await.expect("one snapshot envelope"); - let out = env.result.expect("ok"); - let resources = out - .get("resources") - .and_then(|v| v.as_array()) - .expect("resources array"); - let mut alpns: Vec = resources - .iter() - .map(|r| { - r.get("alpn") - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string() - }) - .collect(); - alpns.sort(); - assert_eq!( - alpns, - vec!["alknet/tty".to_string(), "alknet/tunnel".to_string()] - ); + let env = stream.next().await.expect("one envelope"); + match env.result { + Err(e) => assert_eq!( + e.code, "channel:resources_not_implemented", + "resources/subscribe should return not-implemented (OQ-40)" + ), + Ok(_) => panic!("resources/subscribe should return error"), + } } #[tokio::test] - async fn resources_subscribe_handler_empty_when_no_channels_open() { + async fn resources_subscribe_handler_not_implemented_even_when_empty() { let manager = make_manager().await; let handler = make_resources_subscribe_handler(manager); let mut stream = handler(Value::Null, test_context("res-2")); - let env = stream.next().await.expect("snapshot envelope"); - let out = env.result.expect("ok"); - let resources = out - .get("resources") - .and_then(|v| v.as_array()) - .expect("resources array"); - assert!( - resources.is_empty(), - "no channels open → empty resource set" - ); + let env = stream.next().await.expect("one envelope"); + match env.result { + Err(e) => assert_eq!( + e.code, "channel:resources_not_implemented", + "resources/subscribe should return not-implemented even with no channels (OQ-40)" + ), + Ok(_) => panic!("resources/subscribe should return error"), + } } #[tokio::test]