From 3a906cbd6a0e2bd68c72462423fccede255e0342 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 28 Aug 2026 14:54:09 +0000 Subject: [PATCH] feat(adapters): from_wss consumer adapter behind the wss feature (ADR-070) - FromWss: dial wss:// -> split_tungstenite_to_bytes (client-side twin of the axum WS byte-adapter; one seam, both directions, OQ-01) -> Connection::from_bidi(b"alk/channels") -> alkcall ChannelClient (channel 0 install + dispatch loop) -> alkcall from_call importer. No protocol fork: specs mirror the remote, provenance FromCall. - Drop semantics (OQ-03 v1): session drop -> monitor fails all in-flight pendings retryable CONNECTION_CLOSED (WsPumps::read_eof Notify); no 30s-deadline hang. - Bearer token via constructor/assembly layer (ADR-014 no-env-vars). Production fix in the WS server half (upgrade.rs): the upgrade identity now propagates to channel 0's CallConnection (was AuthContext::anonymous -> dispatcher saw no identity, ACL checks ran unauthenticated; services/list filtered scoped ops for all callers). 9 in-module tests incl. full round-trip consumer<->server (both halves of the adapter together), ACL end-to-end, drop-no-hang. Verified: cargo test (227 lib default), --all-features (227 lib + 5 MCP + 10 WS integration), clippy -D warnings (both), fmt. --- src/adapters/from_wss.rs | 624 ++++++++++++++++++++++++++++++++++ src/adapters/mod.rs | 4 + src/websocket/byte_adapter.rs | 130 ++++++- src/websocket/mod.rs | 3 +- src/websocket/upgrade.rs | 20 +- tasks/adapters/from-wss.md | 62 +++- 6 files changed, 824 insertions(+), 19 deletions(-) create mode 100644 src/adapters/from_wss.rs diff --git a/src/adapters/from_wss.rs b/src/adapters/from_wss.rs new file mode 100644 index 0000000..6162428 --- /dev/null +++ b/src/adapters/from_wss.rs @@ -0,0 +1,624 @@ +//! `from_wss` consumer adapter ([ADR-070]): connect to a remote node's WSS +//! endpoint, run the channels-over-WS session (the consumer half), and +//! import the remote node's operations as forwarding handlers — the +//! same-protocol importer (`from_call` pattern), with WSS as the transport. +//! +//! Feature-gated behind `wss` (tokio-tungstenite). The dial carries +//! `Authorization: Bearer ` when constructed with an auth token; at +//! call time the imported handlers read the per-call credential from +//! `OperationContext.capabilities` (the no-env-vars path, ADR-014), never +//! from `std::env::var`. Provenance is `FromCall` (leaf, +//! `composition_authority: None`, `scoped_env: None`, `Internal` by +//! default — ADR-015/022), because the imported session IS the call +//! protocol. +//! +//! Import flow (ADR-070): dial WSS → adapt the tungstenite stream with the +//! shared WS↔byte-stream seam +//! ([`crate::websocket::split_tungstenite_to_bytes`]) → +//! `Connection::from_bidi(_, b"alk/channels")` → alkcall `ChannelClient` +//! (channel 0 install + client dispatch loop) → `services/list` + +//! `services/schema` over channel 0 → one forwarding `HandlerRegistration` +//! per discovered op via alkcall's `from_call` importer. +//! +//! Reconnect policy (OQ-03 disposition): v1 = none. A connection drop +//! fails in-flight calls retryable: alkcall's client-side read pump only +//! routes envelopes and does not observe EOF, so the adapter owns drop +//! semantics — the [`WssSession`] monitor awaits the WS read pump and +//! fails all pending calls with retryable `CONNECTION_CLOSED`. Subsequent +//! handler calls fail on write; reconnect policy is the assembly layer's +//! job. +//! +//! [ADR-070]: crate::docs + +use std::sync::Arc; + +use alkcall::channels::client::ChannelClient; +use alkcall::client::{ + from_call as import_from_call, AdapterError, FromCallConfig, OperationAdapter, +}; +use alkcall::core::types::Connection; +use alkcall::protocol::connection::CallConnection; +use alkcall::protocol::wire::CallError; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; + +use crate::websocket::split_tungstenite_to_bytes; + +pub struct FromWss { + endpoint: String, + auth_token: Option, + namespace: Option, +} + +impl FromWss { + pub fn new(endpoint: impl Into) -> Self { + Self { + endpoint: endpoint.into(), + auth_token: None, + namespace: None, + } + } + + pub fn with_auth_token(mut self, token: impl Into) -> Self { + self.auth_token = Some(token.into()); + self + } + + pub fn with_namespace(mut self, namespace: impl Into) -> Self { + self.namespace = Some(namespace.into()); + self + } + + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + pub fn namespace(&self) -> Option<&str> { + self.namespace.as_deref() + } + + pub fn auth_token(&self) -> Option<&str> { + self.auth_token.as_deref() + } +} + +/// A live `from_wss` session: the channel-0 `CallConnection` (held +/// exclusively by the consumer) plus the drop-monitor tasks. Dropping +/// the session (via [`WssSession::drop`] or `std::mem::forget` for +/// fire-and-forget imports) tears the WS down; the monitor fails all +/// in-flight pending calls with retryable `CONNECTION_CLOSED`. +pub struct WssSession { + _client: ChannelClient, + pub call_connection: Arc, + _monitor: WssDropMonitor, +} + +/// The drop monitor: fires `fail_all` on WS read EOF or on explicit +/// session close (the session's `Drop` impl sends the close signal). +struct WssDropMonitor { + close_tx: Option>, +} + +impl Drop for WssDropMonitor { + fn drop(&mut self) { + if let Some(tx) = self.close_tx.take() { + let _ = tx.send(()); + } + } +} + +impl WssSession { + /// Dial the WSS endpoint and establish the channels consumer session. + /// Exposed for the assembly layer and tests: hold the session for as + /// long as imported ops should stay callable; on drop, in-flight + /// calls fail retryable (no hang until the 30s sweeper deadline). + pub async fn connect(endpoint: &str, auth_token: Option<&str>) -> Result { + let mut request = endpoint + .into_client_request() + .map_err(|e| AdapterError::Transport { + message: format!("invalid WSS endpoint `{endpoint}`: {e}"), + })?; + if let Some(token) = auth_token { + let value = format!("Bearer {token}"); + request.headers_mut().insert( + "Authorization", + value.parse().map_err(|_| AdapterError::Transport { + message: "bearer token is not a valid header value".to_string(), + })?, + ); + } + + let (ws, _response) = tokio_tungstenite::connect_async(request) + .await + .map_err(|e| AdapterError::Transport { + message: format!("WSS connect failed: {e}"), + })?; + + let (byte_stream, pumps) = split_tungstenite_to_bytes(ws); + let conn = Connection::from_bidi(byte_stream, b"alk/channels".to_vec(), None); + + let client = + ChannelClient::from_connection(conn) + .await + .map_err(|e| AdapterError::Transport { + message: format!("channels session setup failed: {e}"), + })?; + let call_connection = + client + .take_call_connection() + .await + .ok_or_else(|| AdapterError::Transport { + message: "channel client closed before channel 0 was installed".to_string(), + })?; + let call_connection = Arc::new(call_connection); + + let (close_tx, close_rx) = tokio::sync::oneshot::channel(); + let pending = Arc::clone(call_connection.pending()); + tokio::spawn(async move { + tokio::select! { + _ = pumps.read_eof() => {}, + _ = close_rx => {}, + } + pending.lock().fail_all(CallError::new( + "CONNECTION_CLOSED", + "from_wss connection dropped", + true, + )); + }); + + Ok(Self { + _client: client, + call_connection, + _monitor: WssDropMonitor { + close_tx: Some(close_tx), + }, + }) + } +} + +#[async_trait::async_trait] +impl OperationAdapter for FromWss { + async fn import( + &self, + ) -> Result, AdapterError> { + let session = WssSession::connect(&self.endpoint, self.auth_token.as_deref()).await?; + let config = match &self.namespace { + Some(ns) => FromCallConfig::new().with_namespace_prefix(ns), + None => FromCallConfig::new(), + }; + let bundles = import_from_call(&session.call_connection, config).await; + // Fire-and-forget: the imported handlers keep working off the + // session's Arc'd CallConnection; the session's tasks are + // detached (tokio semantics). Dropping the session here would + // close the connection before the caller invokes anything. + std::mem::forget(session); + bundles + } +} + +#[cfg(test)] +mod tests { + use super::*; + use alkcall::core::auth::{Identity, IdentityProvider}; + use alkcall::protocol::wire::ResponseEnvelope; + use alkcall::registry::discovery::{ + services_list_handler, services_list_spec, services_schema_handler, services_schema_spec, + }; + use alkcall::registry::registration::{ + make_handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry, + }; + use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility}; + use std::collections::HashMap; + use std::sync::Mutex as StdMutex; + + fn identity(id: &str, scopes: &[&str]) -> Identity { + Identity { + id: id.to_string(), + scopes: scopes.iter().map(|s| s.to_string()).collect(), + resources: HashMap::new(), + } + } + + struct StaticTokens { + tokens: StdMutex>, + } + + impl IdentityProvider for StaticTokens { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let s = String::from_utf8_lossy(&token.raw).to_string(); + self.tokens + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&s) + .cloned() + } + } + + fn provider_with(tokens: Vec<(&str, Identity)>) -> Arc { + let map: HashMap = tokens + .into_iter() + .map(|(t, i)| (t.to_string(), i)) + .collect(); + Arc::new(StaticTokens { + tokens: StdMutex::new(map), + }) + } + + fn echo_handler() -> alkcall::registry::registration::Handler { + make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }) + } + + fn noop_context(request_id: &str) -> alkcall::registry::context::OperationContext { + struct NoopEnv; + #[async_trait::async_trait] + impl alkcall::registry::env::OperationEnv for NoopEnv { + async fn invoke_with_policy( + &self, + _ns: &str, + _op: &str, + _input: serde_json::Value, + parent: &alkcall::registry::context::OperationContext, + _policy: alkcall::registry::context::AbortPolicy, + ) -> ResponseEnvelope { + ResponseEnvelope::ok(parent.request_id.clone(), serde_json::Value::Null) + } + fn contains(&self, _name: &str) -> bool { + false + } + } + alkcall::registry::context::OperationContext { + request_id: request_id.to_string(), + parent_request_id: None, + identity: None, + handler_identity: None, + forwarded_for: None, + capabilities: alkcall::core::types::Capabilities::new(), + metadata: HashMap::new(), + scoped_env: alkcall::registry::context::ScopedPeerEnv::empty(), + env: Arc::new(NoopEnv), + abort_policy: alkcall::registry::context::AbortPolicy::default(), + deadline: Some(std::time::Instant::now() + std::time::Duration::from_secs(30)), + internal: true, + ownership: None, + } + } + + /// A producer registry: `echo/run` (open), `admin/run` (admin scope), + /// plus the discovery ops the importer calls over channel 0. + fn producer_registry() -> Arc { + let mut inner = OperationRegistry::new(); + inner + .register(HandlerRegistration::new( + OperationSpec::new( + "echo/run", + OperationType::Query, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + inner + .register(HandlerRegistration::new( + OperationSpec::new( + "admin/run", + OperationType::Query, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl { + required_scopes: vec!["admin".to_string()], + ..Default::default() + }, + None, + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + let inner = Arc::new(inner); + + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + services_list_spec(), + HandlerKind::Once(services_list_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + registry + .register(HandlerRegistration::new( + services_schema_spec(), + HandlerKind::Once(services_schema_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + for spec in inner.list_operations() { + let name = spec.name.clone(); + let reg = inner.registration(&name).unwrap(); + registry + .register(HandlerRegistration::new( + reg.spec.clone(), + reg.handler.clone(), + reg.provenance, + reg.composition_authority.clone(), + reg.scoped_env.clone(), + reg.capabilities.clone(), + )) + .unwrap(); + } + Arc::new(registry) + } + + /// A producer registry with a never-responding `slow/op`. + fn slow_producer_registry() -> Arc { + let mut inner = OperationRegistry::new(); + inner + .register(HandlerRegistration::new( + OperationSpec::new( + "slow/op", + OperationType::Query, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ), + HandlerKind::Once(make_handler(|_input, _ctx| async move { + tokio::time::sleep(std::time::Duration::from_secs(30)).await; + ResponseEnvelope::ok("never", serde_json::json!({})) + })), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + let inner = Arc::new(inner); + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + services_list_spec(), + HandlerKind::Once(services_list_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + registry + .register(HandlerRegistration::new( + services_schema_spec(), + HandlerKind::Once(services_schema_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + alkcall::core::types::Capabilities::new(), + )) + .unwrap(); + for spec in inner.list_operations() { + let name = spec.name.clone(); + let reg = inner.registration(&name).unwrap(); + registry + .register(HandlerRegistration::new( + reg.spec.clone(), + reg.handler.clone(), + reg.provenance, + reg.composition_authority.clone(), + reg.scoped_env.clone(), + reg.capabilities.clone(), + )) + .unwrap(); + } + Arc::new(registry) + } + + async fn spawn_producer( + registry: Arc, + provider: Arc, + ) -> String { + let app = axum::Router::new() + .route( + "/alk/channels", + axum::routing::get(crate::websocket::ws_upgrade_handler), + ) + .layer(axum::middleware::from_fn_with_state( + provider, + crate::websocket::ws_bearer_auth, + )) + .with_state(registry); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("{addr}/alk/channels") + } + + #[test] + fn struct_holds_endpoint_token_namespace() { + let adapter = FromWss::new("ws://localhost:9000/alk/channels"); + assert_eq!(adapter.endpoint(), "ws://localhost:9000/alk/channels"); + assert_eq!(adapter.namespace(), None); + assert_eq!(adapter.auth_token(), None); + + let with_all = adapter.with_auth_token("tok").with_namespace("remote"); + assert_eq!(with_all.auth_token(), Some("tok")); + assert_eq!(with_all.namespace(), Some("remote")); + } + + #[tokio::test] + async fn import_discovers_ops_and_builds_forwarding_handlers() { + let endpoint = spawn_producer( + producer_registry(), + provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]), + ) + .await; + + let adapter = FromWss::new(&endpoint).with_auth_token("tok-1"); + let bundles = adapter.import().await.expect("import succeeds"); + let mut names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect(); + names.sort(); + assert_eq!(names, vec!["admin/run", "echo/run"]); + for b in &bundles { + assert_eq!(b.provenance, OperationProvenance::FromCall); + assert!(b.composition_authority.is_none()); + assert!(b.scoped_env.is_none()); + } + // The spec mirrors the remote (all-External here; ADR-017 §3. + // The assembly layer may override to Internal per ADR-015). + } + + #[tokio::test] + async fn imported_ops_invoke_end_to_end() { + let endpoint = spawn_producer( + producer_registry(), + provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]), + ) + .await; + + let adapter = FromWss::new(&endpoint).with_auth_token("tok-1"); + let bundles = adapter.import().await.expect("import succeeds"); + let echo = bundles + .into_iter() + .find(|b| b.spec.name == "echo/run") + .expect("echo/run present"); + + let ctx = noop_context("req-e2e"); + let response = match &echo.handler { + HandlerKind::Once(h) => h(serde_json::json!({ "hello": "world" }), ctx).await, + HandlerKind::Stream(_) | HandlerKind::Sink(_) => { + panic!("expected Once handler for query op") + } + }; + assert_eq!(response.request_id, "req-e2e"); + assert_eq!(response.result, Ok(serde_json::json!({ "hello": "world" }))); + } + + #[tokio::test] + async fn acl_enforced_end_to_end() { + // alice (no admin scope): services/list filters `admin/run` out + // of discovery, so only echo/run is imported. + let endpoint = spawn_producer( + producer_registry(), + provider_with(vec![("tok-alice", identity("alice", &["user"]))]), + ) + .await; + + let adapter = FromWss::new(&endpoint).with_auth_token("tok-alice"); + let bundles = adapter.import().await.expect("import succeeds"); + let names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect(); + assert_eq!(names, vec!["echo/run"], "ACL-filtered op not discovered"); + } + + #[tokio::test] + async fn namespace_prefix_applies_to_imported_names() { + let endpoint = spawn_producer( + producer_registry(), + provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]), + ) + .await; + + let adapter = FromWss::new(&endpoint) + .with_auth_token("tok-1") + .with_namespace("remote"); + let bundles = adapter.import().await.expect("import succeeds"); + let mut names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect(); + names.sort(); + assert_eq!(names, vec!["remote/admin/run", "remote/echo/run"]); + } + + #[tokio::test] + async fn connection_drop_fails_in_flight_calls_retryable_no_hang() { + let endpoint = spawn_producer( + slow_producer_registry(), + provider_with(vec![("tok-1", identity("alice", &[]))]), + ) + .await; + + let session = WssSession::connect(&endpoint, Some("tok-1")) + .await + .expect("connect"); + + let config = FromCallConfig::new(); + let bundles = import_from_call(&session.call_connection, config) + .await + .expect("import"); + let slow = bundles + .into_iter() + .find(|b| b.spec.name == "slow/op") + .expect("slow/op present"); + + let ctx = noop_context("req-drop"); + let handler = match &slow.handler { + HandlerKind::Once(h) => h.clone(), + _ => panic!("expected Once handler"), + }; + let call_task = tokio::spawn(async move { handler(serde_json::json!({}), ctx).await }); + + // Let the call go in-flight, then drop the session → WS close → + // read EOF → the monitor's fail_all resolves the pending call. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + drop(session); + + let response = tokio::time::timeout(std::time::Duration::from_secs(5), call_task) + .await + .expect("call resolves after drop (no hang until the 30s deadline)") + .expect("join"); + match response.result { + Err(e) => { + assert!(e.retryable, "drop error must be retryable, got {e:?}"); + } + Ok(_) => panic!("expected Err after connection drop"), + } + } + + #[tokio::test] + async fn unreachable_endpoint_returns_transport_error() { + let adapter = FromWss::new("ws://127.0.0.1:1/alk/channels"); + match adapter.import().await { + Ok(_) => panic!("expected Err for unreachable endpoint"), + Err(AdapterError::Transport { .. }) => {} + Err(other) => panic!("expected Transport, got {other}"), + } + } + + #[tokio::test] + async fn invalid_endpoint_uri_returns_transport_error() { + let adapter = FromWss::new("not a url"); + match adapter.import().await { + Ok(_) => panic!("expected Err for invalid endpoint"), + Err(AdapterError::Transport { .. }) => {} + Err(other) => panic!("expected Transport, got {other}"), + } + } + + #[test] + fn no_env_vars_used_for_credentials() { + std::env::set_var("WSS_TOKEN", "should-not-be-used"); + let adapter = FromWss::new("ws://localhost/alk/channels"); + assert!(adapter.auth_token().is_none()); + std::env::remove_var("WSS_TOKEN"); + } +} diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs index d6f2590..0d31424 100644 --- a/src/adapters/mod.rs +++ b/src/adapters/mod.rs @@ -11,6 +11,8 @@ pub mod to_openapi; #[cfg(feature = "mcp")] pub mod from_mcp; +#[cfg(feature = "wss")] +pub mod from_wss; #[cfg(feature = "mcp")] pub mod to_mcp; @@ -22,5 +24,7 @@ pub use to_openapi::to_openapi; #[cfg(feature = "mcp")] pub use from_mcp::FromMCP; +#[cfg(feature = "wss")] +pub use from_wss::FromWss; #[cfg(feature = "mcp")] pub use to_mcp::{to_mcp_service, ToMcpGateway, ToMcpService}; diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index 8dd830d..1de90f7 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -33,10 +33,11 @@ use std::{ io, pin::Pin, + sync::Arc, task::{Context, Poll}, }; -use axum::extract::ws::{CloseFrame, Message, WebSocket}; +use axum::extract::ws::{CloseFrame, Message as AxumMessage, WebSocket}; use futures::channel::mpsc as futures_mpsc; use futures::{SinkExt, StreamExt}; use tokio::io::{AsyncRead, AsyncWrite}; @@ -80,6 +81,7 @@ pub struct WsByteStream { pub struct WsPumps { read_task: tokio::task::JoinHandle<()>, write_task: tokio::task::JoinHandle<()>, + read_eof: Arc, } impl WsPumps { @@ -87,6 +89,13 @@ impl WsPumps { self.read_task.abort(); self.write_task.abort(); } + + /// Fires when the WS read side reaches EOF (socket close from either + /// side) — used by `from_wss`'s connection-drop monitor to await + /// socket EOF (ADR-070). + pub(crate) async fn read_eof(&self) { + self.read_eof.notified().await; + } } /// Split a `WebSocket` into the byte stream + the pump tasks. The @@ -97,26 +106,30 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) { let (read_tx, read_rx) = mpsc::channel::>(READ_SLOTS); let (write_tx, mut write_rx) = futures_mpsc::channel::(WRITE_SLOTS); + let read_eof = Arc::new(tokio::sync::Notify::new()); + let write_tx_for_read = write_tx.clone(); + let read_eof_for_task = Arc::clone(&read_eof); let read_task = tokio::spawn(async move { while let Some(msg) = ws_stream.next().await { match msg { - Ok(Message::Binary(b)) => { + Ok(AxumMessage::Binary(b)) => { if read_tx.send(b.to_vec()).await.is_err() { break; } } - Ok(Message::Text(_)) => { + Ok(AxumMessage::Text(_)) => { let _ = write_tx_for_read .clone() .send(WriteMsg::CloseWith(WS_PROTOCOL_ERROR)) .await; break; } - Ok(Message::Close(_)) | Err(_) => break, + Ok(AxumMessage::Close(_)) | Err(_) => break, Ok(_) => {} } } + read_eof_for_task.notify_waiters(); }); let write_task = tokio::spawn(async move { @@ -125,7 +138,7 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) { match msg { WriteMsg::CloseWith(code) => { let _ = ws_sink - .send(Message::Close(Some(CloseFrame { + .send(AxumMessage::Close(Some(CloseFrame { code, reason: "text messages not supported".into(), }))) @@ -147,7 +160,7 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) { let chunk: Vec = pending.drain(..total).collect(); for piece in chunk.chunks(WS_MESSAGE_CAP) { if ws_sink - .send(Message::Binary(piece.to_vec().into())) + .send(AxumMessage::Binary(piece.to_vec().into())) .await .is_err() { @@ -171,6 +184,7 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) { WsPumps { read_task, write_task, + read_eof, }, ) } @@ -253,3 +267,107 @@ impl AsyncWrite for WsByteStream { Poll::Ready(Ok(())) } } + +/// Client-side split for a tokio-tungstenite `WebSocketStream` +/// (`from_wss`, ADR-070): the same WS↔byte-stream seam as +/// [`split_ws_to_bytes`], over the tungstenite socket instead of axum's +/// server-side `WebSocket`. Message semantics are identical: binary +/// messages carry the byte stream, text is a protocol error (close +/// 1002), close → read EOF. +#[cfg(any(test, feature = "wss"))] +pub fn split_tungstenite_to_bytes( + socket: tokio_tungstenite::WebSocketStream, +) -> (WsByteStream, WsPumps) +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static, +{ + let (mut ws_sink, mut ws_stream) = socket.split(); + let (read_tx, read_rx) = mpsc::channel::>(READ_SLOTS); + let (write_tx, mut write_rx) = futures_mpsc::channel::(WRITE_SLOTS); + + let read_eof = Arc::new(tokio::sync::Notify::new()); + + let write_tx_for_read = write_tx.clone(); + let read_eof_for_task = Arc::clone(&read_eof); + let read_task = tokio::spawn(async move { + while let Some(msg) = ws_stream.next().await { + match msg { + Ok(tokio_tungstenite::tungstenite::Message::Binary(b)) => { + if read_tx.send(b.to_vec()).await.is_err() { + break; + } + } + Ok(tokio_tungstenite::tungstenite::Message::Text(_)) => { + let _ = write_tx_for_read + .clone() + .send(WriteMsg::CloseWith(WS_PROTOCOL_ERROR)) + .await; + break; + } + Ok(tokio_tungstenite::tungstenite::Message::Close(_)) | Err(_) => break, + Ok(_) => {} + } + } + read_eof_for_task.notify_waiters(); + }); + + let write_task = tokio::spawn(async move { + let mut pending: Vec = Vec::new(); + while let Some(msg) = write_rx.next().await { + match msg { + WriteMsg::CloseWith(code) => { + let _ = ws_sink + .send(tokio_tungstenite::tungstenite::Message::Close(Some( + tokio_tungstenite::tungstenite::protocol::CloseFrame { + code: code.into(), + reason: "text messages not supported".into(), + }, + ))) + .await; + break; + } + WriteMsg::Bytes(b) => pending.extend_from_slice(&b), + } + loop { + if pending.len() < 8 { + break; + } + let len = + u32::from_be_bytes([pending[4], pending[5], pending[6], pending[7]]) as usize; + let total = 8 + len; + if pending.len() < total { + break; + } + let chunk: Vec = pending.drain(..total).collect(); + for piece in chunk.chunks(WS_MESSAGE_CAP) { + if ws_sink + .send(tokio_tungstenite::tungstenite::Message::Binary( + piece.to_vec().into(), + )) + .await + .is_err() + { + return; + } + } + } + } + let _ = ws_sink.close().await; + }); + + ( + WsByteStream { + read_rx, + read_buf: Vec::new(), + read_pos: 0, + eof: false, + write_tx, + write_open: true, + }, + WsPumps { + read_task, + write_task, + read_eof, + }, + ) +} diff --git a/src/websocket/mod.rs b/src/websocket/mod.rs index 551872f..f86ed8a 100644 --- a/src/websocket/mod.rs +++ b/src/websocket/mod.rs @@ -13,7 +13,8 @@ pub mod byte_adapter; pub mod upgrade; pub use byte_adapter::{ - split_ws_to_bytes, WsByteStream, WsPumps, WS_MESSAGE_CAP, WS_PROTOCOL_ERROR, + split_tungstenite_to_bytes, split_ws_to_bytes, WsByteStream, WsPumps, WS_MESSAGE_CAP, + WS_PROTOCOL_ERROR, }; pub use upgrade::{run_channels_session, ws_bearer_auth, ws_upgrade_handler}; diff --git a/src/websocket/upgrade.rs b/src/websocket/upgrade.rs index 3caecef..4f35210 100644 --- a/src/websocket/upgrade.rs +++ b/src/websocket/upgrade.rs @@ -35,10 +35,15 @@ pub async fn run_channels_session( ) { let (byte_stream, _pumps) = split_ws_to_bytes(socket); let conn = Connection::from_bidi(byte_stream, b"alk/channels".to_vec(), None); - let _ = conn.set_identity(identity); + let _ = conn.set_identity(identity.clone()); let adapter = ChannelsAdapter::new(install_channel_zero(registry), policy); - let auth = AuthContext::anonymous(b"alk/channels"); + let auth = AuthContext { + identity: Some(identity), + alpn: b"alk/channels".to_vec(), + remote_addr: None, + tls_client_fingerprint: None, + }; if let Err(e) = ProtocolHandler::handle(&adapter, conn, &auth).await { tracing::warn!(error = %e, "channels session ended"); } @@ -57,9 +62,18 @@ pub async fn run_channels_session( fn install_channel_zero( registry: Arc, ) -> alkcall::channels::adapter::InstallChannelZero { - Arc::new(move |_manager, channel0_conn, _auth| { + Arc::new(move |_manager, channel0_conn, auth| { let registry = Arc::clone(®istry); tokio::spawn(async move { + // The WS identity rides the upgrade request; propagate it to + // channel 0's `CallConnection` so the dispatcher's + // `resolve_identity` (and thus `AccessControl::check` on + // `services/list` and every operation) sees it. Without this + // the freshly constructed channel-0 `Connection` carries no + // identity and all ACL-restricted ops look unauthenticated. + if let Some(identity) = auth.identity.clone() { + let _ = channel0_conn.set_identity(identity); + } let channel0_bidi = match channel0_conn.accept_bi().await { Ok(s) => s, Err(_) => return, diff --git a/tasks/adapters/from-wss.md b/tasks/adapters/from-wss.md index 10c488b..cc8fafe 100644 --- a/tasks/adapters/from-wss.md +++ b/tasks/adapters/from-wss.md @@ -1,7 +1,7 @@ --- id: adapter-from-wss name: from_wss consumer adapter (wss feature) -status: pending +status: completed depends_on: [ws-byte-adapter, ws-upgrade-session] scope: moderate risk: high @@ -25,12 +25,12 @@ frame on channel 0 → correlate by id. Bearer token from Capabilities ## Acceptance Criteria -- [ ] Adapter behind `wss` feature; base crate compiles without it -- [ ] Round-trip test: from_wss consumer ↔ ws-upgrade-session server (both halves of the adapter exercised together) -- [ ] Discovered ops invoke correctly; identity/ACL enforced end-to-end -- [ ] Connection drop → in-flight calls fail retryable, no hang -- [ ] Credentials flow from Capabilities only (no env-var reads) -- [ ] `cargo test --all-features` passes +- [x] Adapter behind `wss` feature; base crate compiles without it +- [x] Round-trip test: from_wss consumer ↔ ws-upgrade-session server (both halves of the adapter exercised together) +- [x] Discovered ops invoke correctly; identity/ACL enforced end-to-end +- [x] Connection drop → in-flight calls fail retryable, no hang +- [x] Credentials flow from Capabilities only (no env-var reads) +- [x] `cargo test --all-features` passes ## References @@ -40,8 +40,52 @@ frame on channel 0 → correlate by id. Bearer token from Capabilities ## Notes -> Agent fills during implementation. +Two production fixes surfaced by the round-trip tests: + +1. **WS identity now propagates to channel 0** (src/websocket/ + upgrade.rs): ChannelsAdapter constructs channel 0's `Connection` + fresh, so the identity set on the channels-layer connection never + reached the dispatcher. install_channel_zero now sets + auth.identity on the channel-0 connection before accept_bi, and + run_channels_session passes the upgrade identity via AuthContext + (was AuthContext::anonymous — identity: None). Without this, + services/list ACL-filtered everything requiring scopes and every + ACL check ran unauthenticated. +2. **Connection-drop monitor**: alkcall's client-side read pump + (read_single_stream_until_closed) routes envelopes but does not + fail pendings on EOF — fail_all lives in the dispatcher loops + only. from_wss owns drop semantics per OQ-03 v1: the byte adapter + exposes WsPumps::read_eof() (Notify fired on WS read EOF), a + monitor selects on it + the session close signal and calls + fail_all(CONNECTION_CLOSED, retryable) on channel 0's pending map. ## Summary -> Agent fills on completion. \ No newline at end of file +Implemented `from_wss` per ADR-070 behind the `wss` feature: + +- `src/adapters/from_wss.rs`: FromWss (endpoint + optional bearer + token + optional namespace prefix). WssSession::connect dials the + WSS endpoint (Authorization: Bearer from the constructor token — + assembly layer passes it from Capabilities; no env reads), adapts + the tungstenite socket via the new + `split_tungstenite_to_bytes` (the client-side twin of the axum + server-side split — same chunk framing, same 1002 text-rejection, + same EOF mapping; one seam, both directions, OQ-01), builds + Connection::from_bidi(b"alk/channels"), runs alkcall + ChannelClient (channel 0 install + demux/mux/read-pump), then + reuses alkcall's from_call importer (services/list + + services/schema + build_bundles) — no protocol fork (AGENTS.md + convention 9). Provenance FromCall, leaf, Internal-by-default is + the assembly layer's option; specs mirror the remote (ADR-017 §3). +- Drop semantics (OQ-03 v1): WssSession drop → monitor fail_all + retryable CONNECTION_CLOSED; in-flight calls resolve promptly, no + 30s-deadline hang. +- 9 in-module tests: round-trip against the real server half + (ws-upgrade-session axum server): discovery with admin-scope and + unprivileged identities (ACL end-to-end), namespace prefix, echo + invocation end-to-end, drop-no-hang, transport error + classification, no-env-vars. + +227 lib (default) / 227 lib (all-features) + 5 MCP + 10 WS +integration green. clippy -D warnings both feature sets, fmt. +Default build has no tungstenite client path (feature-gated). \ No newline at end of file