diff --git a/Cargo.toml b/Cargo.toml index fd6bf21..8bfb134 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,4 +42,5 @@ sha2 = "0.10" tokio-postgres = { version = "0.7", optional = true } [dev-dependencies] +async-trait = "0.1" tokio = { version = "1", default-features = false, features = ["rt-multi-thread"] } diff --git a/src/ops.rs b/src/ops.rs index d57569e..a5c58cb 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -1294,3 +1294,720 @@ mod tests { ); } } + +/// The in-process alkcall harness (ops-surface.md §Why alkcall — the +/// embedder's dial-and-hand-over shape, realized server-less): the +/// serving assembly (`OperationRegistryBuilder` receiving the ops +/// family's spec/handler pairs → the dispatch serving stack → the +/// dispatch run loop over one duplex end) plus the client side +/// ([`alkcall::client::CallClient`] + `spawn_dispatch`, and +/// [`alkcall::channels::client::ChannelClient`] for the binary ops' +/// `channel_open` path) and the test identity/ACL wiring the handler +/// tasks exercise their `AccessControl` gates against — namespace → +/// identities → granted actions, closed by default (OQ-09). +/// +/// The stub ops registered through this harness (echo, namespace +/// probe, channel open) live only in this test tree; nothing here +/// enters the shipped module surface. +#[cfg(test)] +mod harness { + use super::*; + + use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque}; + use std::future::{pending, Future}; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + use std::time::Duration; + + use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; + use alkcall::channels::client::ChannelClient; + use alkcall::channels::operations::{ + ChannelCore, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, + }; + use alkcall::channels::policy::NoCap; + use alkcall::client::CallClient; + use alkcall::core::auth::{AuthContext, AuthToken, Identity, IdentityProvider}; + use alkcall::core::ownership::OwnershipProvider; + use alkcall::core::types::{ + BiStream, BidiStreamSource, Capabilities, Connection, ProtocolHandler, StreamError, + }; + use alkcall::protocol::adapter::CallAdapter; + use alkcall::protocol::connection::{split_single_stream, CallConnection}; + use alkcall::protocol::dispatch::Dispatcher; + use alkcall::protocol::wire::ResponseEnvelope; + use alkcall::registry::registration::{make_handler, Handler, OperationRegistry}; + use alkcall::registry::spec::{ + AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility, + }; + use alkcall::OperationRegistryBuilder; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + const DUPLEX_BUFFER: usize = 64 * 1024; + const CALL_STREAMS: usize = 8; + const TEST_TIMEOUT: Duration = Duration::from_secs(15); + + /// The harness's loopback stream source (the ADR-008 extension + /// point): the QUIC-shaped in-memory stand-in — `open_bi` hands out + /// pre-wired stream pairs (the stream-per-request model the + /// top-level `alk/call` protocol runs; a single `from_bidi` half + /// cannot serve it), `accept_bi` at idle parks (an open connection + /// with no inbound streams), drained queues end it. + struct LoopbackSource { + open: parking_lot::Mutex>, + accept: parking_lot::Mutex>, + addr: Option, + } + + #[async_trait::async_trait] + impl BidiStreamSource for LoopbackSource { + async fn accept_bi(&self) -> Result { + let next = self.accept.lock().pop_front(); + match next { + Some(stream) => Ok(stream), + None => pending::>().await, + } + } + + async fn open_bi(&self) -> Result { + self.open + .lock() + .pop_front() + .ok_or(StreamError::StreamClosed) + } + + fn remote_addr(&self) -> Option { + self.addr + } + + fn close(&self, _code: u32, _reason: &str) { + self.open.lock().clear(); + self.accept.lock().clear(); + } + } + + /// One duplex pair per pre-wired call stream: the client's `open_bi` + /// yields end A, the serving side's `accept_bi` yields end B — the + /// wire shape alkcall's own e2e tests run, extended from one stream + /// to many so the harness client can call repeatedly. + fn loopback_call_pair(streams: usize) -> (Connection, Connection) { + let addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 43_21)); + let mut client_open = VecDeque::new(); + let mut server_accept = VecDeque::new(); + for _ in 0..streams { + let (client_end, server_end) = tokio::io::duplex(DUPLEX_BUFFER); + let (client_read, client_write) = tokio::io::split(client_end); + let (server_read, server_write) = tokio::io::split(server_end); + client_open.push_back(BiStream::from_joined(client_read, client_write)); + server_accept.push_back(BiStream::from_joined(server_read, server_write)); + } + ( + Connection::from_source( + LoopbackSource { + open: parking_lot::Mutex::new(client_open), + accept: parking_lot::Mutex::new(VecDeque::new()), + addr, + }, + b"alk/call".to_vec(), + ), + Connection::from_source( + LoopbackSource { + open: parking_lot::Mutex::new(VecDeque::new()), + accept: parking_lot::Mutex::new(server_accept), + addr, + }, + b"alk/call".to_vec(), + ), + ) + } + + fn identity_named(id: &str) -> Identity { + Identity { + id: id.to_owned(), + scopes: Vec::new(), + resources: HashMap::new(), + } + } + + fn auth_for(identity: Option, alpn: &[u8]) -> AuthContext { + AuthContext { + identity, + alpn: alpn.to_vec(), + remote_addr: None, + tls_client_fingerprint: None, + } + } + + /// The test identity provider: token → identity (the alkcall + /// identity seam's token-resolution half; the fingerprint half + /// resolves nothing here). + #[derive(Default)] + struct TestIdentityProvider { + tokens: parking_lot::Mutex>, + } + + impl IdentityProvider for TestIdentityProvider { + fn resolve_from_fingerprint(&self, _fingerprint: &str) -> Option { + None + } + + fn resolve_from_token(&self, token: &AuthToken) -> Option { + self.tokens + .lock() + .get(String::from_utf8_lossy(&token.raw).as_ref()) + .cloned() + } + } + + /// The test ownership provider — the embedder-registry stand-in: + /// namespace → identities → granted actions, **closed by default** + /// (OQ-09): a namespace with no grants denies every unlisted caller. + #[derive(Default)] + struct TestOwnership { + grants: parking_lot::Mutex>>>, + } + + impl TestOwnership { + fn grant(&self, identity_id: &str, namespace: &str, actions: &[&str]) { + let mut grants = self.grants.lock(); + let namespaces = grants.entry(identity_id.to_owned()).or_default(); + let granted_actions = namespaces.entry(namespace.to_owned()).or_default(); + for action in actions { + granted_actions.insert((*action).to_owned()); + } + } + } + + impl OwnershipProvider for TestOwnership { + fn owns( + &self, + identity: &Identity, + _resource_type: &str, + resource_id: &str, + action: &str, + ) -> bool { + self.grants + .lock() + .get(&identity.id) + .and_then(|namespaces| namespaces.get(resource_id)) + .is_some_and(|actions| actions.contains(action)) + } + + fn owned_resources(&self, identity: &Identity, _resource_type: &str) -> Vec { + self.grants + .lock() + .get(&identity.id) + .map(|namespaces| namespaces.keys().cloned().collect()) + .unwrap_or_default() + } + + fn owns_any(&self, identity: &Identity, _resource_type: &str) -> bool { + self.grants + .lock() + .get(&identity.id) + .is_some_and(|namespaces| !namespaces.is_empty()) + } + } + + /// A channel-open op's registration bundle (ADR-047 §3 as amended by + /// ADR-049): the spec (carrying the `channel_open` marker), the + /// establisher (the awaited Establishment phase), and the open + /// handler (the spawned pump over the channel's `Connection`). + #[derive(Clone)] + struct ChannelOpenable { + spec: OperationSpec, + establisher: OpenEstablisher, + open_handler: OpenHandler, + } + + struct OpsHarness { + registry: Arc, + identity_provider: Arc, + ownership: Arc, + } + + impl OpsHarness { + /// Assemble over the ops family's spec/handler pairs — the + /// `OperationRegistryBuilder`-mediated registration the embedder + /// assembly runs; the harness serves whatever is registered and + /// nothing else. + fn assemble(specs: Vec<(OperationSpec, Handler)>) -> OpsHarness { + let mut builder = OperationRegistryBuilder::new(); + for (spec, handler) in specs { + builder = builder + .with_local(spec, handler, None, None, Capabilities::new()) + .expect("stub op registration compiles"); + } + OpsHarness { + registry: Arc::new(builder.build()), + identity_provider: Arc::new(TestIdentityProvider::default()), + ownership: Arc::new(TestOwnership::default()), + } + } + + fn with_token(&self, token: &str, identity: Identity) { + self.identity_provider + .tokens + .lock() + .insert(token.to_owned(), identity); + } + + fn grant(&self, identity_id: &str, namespace: &str, actions: &[&str]) { + self.ownership.grant(identity_id, namespace, actions); + } + + /// The call shape's halves over one loopback connection: the + /// serving side runs [`CallAdapter`]'s dispatch run loop + /// (the identity + ownership provider wired); the client side + /// runs [`CallClient::spawn_dispatch`] over the other end. + fn spawn_call_halves(&self) -> CallConnection { + let (client_conn, server_conn) = loopback_call_pair(CALL_STREAMS); + let adapter = CallAdapter::new( + Arc::clone(&self.registry), + Arc::clone(&self.identity_provider) as Arc, + ) + .with_ownership_provider(Arc::clone(&self.ownership) as Arc); + tokio::spawn(async move { + let _ = + ProtocolHandler::handle(&adapter, server_conn, &auth_for(None, b"alk/call")) + .await; + }); + let client = CallClient::new( + Arc::clone(&self.registry), + Arc::clone(&self.identity_provider) as Arc, + ); + client.spawn_dispatch(client_conn) + } + + /// The channel shape's halves over one duplex pair: the serving + /// side installs channel 0 through a [`ChannelsAdapter`], forks + /// the assembled registry (ADR-047 §4's per-connection seam), + /// registers the openables through + /// `register_openable_with_establisher`, and dispatches over the + /// fork — the transport-authenticated identity + /// [`Connection::set_identity`] on channel 0 when given. The + /// client side is [`ChannelClient::from_connection`]. + async fn spawn_channel_halves( + &self, + openables: Vec, + served_identity: Option, + ) -> ChannelClient { + let (client_end, server_end) = tokio::io::duplex(DUPLEX_BUFFER); + let server_conn = Connection::from_bidi(server_end, b"alk/channels".to_vec(), None); + let install: InstallChannelZero = { + let registry = Arc::clone(&self.registry); + let provider = Arc::clone(&self.identity_provider) as Arc; + let ownership = Arc::clone(&self.ownership) as Arc; + Arc::new(move |manager, channel0_conn, install_auth| { + let registry = Arc::clone(®istry); + let openables = openables.clone(); + let provider = Arc::clone(&provider); + let ownership = Arc::clone(&ownership); + let auth = install_auth; + tokio::spawn(async move { + if let Some(identity) = auth.identity.clone() { + let _ = channel0_conn.set_identity(identity); + } + let channel0_bidi = match channel0_conn.accept_bi().await { + Ok(bidi) => bidi, + Err(_) => return, + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let fork = registry.fork(); + let core = ChannelCore::new(manager, Arc::new(NoCap)); + for openable in &openables { + core.register_openable_with_establisher( + openable.spec.clone(), + Some(Arc::clone(&openable.establisher)), + Arc::clone(&openable.open_handler), + &fork, + auth.clone(), + None, + ) + .expect("stub openable registration compiles"); + } + let dispatcher = Dispatcher::new(Arc::new(fork), provider) + .with_ownership_provider(ownership); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + dispatcher + .run_loop_single_stream(call_connection, reader, writer) + .await; + }) + }) + }; + let adapter = ChannelsAdapter::new(install, Arc::new(NoCap)); + let install_auth = auth_for(served_identity, b"alk/channels"); + tokio::spawn(async move { + let _ = ProtocolHandler::handle(&adapter, server_conn, &install_auth).await; + }); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), None); + ChannelClient::from_connection(client_conn) + .await + .expect("channel client establishes") + } + } + + /// The stub JSON op — no ACL, echoes the input back. The trivial + /// shape every real call op's envelope ride is proven against. + fn stub_echo() -> (OperationSpec, Handler) { + ( + OperationSpec::new( + "blobs/stub-echo", + OperationType::Query, + Visibility::External, + serde_json::json!({"type": "object"}), + serde_json::json!({"type": "object"}), + Vec::new(), + AccessControl::default(), + None, + ), + make_handler(|input, context| async move { + ResponseEnvelope::ok(context.request_id, input) + }), + ) + } + + /// The stub ACL-carrying op — the ops surface's namespace gate in + /// miniature (ops-surface.md §ACL mapping): resource + /// `blob-namespace`, action `read`, resource id selected from the + /// payload's `namespace` field; the ACL verdict is the alkcall + /// dispatch's, evaluated against the harness's ownership wiring. + fn stub_namespace_probe() -> (OperationSpec, Handler) { + ( + OperationSpec::new( + "blobs/stub-ns-probe", + OperationType::Query, + Visibility::External, + serde_json::json!({ + "type": "object", + "properties": { "namespace": { "type": "string" } }, + "required": ["namespace"] + }), + serde_json::json!({"type": "object"}), + Vec::new(), + AccessControl { + resource_type: Some("blob-namespace".to_owned()), + resource_action: Some("read".to_owned()), + ..AccessControl::default() + }, + Some("$.namespace".to_owned()), + ), + make_handler(|input, context| async move { + ResponseEnvelope::ok(context.request_id, input) + }), + ) + } + + /// The stub channel-open op (the fetch shape: `Sub`, ADR-021/047): + /// `channels/stub/sub` over the `alk/stub` data-plane ALPN. The + /// establisher is the awaited Establishment phase — it records its + /// invocation (so the tests prove it ran ahead of the open reply) + /// and admits `stub-ns` only (`UnknownResource` otherwise, ADR-049 + /// §3's open-failed vocabulary), returning [`Establishment::default`] + /// (validation-only, no plan). The pump reads one client→server + /// chunk, records it, writes `b"pong"` back, and ends — the R-02 + /// duty (the handler task tracks the data plane) in one act. + fn stub_channel_openable( + establishments: Arc>>, + received: Arc>>>, + ) -> ChannelOpenable { + let establisher: OpenEstablisher = { + let establishments = Arc::clone(&establishments); + Arc::new(move |input, _auth| { + let namespace = input + .get("namespace") + .and_then(|value| value.as_str()) + .unwrap_or_default() + .to_owned(); + let establishments = Arc::clone(&establishments); + Box::pin(async move { + establishments.lock().push(namespace.clone()); + if namespace == "stub-ns" { + Ok(Establishment::default()) + } else { + Err(EstablishmentError::UnknownResource { + message: format!("no serving source for namespace {namespace}"), + }) + } + }) + }) + }; + let open_handler: OpenHandler = { + let received = Arc::clone(&received); + Arc::new(move |_input, _plan, channel_conn, _auth| { + let received = Arc::clone(&received); + tokio::spawn(async move { + let mut stream = match channel_conn.accept_bi().await { + Ok(stream) => stream, + Err(_) => return, + }; + let mut chunk = [0u8; 4]; + if stream.read_exact(&mut chunk).await.is_err() { + return; + } + received.lock().push(chunk.to_vec()); + let _ = stream.write_all(b"pong").await; + }) + }) + }; + ChannelOpenable { + spec: OperationSpec::new( + "channels/stub/sub", + OperationType::Sub, + Visibility::External, + serde_json::json!({ + "type": "object", + "properties": { "namespace": { "type": "string" } }, + "required": ["namespace"] + }), + serde_json::json!({ + "type": "object", + "properties": { "channel_id": { "type": "integer" } } + }), + Vec::new(), + AccessControl::default(), + Some("$.namespace".to_owned()), + ) + .with_channel_open(ChannelOpenSpec::new("alk/stub")), + establisher, + open_handler, + } + } + + async fn await_call(fut: impl Future) -> ResponseEnvelope { + tokio::time::timeout(TEST_TIMEOUT, fut) + .await + .expect("call resolves") + } + + /// The JSON stub op round-trips through the two halves: the + /// dispatch run loop serves, the client calls, the response + /// envelope arrives. + #[tokio::test] + async fn json_stub_op_round_trips_call_to_response_envelope() { + let harness = OpsHarness::assemble(vec![stub_echo()]); + let client = harness.spawn_call_halves(); + let response = await_call(client.call( + "blobs/stub-echo", + serde_json::json!({ + "message": "round-trip" + }), + )) + .await; + assert!( + response.result.is_ok(), + "the stub echo resolves, got {:?}", + response.result + ); + assert_eq!( + response.result.unwrap(), + serde_json::json!({ "message": "round-trip" }) + ); + } + + /// The ACL path end-to-end over the call halves (criterion: the + /// identity/ownership wiring exercisable pre-handler): closed by + /// default (OQ-09) — no token → `authentication required`; an + /// identity with no grant → `not owner`; the granted identity → + /// admitted and the handler runs. + #[tokio::test] + async fn grant_gated_stub_denies_ungranted_identities_and_admits_the_granted_one() { + let harness = OpsHarness::assemble(vec![stub_namespace_probe()]); + harness.with_token("token:alice", identity_named("alice")); + harness.with_token("token:stranger", identity_named("stranger")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness.spawn_call_halves(); + + let payload = |token: Option<&str>| match token { + Some(token) => serde_json::json!({ + "operationId": "blobs/stub-ns-probe", + "input": { "namespace": "ns-1" }, + "auth_token": token, + }), + None => serde_json::json!({ + "operationId": "blobs/stub-ns-probe", + "input": { "namespace": "ns-1" }, + }), + }; + + let anonymous = await_call(client.call_with_payload(payload(None))).await; + match anonymous.result { + Err(error) => { + assert_eq!(error.code, "FORBIDDEN"); + assert_eq!(error.message, "authentication required"); + } + other => panic!("the unauthenticated probe is denied, got {other:?}"), + } + + let stranger = await_call(client.call_with_payload(payload(Some("token:stranger")))).await; + match stranger.result { + Err(error) => { + assert_eq!(error.code, "FORBIDDEN"); + assert_eq!( + error.message, "not owner of resource: blob-namespace/ns-1", + "closed-by-default: a resolved identity without the grant is denied" + ); + } + other => panic!("the ungranted identity is denied, got {other:?}"), + } + + let admitted = await_call(client.call_with_payload(payload(Some("token:alice")))).await; + assert!( + admitted.result.is_ok(), + "the granted identity is admitted, got {:?}", + admitted.result + ); + assert_eq!( + admitted.result.unwrap(), + serde_json::json!({ "namespace": "ns-1" }), + "the handler ran with input intact" + ); + } + + /// The channel path's pipe proven (criterion: the stub + /// `channel_open` op establishes through the awaited Establishment + /// phase and bytes flow both directions): the establisher runs + /// ahead of the open reply, the open op answers `channel_id`, and + /// the client's write crosses to the pump (recorded) and the + /// pump's echo comes back. + #[tokio::test] + async fn channel_stub_op_establishes_through_the_establishment_phase_and_bytes_flow() { + let establishments = Arc::new(parking_lot::Mutex::new(Vec::new())); + let received = Arc::new(parking_lot::Mutex::new(Vec::new())); + let harness = OpsHarness::assemble(vec![stub_echo()]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "stub-ns", &["read"]); + let client = harness + .spawn_channel_halves( + vec![stub_channel_openable( + Arc::clone(&establishments), + Arc::clone(&received), + )], + Some(identity_named("alice")), + ) + .await; + + let (channel_id, mut send, mut recv) = tokio::time::timeout( + TEST_TIMEOUT, + client.open_channel( + "channels/stub/sub", + serde_json::json!({ "namespace": "stub-ns" }), + "alk/stub", + ), + ) + .await + .expect("the open resolves") + .expect("the stub channel establishes"); + assert!(channel_id > 0, "a data channel, not channel 0 itself"); + assert_eq!( + *establishments.lock(), + vec!["stub-ns".to_owned()], + "the Establishment phase ran before the open reply" + ); + + send.write_all(b"ping") + .await + .expect("the client's write crosses"); + let mut back = [0u8; 4]; + tokio::time::timeout(TEST_TIMEOUT, recv.read_exact(&mut back)) + .await + .expect("the echo arrives") + .expect("the echo reads"); + assert_eq!( + &back, b"pong", + "bytes flowed client→server and server→client" + ); + assert_eq!( + *received.lock(), + vec![b"ping".to_vec()], + "the pump received the client's chunk" + ); + } + + /// The channel path's refusals (criterion: the ACL path is + /// exercisable on the binary ops' `channel_open` seam too, and the + /// Establishment phase's failure answers in the open reply): + /// an ungranted identity is denied by the registry's ACL gate + /// before the establisher ever runs (establishments empty); a + /// granted identity against a namespace the establisher refuses + /// answers `channel:open_failed` with reason `unknown_resource` + /// through [`ChannelOpenError`]'s typed shape (ADR-049 §3/§4). + #[tokio::test] + async fn channel_open_refusals_deny_before_establishment_and_answer_open_failed() { + let establishments = Arc::new(parking_lot::Mutex::new(Vec::new())); + let received = Arc::new(parking_lot::Mutex::new(Vec::new())); + let mut spec = stub_channel_openable(Arc::clone(&establishments), Arc::clone(&received)); + spec.spec.access_control = AccessControl { + resource_type: Some("blob-namespace".to_owned()), + resource_action: Some("read".to_owned()), + ..AccessControl::default() + }; + let harness = OpsHarness::assemble(vec![stub_echo()]); + harness.with_token("token:stranger", identity_named("stranger")); + let client = harness + .spawn_channel_halves(vec![spec], Some(identity_named("stranger"))) + .await; + + let denied = tokio::time::timeout( + TEST_TIMEOUT, + client.call_open_op( + "channels/stub/sub", + serde_json::json!({ "namespace": "stub-ns" }), + ), + ) + .await + .expect("the denied open resolves"); + match denied.result { + Err(error) => { + assert_eq!(error.code, "FORBIDDEN", "the ACL gate denies first"); + } + other => panic!("the ungranted opener is denied, got {other:?}"), + } + assert!( + establishments.lock().is_empty(), + "the ACL gate runs before the Establishment phase" + ); + + let second = harness + .spawn_channel_halves( + vec![stub_channel_openable( + Arc::clone(&establishments), + Arc::clone(&received), + )], + Some(identity_named("stranger")), + ) + .await; + let refused = tokio::time::timeout( + TEST_TIMEOUT, + second.open_channel( + "channels/stub/sub", + serde_json::json!({ "namespace": "no-such-ns" }), + "alk/stub", + ), + ) + .await + .expect("the refused open resolves"); + match refused { + Err(error) => { + assert_eq!( + error + .call_error() + .map(|call_error| call_error.code.as_str()), + Some("channel:open_failed"), + "the establisher's refusal" + ); + assert_eq!( + error.establishment_reason(), + Some("unknown_resource"), + "the refusal reason reaches the typed open error" + ); + } + Ok(_) => panic!("the unknown resource was refused, got a channel"), + } + assert!( + received.lock().is_empty(), + "no channel opened; the pump never saw bytes" + ); + } +} diff --git a/tasks/ops-call-harness.md b/tasks/ops-call-harness.md index b391ec5..93b2131 100644 --- a/tasks/ops-call-harness.md +++ b/tasks/ops-call-harness.md @@ -1,7 +1,7 @@ --- id: ops-call-harness name: the in-process alkcall harness — the serving assembly and the client side over one duplex pair -status: pending +status: completed depends_on: [ops-foundation] scope: narrow risk: low @@ -45,17 +45,17 @@ surface is testable server-less. ## Acceptance Criteria -- [ ] The harness lives in the ops module's test tree +- [x] The harness lives in the ops module's test tree (`#[cfg(test)]` under feature `ops`), parameterized over the assembled registry (ops register; the harness serves) -- [ ] Server + client halves over one duplex pair; the dispatch run +- [x] Server + client halves over one duplex pair; the dispatch run loop serves; the client calls and receives envelopes -- [ ] The identity/ownership wiring exercisable: a stub op carrying +- [x] The identity/ownership wiring exercisable: a stub op carrying an `AccessControl` denies an ungranted identity and admits a granted one (the ACL path proven end-to-end pre-handler) -- [ ] The channel path proven: a stub `channel_open` op establishes +- [x] The channel path proven: a stub `channel_open` op establishes (the `Establishment` phase) and bytes flow both directions -- [ ] `cargo test --features ops` green; the default build unaffected +- [x] `cargo test --features ops` green; the default build unaffected (the harness is feature-gated test code); `cargo clippy --all-targets -- -D warnings` and `cargo fmt --check` green on both shapes @@ -78,8 +78,95 @@ surface is testable server-less. ## Notes -> To be filled by implementation agent +> Landed as a `#[cfg(test)] mod harness` appended to +> `src/ops.rs` (the ops module's test tree — feature-gated by +> construction, the default build compiles it out; the shipped module +> surface is untouched). All wiring is alkcall's public API: +> +> - **The loopback connection source** (`LoopbackSource` + `Connection::from_source`, +> the ADR-008 extension point): the QUIC-shaped in-memory stand-in the +> top-level `alk/call` shape needs — `open_bi` hands out pre-wired +> stream pairs (one `tokio::io::duplex` pair per stream, joined via +> the public `BiStream::from_joined`), `open_bi` at exhaustion answers +> `StreamClosed`, `accept_bi` at idle parks (an open connection with +> no inbound streams), `close` drains. This was the one place the +> task text needed a ruling: **`spawn_dispatch` + stream-per-request +> cannot complete a call over a single `from_bidi` duplex end** +> (the accept loop consumes the one stream and `call()`'s `open_bi` +> then fails; alkcall's own e2e shape is the same pre-wired +> `BidiStreamSource` stand-in, `pub(crate)` there — reimplemented +> here as the documented downstream seam, N streams wide so the +> client can call repeatedly). The channel shape uses +> `Connection::from_bidi` halves directly (the channels protocol +> multiplexes internally). +> - **The serving assembly**: `OperationRegistryBuilder` receiving the +> stub ops family's spec/handler pairs (`with_local`) → +> `CallAdapter::new(...).with_ownership_provider(...)` → +> `ProtocolHandler::handle` (the dispatch run loop) over its +> `from_source` half; client side +> `CallClient::new(...).spawn_dispatch(...)` over the other. +> `GatewayDispatch` (alkcall's `gateway`-gated transport-neutral +> spine) was not wired — no gateway front-end rides the duplex; the +> wire-serving stack is the `CallAdapter` half. +> - **The channel assembly**: `ChannelsAdapter::new(install_hook,\ NoCap)` +> — the install hook (the ADR-036 amendment shape): sets the +> transport-authenticated identity on channel 0's `Connection` +> (`Connection::set_identity`, the CF-005 (b) path), forks the +> assembled registry (ADR-047 §4's per-connection registration), and +> registers the stub openables through +> `ChannelCore::register_openable_with_establisher` — the awaited +> **Establishment phase** (establisher records its invocation, +> admits `stub-ns` only, returns `Establishment::default()`; refusal +> → `channel:open_failed`/`unknown_resource`) — then +> `Dispatcher::run_loop_single_stream` over +> `CallConnection::new_single_stream`. Client side: +> `ChannelClient::from_connection`, `open_channel` for the binary +> data plane, `call_open_op` for the refusal arms. +> - **The identity/ACL wiring** (`TestIdentityProvider` — token → +> identity — and `TestOwnership` — namespace → identities → granted +> actions, closed by default): the call shape exercises the +> `auth_token` → `IdentityProvider` route (payload `auth_token` via +> `call_with_payload`, alkcall ADR-017 §7's precedence); the channel +> shape exercises the transport-identity route (identity set on +> channel 0's connection in the install hook). Both resolve through +> the dispatch's `AccessControl::check` with the ownership provider +> in context — the ops surface's namespace-as-resource mapping +> (`blob-namespace`, action `read`, `resource_id_path: "$.namespace"`), +> never a re-implemented check (convention 6). +> - **The stubs**: `blobs/stub-echo` (no ACL, echo), `blobs/stub-ns-probe` +> (the ACL op above), `channels/stub/sub` (the channel_open op, over +> the `alk/stub` data-plane ALPN, whose pump reads one chunk, records +> it, and echoes `b"pong"` — the R-02 duty in one act). All live in +> the test tree; nothing registers into the shipped surface. +> +> One dev-dependency entered: `async-trait = "0.1"` (the +> `BidiStreamSource` impl's annotation; cargo dedupes to alkcall's +> version). The lib dependency tree is unchanged. ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-13 (wave 7). The in-process alkcall harness landed in +> `src/ops.rs`'s test tree: two wire shapes over in-memory connections — +> the call shape (stub JSON ops registered through +> `OperationRegistryBuilder`; serving side `CallAdapter` with the +> identity + ownership providers wired; client side `CallClient` + +> `spawn_dispatch` over the loopback pair's stream-per-request source) +> and the channel shape (stub openable registered per-connection through +> `ChannelCore::register_openable_with_establisher` — the awaited +> Establishment phase; serving side `ChannelsAdapter`'s install hook +> forking the assembled registry and running `run_loop_single_stream`; +> client side `ChannelClient::from_connection` + `open_channel`). +> Proven by the stub ops: the JSON stub round-trips call → response +> envelope; the ACL stub (namespace = `blob-namespace` resource) denies +> anonymous and ungranted identities (closed-by-default, OQ-09) and +> admits the granted one; the channel stub establishes and bytes flow +> both directions (`ping`/`pong`); the refusals arms prove the ACL gate +> runs before the establisher and that an establisher refusal answers +> `channel:open_failed`/`unknown_resource` through the typed +> `ChannelOpenError`. Four harness tests green (lib 88 with `--features +> ops`, 73 default; all conformance/interop columns unchanged). Gates +> green: `cargo test --features ops`, `cargo test` (default, harness +> compiled out), `cargo clippy --all-targets -- -D warnings` on both +> shapes, `cargo fmt --check`, `cargo doc --no-deps` clean. The +> handler tasks (ops-json-ops, ops-fetch, ops-put) now register their +> spec/handler pairs through `OpsHarness` instead of inventing wiring. \ No newline at end of file