alkblobs 0.2.2

//! 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.

use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::future::{pending, Future};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;

use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
use alkcall::channels::client::ChannelClient;
use alkcall::channels::operations::{
    ChannelCore, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, PubOpenHandler,
};
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::context::{
    AbortPolicy, CompositionAuthority, OperationContext, ScopedPeerEnv,
};
use alkcall::registry::env::OperationEnv;
use alkcall::registry::registration::{make_handler, Handler, OperationRegistry};
use alkcall::registry::spec::{
    AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility,
};
use alkcall::OperationRegistryBuilder;
use serde_json::Value;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

const DUPLEX_BUFFER: usize = 64 * 1024;
/// The prewired call streams per loopback connection — sized for the
/// error-arm sweep round (the malformed-payload tests call the JSON
/// plane dozens of times over one assembled connection), not just the
/// single-flow shapes.
const CALL_STREAMS: usize = 64;
pub(crate) 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<VecDeque<BiStream>>,
    accept: parking_lot::Mutex<VecDeque<BiStream>>,
    addr: Option<SocketAddr>,
}

#[async_trait::async_trait]
impl BidiStreamSource for LoopbackSource {
    async fn accept_bi(&self) -> Result<BiStream, StreamError> {
        let next = self.accept.lock().pop_front();
        match next {
            Some(stream) => Ok(stream),
            None => pending::<Result<BiStream, StreamError>>().await,
        }
    }

    async fn open_bi(&self) -> Result<BiStream, StreamError> {
        self.open
            .lock()
            .pop_front()
            .ok_or(StreamError::StreamClosed)
    }

    fn remote_addr(&self) -> Option<SocketAddr> {
        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(),
        ),
    )
}

pub(crate) fn identity_named(id: &str) -> Identity {
    Identity {
        id: id.to_owned(),
        scopes: Vec::new(),
        resources: HashMap::new(),
    }
}

pub(crate) fn auth_for(identity: Option<Identity>, 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<BTreeMap<String, Identity>>,
}

impl IdentityProvider for TestIdentityProvider {
    fn resolve_from_fingerprint(&self, _fingerprint: &str) -> Option<Identity> {
        None
    }

    fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
        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<BTreeMap<String, BTreeMap<String, BTreeSet<String>>>>,
}

impl TestOwnership {
    pub(crate) 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<String> {
        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 either the
/// open handler (the spawned pump over the channel's `Connection` —
/// the Sub/Query/Mutation shapes) or the pub open handler (the
/// awaited put handler — the marked-Pub shape, ADR-047 amendment 4).
///
/// A channel-open op's registration bundle ...
#[derive(Clone)]
pub(crate) struct ChannelOpenable {
    spec: OperationSpec,
    establisher: OpenEstablisher,
    open_handler: Option<OpenHandler>,
    pub_handler: Option<PubOpenHandler>,
}

impl ChannelOpenable {
    /// The Sub/Query/Mutation shape's bundle.
    pub(crate) fn pump(
        spec: OperationSpec,
        establisher: OpenEstablisher,
        open_handler: OpenHandler,
    ) -> Self {
        ChannelOpenable {
            spec,
            establisher,
            open_handler: Some(open_handler),
            pub_handler: None,
        }
    }

    /// The marked-Pub shape's bundle (the put path).
    pub(crate) fn put(
        spec: OperationSpec,
        establisher: OpenEstablisher,
        pub_handler: PubOpenHandler,
    ) -> Self {
        ChannelOpenable {
            spec,
            establisher,
            open_handler: None,
            pub_handler: Some(pub_handler),
        }
    }
}

/// A registration bundle the harness's assembly carries: the
/// spec/handler pair plus the registration-side authority/seam
/// declarations the embedder's assembly layer makes (ADR-017/ADR-019
/// — the composition authority the internal ACL checks run under
/// and the scoped composition env the reachability gate reads).
/// The plain `assemble` passes `None`/`None` (the common external
/// pair).
pub(crate) struct OpsHarnessRegistration {
    spec: OperationSpec,
    handler: Handler,
    composition_authority: Option<CompositionAuthority>,
    scoped_env: Option<ScopedPeerEnv>,
}

impl OpsHarnessRegistration {
    pub(crate) fn pair(pair: (OperationSpec, Handler)) -> Self {
        let (spec, handler) = pair;
        OpsHarnessRegistration {
            spec,
            handler,
            composition_authority: None,
            scoped_env: None,
        }
    }

    pub(crate) fn with_composition_authority(self, authority: CompositionAuthority) -> Self {
        OpsHarnessRegistration {
            composition_authority: Some(authority),
            ..self
        }
    }

    pub(crate) fn with_scoped_env(self, scoped_env: ScopedPeerEnv) -> Self {
        OpsHarnessRegistration {
            scoped_env: Some(scoped_env),
            ..self
        }
    }
}

pub(crate) struct OpsHarness {
    registry: Arc<OperationRegistry>,
    identity_provider: Arc<TestIdentityProvider>,
    ownership: Arc<TestOwnership>,
}

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.
    pub(crate) fn assemble(specs: Vec<(OperationSpec, Handler)>) -> OpsHarness {
        Self::assemble_registrations(
            specs
                .into_iter()
                .map(OpsHarnessRegistration::pair)
                .collect(),
        )
    }

    /// The assembly with the authority/seam declarations carried
    /// (the delete tests' internal-authority and composition shapes).
    pub(crate) fn assemble_registrations(entries: Vec<OpsHarnessRegistration>) -> OpsHarness {
        let mut builder = OperationRegistryBuilder::new();
        for entry in entries {
            builder = builder
                .with_local(
                    entry.spec,
                    entry.handler,
                    entry.composition_authority,
                    entry.scoped_env,
                    Capabilities::new(),
                )
                .expect("op registration compiles");
        }
        OpsHarness {
            registry: Arc::new(builder.build()),
            identity_provider: Arc::new(TestIdentityProvider::default()),
            ownership: Arc::new(TestOwnership::default()),
        }
    }

    pub(crate) fn with_token(&self, token: &str, identity: Identity) {
        self.identity_provider
            .tokens
            .lock()
            .insert(token.to_owned(), identity);
    }

    pub(crate) 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.
    pub(crate) 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<dyn IdentityProvider>,
        )
        .with_ownership_provider(Arc::clone(&self.ownership) as Arc<dyn OwnershipProvider>);
        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<dyn IdentityProvider>,
        );
        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`].
    pub(crate) async fn spawn_channel_halves(
        &self,
        openables: Vec<ChannelOpenable>,
        served_identity: Option<Identity>,
    ) -> 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<dyn IdentityProvider>;
            let ownership = Arc::clone(&self.ownership) as Arc<dyn OwnershipProvider>;
            Arc::new(move |manager, channel0_conn, install_auth| {
                let registry = Arc::clone(&registry);
                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 {
                        match (&openable.open_handler, &openable.pub_handler) {
                            (Some(open_handler), _) => {
                                core.register_openable_with_establisher(
                                    openable.spec.clone(),
                                    Some(Arc::clone(&openable.establisher)),
                                    Arc::clone(open_handler),
                                    &fork,
                                    auth.clone(),
                                    None,
                                )
                                .expect("stub openable registration compiles");
                            }
                            (None, Some(pub_handler)) => {
                                core.register_pub_openable(
                                    openable.spec.clone(),
                                    Some(Arc::clone(&openable.establisher)),
                                    Arc::clone(pub_handler),
                                    &fork,
                                    auth.clone(),
                                    None,
                                )
                                .expect("stub openable registration compiles");
                            }
                            (None, None) => {
                                panic!("a channel openable carries a handler");
                            }
                        }
                    }
                    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.
pub(crate) 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.
pub(crate) 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 composition parent for the delete tests: an External stub
/// op whose handler composes `blobs/delete` through the operation
/// env — the internal authority context's entry (the composed call
/// runs `internal: true`, ACL-checked against `blobs/delete`'s
/// declared composition authority — the switch, never the skip;
/// ADR-017).
pub(crate) fn composer_pair() -> (OperationSpec, Handler) {
    (
        OperationSpec::new(
            "blobs/compose-delete",
            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 {
            let composed = context
                .env
                .invoke_with_policy("blobs", "delete", input, &context, context.abort_policy)
                .await;
            match composed.result {
                Ok(output) => ResponseEnvelope::ok(context.request_id, output),
                Err(error) => ResponseEnvelope::error(context.request_id, error),
            }
        }),
    )
}

/// A bare `OperationContext` for direct handler invocation (the
/// handler-level ingress/arity tests — the shapes the wire's
/// schema gate would reject first): no identity, no env dispatch.
pub(crate) fn bare_context() -> OperationContext {
    struct NoCompositionEnv;
    #[async_trait::async_trait]
    impl OperationEnv for NoCompositionEnv {
        async fn invoke_with_policy(
            &self,
            namespace: &str,
            operation: &str,
            _input: Value,
            _parent: &OperationContext,
            _policy: AbortPolicy,
        ) -> ResponseEnvelope {
            ResponseEnvelope::not_found("test", &format!("{namespace}/{operation}"))
        }
    }
    OperationContext {
        request_id: "test-request".to_owned(),
        parent_request_id: None,
        identity: None,
        handler_identity: None,
        forwarded_for: None,
        capabilities: Capabilities::new(),
        metadata: std::collections::HashMap::new(),
        scoped_env: ScopedPeerEnv::empty(),
        env: Arc::new(NoCompositionEnv),
        abort_policy: alkcall::registry::context::AbortPolicy::default(),
        deadline: None,
        internal: false,
        ownership: None,
    }
}

/// 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.
pub(crate) fn stub_channel_openable(
    establishments: Arc<parking_lot::Mutex<Vec<String>>>,
    received: Arc<parking_lot::Mutex<Vec<Vec<u8>>>>,
) -> 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: Some(open_handler),
        pub_handler: None,
    }
}

pub(crate) async fn await_call(fut: impl Future<Output = ResponseEnvelope>) -> 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"
    );
}