alkhttp/tests/ws_upgrade_session.rs

//! WS upgrade + channels session integration tests (the ws-upgrade-session
//! acceptance gates): tokio-tungstenite client ↔ axum WS server
//! (upgrade → byte-adapter → ChannelsAdapter + channel-0 Dispatcher).
//! Grows from the ws-byte-adapter POC's validated suite.

use alkcall::core::auth::{Identity, IdentityProvider};
use alkcall::protocol::wire::{EventEnvelope, ResponseEnvelope, EVENT_ERROR, EVENT_RESPONDED};
use alkcall::registry::discovery::{services_list_handler, services_list_spec};
use alkcall::registry::registration::{
    make_handler, Handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
};
use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
use alkhttp::websocket::{
    frame_channel0_chunk, ChunkAssembler, FrameAssembler, WsClient, INBOUND_WS_MESSAGE_CAP,
    WS_MESSAGE_CAP,
};
use std::collections::HashMap;
use std::sync::Arc;

fn identity(id: &str, scopes: &[&str]) -> Identity {
    Identity {
        id: id.to_string(),
        scopes: scopes.iter().map(|s| s.to_string()).collect(),
        resources: HashMap::new(),
    }
}

fn echo_handler() -> Handler {
    make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) })
}

fn echo_registry() -> Arc<OperationRegistry> {
    let registry = OperationRegistry::new();
    registry
        .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();
    Arc::new(registry)
}

fn slow_and_echo_registry() -> Arc<OperationRegistry> {
    let registry = OperationRegistry::new();
    registry
        .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();
    registry
        .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();
    Arc::new(registry)
}

fn internal_registry() -> Arc<OperationRegistry> {
    let registry = OperationRegistry::new();
    registry
        .register(HandlerRegistration::new(
            OperationSpec::new(
                "secret/op",
                OperationType::Query,
                Visibility::Internal,
                serde_json::json!({}),
                serde_json::json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            alkcall::core::types::Capabilities::new(),
        ))
        .unwrap();
    Arc::new(registry)
}

fn restricted_registry() -> Arc<OperationRegistry> {
    let registry = OperationRegistry::new();
    registry
        .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();
    Arc::new(registry)
}

fn registry_with_services_list(inner_ops: Vec<HandlerRegistration>) -> Arc<OperationRegistry> {
    let inner = OperationRegistry::new();
    for op in inner_ops {
        inner.register(op).unwrap();
    }
    let inner = Arc::new(inner);
    let 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();
    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)
}

struct StaticTokens {
    tokens: std::sync::Mutex<HashMap<String, Identity>>,
}

impl IdentityProvider for StaticTokens {
    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
        None
    }
    fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
        let s = String::from_utf8_lossy(&token.raw).to_string();
        self.tokens.lock().unwrap().get(&s).cloned()
    }
}

fn provider_with(tokens: Vec<(&str, Identity)>) -> Arc<dyn IdentityProvider> {
    let map: HashMap<String, Identity> = tokens
        .into_iter()
        .map(|(t, i)| (t.to_string(), i))
        .collect();
    Arc::new(StaticTokens {
        tokens: std::sync::Mutex::new(map),
    })
}

async fn spawn_ws_server(
    registry: Arc<OperationRegistry>,
    provider: Arc<dyn IdentityProvider>,
) -> String {
    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            provider,
            alkhttp::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();
    });
    addr
}

async fn call_and_await(
    ws: &mut WsClient,
    request_id: &str,
    operation: &str,
    input: serde_json::Value,
) -> EventEnvelope {
    let frame = EventEnvelope::requested(
        request_id,
        serde_json::json!({ "operationId": operation, "input": input }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;

    let mut chunks = ChunkAssembler::new();
    let mut frames = FrameAssembler::new();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "timed out waiting for response to {request_id}"
        );
        // A prior request's trailing `call.completed` (a Sub pump's
        // natural end) may still be in flight — skip frames for other
        // request ids. Data-channel chunks (id != 0) are skipped too:
        // this helper asserts only channel-0 responses.
        if let Some(env) = frames.next_frame() {
            if env.id == request_id {
                return env;
            }
        }
        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
        match bin {
            Some(bytes) => {
                chunks.push(&bytes);
                while let Some((channel_id, payload)) = chunks.next_chunk() {
                    if channel_id == 0 {
                        frames.push(&payload);
                    }
                }
            }
            None => panic!("ws closed unexpectedly while awaiting {request_id}"),
        }
    }
}

#[tokio::test]
async fn end_to_end_call_round_trip_over_ws() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let env = call_and_await(
        &mut ws,
        "req-1",
        "echo/run",
        serde_json::json!({ "hello": "world" }),
    )
    .await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    assert_eq!(env.id, "req-1");
    assert_eq!(env.payload["output"]["hello"], "world");
    ws.close().await;
}

#[tokio::test]
async fn upgrade_without_token_rejected_401() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let status = WsClient::connect_status(&format!("{addr}/alk/channels"), None).await;
    assert_eq!(status, Some(401));
}

#[tokio::test]
async fn upgrade_with_unresolvable_token_rejected_401() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let status =
        WsClient::connect_status(&format!("{addr}/alk/channels"), Some("wrong-token")).await;
    assert_eq!(status, Some(401));
}

#[tokio::test]
async fn oversized_payload_splits_across_ws_messages_and_round_trips() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let payload_len = 3 * 1024 * 1024;
    let blob = "x".repeat(payload_len);
    let frame = EventEnvelope::requested(
        "req-big",
        serde_json::json!({ "operationId": "echo/run", "input": { "blob": blob } }),
    );
    let chunk = frame_channel0_chunk(&frame);
    assert!(
        chunk.len() > WS_MESSAGE_CAP,
        "test precondition: request exceeds message cap"
    );

    for piece in chunk.chunks(WS_MESSAGE_CAP) {
        ws.send_binary_piece(piece).await;
    }

    let mut chunks = ChunkAssembler::new();
    let mut frames = FrameAssembler::new();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
    loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "timed out on big payload"
        );
        if let Some(env) = frames.next_frame() {
            assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
            assert_eq!(env.id, "req-big");
            let out = env.payload["output"]["blob"].as_str().unwrap();
            assert_eq!(out.len(), payload_len, "blob round-tripped intact");
            assert!(out.bytes().all(|b| b == b'x'), "byte integrity");
            ws.close().await;
            return;
        }
        let bytes = ws
            .next_binary(std::time::Duration::from_secs(2))
            .await
            .expect("ws closed unexpectedly on big payload");
        chunks.push(&bytes);
        while let Some((channel_id, payload)) = chunks.next_chunk() {
            assert_eq!(channel_id, 0);
            frames.push(&payload);
        }
    }
}

#[tokio::test]
async fn interleaved_calls_reassemble_without_corruption() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    const N: usize = 20;
    for i in 0..N {
        let frame = EventEnvelope::requested(
            format!("req-{i}"),
            serde_json::json!({ "operationId": "echo/run", "input": { "i": i } }),
        );
        ws.send_binary(frame_channel0_chunk(&frame)).await;
    }

    let mut chunks = ChunkAssembler::new();
    let mut frames = FrameAssembler::new();
    let mut seen: Vec<String> = Vec::new();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
    while seen.len() < N {
        assert!(
            tokio::time::Instant::now() < deadline,
            "only got {} of {}",
            seen.len(),
            N
        );
        let bytes = ws
            .next_binary(std::time::Duration::from_millis(500))
            .await
            .expect("ws closed unexpectedly");
        chunks.push(&bytes);
        while let Some((channel_id, payload)) = chunks.next_chunk() {
            assert_eq!(channel_id, 0);
            frames.push(&payload);
            while let Some(env) = frames.next_frame() {
                assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
                let i: usize = env.payload["output"]["i"].as_u64().unwrap() as usize;
                assert_eq!(env.id, format!("req-{i}"), "correlation intact");
                seen.push(env.id);
            }
        }
    }
    seen.sort();
    let mut expected: Vec<String> = (0..N).map(|i| format!("req-{i}")).collect();
    expected.sort();
    assert_eq!(seen, expected);
    ws.close().await;
}

#[tokio::test]
async fn disconnect_mid_call_does_not_hang_the_server() {
    let addr = spawn_ws_server(
        slow_and_echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let frame = EventEnvelope::requested(
        "req-abandon",
        serde_json::json!({ "operationId": "slow/op", "input": {} }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;

    // Abrupt client disconnect (drop, no close handshake) while the
    // slow handler is still running.
    drop(ws);

    // The server must not hang: a fresh session completes an echo call
    // promptly — the abandoned session's teardown (pending failure,
    // overlay drop) did not wedge the runtime or the listener.
    let mut ws2 = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();
    let env = call_and_await(
        &mut ws2,
        "req-after",
        "echo/run",
        serde_json::json!({"ok": true}),
    )
    .await;
    assert_eq!(env.r#type, EVENT_RESPONDED);
    assert_eq!(env.payload["output"]["ok"], true);
    ws2.close().await;
}

#[tokio::test]
async fn acl_filtered_call_forbidden_over_ws() {
    let addr = spawn_ws_server(
        restricted_registry(),
        provider_with(vec![("tok-1", identity("alice", &["user"]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let env = call_and_await(&mut ws, "req-403", "admin/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, "call.error", "got {}", env.r#type);
    assert_eq!(env.id, "req-403");
    assert_eq!(env.payload["code"], "FORBIDDEN");
    ws.close().await;
}

#[tokio::test]
async fn text_message_rejected_with_protocol_close() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    ws.send_text("hello as text").await;
    let close = ws
        .next_close(std::time::Duration::from_secs(5))
        .await
        .expect("expected a close frame");
    assert_eq!(close, Some(1002));
}

#[tokio::test]
async fn internal_op_not_callable_from_wire() {
    let addr = spawn_ws_server(
        internal_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let env = call_and_await(&mut ws, "req-secret", "secret/op", serde_json::json!({})).await;
    assert_eq!(env.r#type, "call.error", "got {}", env.r#type);
    assert_eq!(env.id, "req-secret");
    assert_eq!(env.payload["code"], "NOT_FOUND");
    ws.close().await;
}

#[tokio::test]
async fn services_list_over_channel0_is_access_control_filtered() {
    let registry = registry_with_services_list(vec![
        HandlerRegistration::new(
            OperationSpec::new(
                "public/echo",
                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(),
        ),
        HandlerRegistration::new(
            OperationSpec::new(
                "admin/secret",
                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(),
        ),
    ]);
    let addr = spawn_ws_server(
        registry,
        provider_with(vec![("tok-1", identity("regular", &["user"]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let env = call_and_await(&mut ws, "req-list", "services/list", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    assert_eq!(env.id, "req-list");
    let ops = env.payload["output"]["operations"]
        .as_array()
        .expect("operations array");
    let names: Vec<&str> = ops.iter().filter_map(|o| o["name"].as_str()).collect();
    assert!(names.contains(&"public/echo"), "got: {names:?}");
    assert!(!names.contains(&"admin/secret"), "got: {names:?}");
    ws.close().await;
}

#[tokio::test]
async fn inbound_message_over_cap_fails_the_connection() {
    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    // One WS binary message larger than INBOUND_WS_MESSAGE_CAP: the
    // server's explicit max_message_size (WS-06) must fail the
    // connection instead of accepting it (the byte-adapter boundary is
    // the chunk header, so the adapter never dissects messages; the WS
    // library cap is the enforcement point).
    let oversized = vec![b'x'; INBOUND_WS_MESSAGE_CAP + 1];
    ws.send_binary_piece(&oversized).await;

    let close = ws
        .next_close(std::time::Duration::from_secs(5))
        .await
        .expect("server must fail the oversized-inbound connection (WS-06)");
    assert!(
        close.is_some(),
        "expected a close or stream end, got {close:?}"
    );
}

/// WS-08 acceptance: the upgrade retains each session's pump handle in
/// the shared `WsSessions` registry; `abort()` evicts a stuck session
/// (the peer's socket closes) and the entry is removed when the
/// session task ends, so the registry only tracks live sessions.
#[tokio::test]
async fn ws_sessions_registry_tracks_and_aborts_live_sessions() {
    use alkhttp::websocket::WsSessions;

    let sessions = WsSessions::new();
    let registry = echo_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &[]))]);

    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            Arc::clone(&provider),
            alkhttp::websocket::ws_bearer_auth,
        ))
        .with_state(registry)
        .layer(axum::Extension(sessions.clone()));
    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();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();
    let env = call_and_await(&mut ws, "req-1", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "session established");

    // The upgrade registered the session.
    let saw_session = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            if !sessions.is_empty() {
                return true;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    })
    .await
    .expect("session registered in the shared registry");
    assert!(saw_session);
    assert_eq!(sessions.len(), 1);

    // Forced teardown: abort() closes the peer's socket.
    sessions.abort();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
    let close = loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "abort never reached the peer"
        );
        match ws.next_close(std::time::Duration::from_millis(250)).await {
            Some(x) => break Some(x),
            None => continue,
        }
    };
    assert!(close.is_some(), "peer observes the teardown: {close:?}");

    // The self-removing guard drops the entry once the session task ends.
    let drained = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            if sessions.is_empty() {
                return true;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    })
    .await
    .expect("session entry removed after the session task ends");
    assert!(drained);
    ws.close().await;
}

/// WS-26/WS-28 acceptance: the channel-0 `CallConnection` handle is
/// visible in the shared `WsSessions` registry while the session is
/// live (`live_connections()` non-empty mid-session, after a completed
/// call proves the dispatcher task is up) and drains when the session
/// ends — the guard lives for the channel-0 task, not the `if let`
/// block that inserts it. This is the only gate exercising the WS-26
/// surface; its absence is why `030c5ef` landed green (review 007
/// WS-28, reproduced empirically).
#[tokio::test]
async fn live_connections_visible_mid_session_and_drain_after_teardown() {
    use alkhttp::websocket::WsSessions;

    let sessions = WsSessions::new();
    let registry = echo_registry();

    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            provider_with(vec![("tok-1", identity("alice", &[]))]),
            alkhttp::websocket::ws_bearer_auth,
        ))
        .with_state(registry)
        .layer(axum::Extension(sessions.clone()));
    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();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();
    let env = call_and_await(&mut ws, "req-1", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "session established");

    // The handle is retained for the channel-0 task's lifetime: visible
    // mid-session (the dispatcher loop is fully up — the call returned).
    let saw_handle = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            if sessions.live_connection_count() == 1 {
                return true;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    })
    .await
    .expect("channel-0 connection handle visible mid-session (WS-28)");
    assert!(saw_handle);
    assert_eq!(sessions.live_connection_count(), 1);
    let handles = sessions.live_connections();
    assert_eq!(handles.len(), 1, "one handle for the one session");

    // Teardown: the guard's drop removes the handle on any end path.
    ws.close().await;
    let drained = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            if sessions.live_connection_count() == 0 {
                return true;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    })
    .await
    .expect("channel-0 connection handle removed after the session ends");
    assert!(drained);
}

/// WS-09 acceptance: the session cap is enforced on the upgrade — a
/// caller over the configured cap is rejected with 503, and an ended
/// session frees its slot for the next caller.
#[tokio::test]
async fn session_cap_rejects_over_limit_with_503_and_frees_slots_on_end() {
    use alkhttp::server::HttpAdapter;

    let registry_val = OperationRegistry::new();
    registry_val
        .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();
    let registry = Arc::new(registry_val);
    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            (s == "tok-1").then(|| identity("alice", &[]))
        }
    }

    // The adapter's built-in surface: WS route + bearer middleware, the
    // whole deployment shape the cap knob configures.
    let adapter =
        HttpAdapter::new(Arc::new(StaticTok), Arc::clone(&registry)).with_ws_max_sessions(1);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = format!("ws://{}", listener.local_addr().unwrap());
    let app: axum::Router = adapter.router().clone();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let url = format!("{addr}/alk/channels");

    // First session: admitted.
    let mut first = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
    let env = call_and_await(&mut first, "req-1", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "first session admitted");

    // Second session over the cap: rejected with 503 at the upgrade.
    let status = WsClient::connect_status(&url, Some("tok-1")).await;
    assert_eq!(status, Some(503), "second session rejected over the cap");

    // Ending the first session frees the slot: the next caller is admitted.
    first.close().await;
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let mut third = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
    let env = call_and_await(&mut third, "req-2", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "slot freed after session end");
    third.close().await;
}

/// WS-13 acceptance on the axum path (the integration mirror; also the
/// COV-11b dark-knob gate for `with_ws_idle_timeout`): a client
/// dribbling a declared chunk's payload one byte per message, each gap
/// inside the idle window, is evicted with 1001 — message arrival does
/// not reset the deadline, only a completed chunk would.
#[tokio::test]
async fn idle_progress_knob_evicts_forever_dribble_over_axum_upgrade() {
    use alkhttp::server::HttpAdapter;

    let registry_val = OperationRegistry::new();
    let registry = Arc::new(registry_val);
    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            (s == "tok-1").then(|| identity("alice", &[]))
        }
    }
    let adapter = HttpAdapter::new(Arc::new(StaticTok), Arc::clone(&registry))
        .with_ws_idle_timeout(Some(std::time::Duration::from_millis(150)));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = format!("ws://{}", listener.local_addr().unwrap());
    let app: axum::Router = adapter.router().clone();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let dribbler = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();
    let dribble = tokio::spawn(async move {
        let mut dribbler = dribbler;
        let mut header = vec![0u8; 8];
        header[4..8].copy_from_slice(&64u32.to_be_bytes());
        dribbler.send_binary(header).await;
        loop {
            tokio::time::sleep(std::time::Duration::from_millis(40)).await;
            dribbler.send_binary(vec![0u8]).await;
        }
    });

    let close = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        loop {
            match ws.next_close(std::time::Duration::from_millis(250)).await {
                Some(Some(code)) => break code,
                Some(None) => break 1006,
                None => continue,
            }
        }
    })
    .await
    .expect("dribbling client must be evicted within 10 s");
    dribble.abort();
    assert_eq!(
        close,
        alkhttp::websocket::WS_GOING_AWAY,
        "the forever-dribble hits the progress deadline despite arriving messages"
    );
}

/// WS-13 survivor side on the axum path: a session whose traffic keeps
/// completing chunks (each chunk inside the window — the scaled-down
/// "messages flowing that make progress" shape) is NOT evicted, and
/// calls keep round-tripping across many windows.
#[tokio::test]
async fn idle_progress_knob_survives_productive_sessions_over_axum_upgrade() {
    use alkhttp::server::HttpAdapter;

    let registry = echo_registry();
    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            (s == "tok-1").then(|| identity("alice", &[]))
        }
    }
    let adapter = HttpAdapter::new(Arc::new(StaticTok), Arc::clone(&registry))
        .with_ws_idle_timeout(Some(std::time::Duration::from_millis(150)));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = format!("ws://{}", listener.local_addr().unwrap());
    let app: axum::Router = adapter.router().clone();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    for i in 0..6 {
        tokio::time::sleep(std::time::Duration::from_millis(75)).await;
        let n = i * 7 + 1;
        let env = call_and_await(
            &mut ws,
            &format!("req-{n}"),
            "echo/run",
            serde_json::json!({ "n": n }),
        )
        .await;
        assert_eq!(env.r#type, EVENT_RESPONDED);
        assert_eq!(env.payload["output"]["n"], n);
    }
    ws.close().await;
}

/// WS-13 `None` arm on the axum path: with the knob disabled a
/// dribbling client that completes no chunk is never evicted.
#[tokio::test]
async fn idle_progress_knob_none_disables_eviction_over_axum_upgrade() {
    use alkhttp::server::HttpAdapter;

    let registry = echo_registry();
    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            (s == "tok-1").then(|| identity("alice", &[]))
        }
    }
    let adapter =
        HttpAdapter::new(Arc::new(StaticTok), Arc::clone(&registry)).with_ws_idle_timeout(None);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = format!("ws://{}", listener.local_addr().unwrap());
    let app: axum::Router = adapter.router().clone();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let mut header = vec![0u8; 8];
    header[4..8].copy_from_slice(&8u32.to_be_bytes());
    ws.send_binary(header).await;
    for _ in 0..4 {
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;
        ws.send_binary_piece(&[0u8]).await;
    }
    let no_close = ws.next_close(std::time::Duration::from_millis(400)).await;
    assert!(
        !matches!(no_close, Some(Some(_))),
        "knob disabled: no eviction arrives, got {no_close:?}"
    );
    ws.close().await;
}

/// WS-17 acceptance: a bare-registry upgrade route takes the pump
/// knobs per request via the `WsTimeouts` extension (mirroring
/// `ChannelsPolicy`) — a client that completes no chunk is evicted
/// with 1001 inside the extension's short idle window, not the 60 s
/// default.
#[tokio::test]
async fn ws_timeouts_extension_sets_the_idle_window_on_a_bare_registry_route() {
    let registry = std::sync::Arc::new(OperationRegistry::new());
    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            (s == "tok-1").then(|| identity("alice", &[]))
        }
    }
    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            std::sync::Arc::new(StaticTok) as std::sync::Arc<dyn IdentityProvider>,
            alkhttp::websocket::ws_bearer_auth,
        ))
        .layer(axum::Extension(alkhttp::websocket::WsTimeouts {
            idle: Some(std::time::Duration::from_millis(150)),
            write: None,
        }))
        .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();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let close = ws.next_close(std::time::Duration::from_secs(5)).await;
    assert_eq!(
        close,
        Some(Some(alkhttp::websocket::WS_GOING_AWAY)),
        "evicted with 1001 inside the extension's idle window"
    );
    ws.close().await;
}

// --- Unit 2/3 gates (review 006): the data-channel + op/register wiring ---

use alkcall::channels::operations::OpenHandler;

/// The data-plane echo the `OpenHandler` runs on the allocated
/// channel's `Connection`: echo every inbound byte block back, then
/// drain until EOF. The WS-test client writes raw (channel_id,
/// payload) chunks; the handler reads the channel's BiStream and
/// writes the echo — the exact data-plane shape ADR-067 promises.
fn echo_open_handler() -> OpenHandler {
    Arc::new(move |_input, channel_conn, _auth| {
        tokio::spawn(async move {
            let mut stream = match channel_conn.accept_bi().await {
                Ok(s) => s,
                Err(_) => return,
            };
            use tokio::io::{AsyncReadExt, AsyncWriteExt};
            let mut buf = [0u8; 4096];
            loop {
                match stream.read(&mut buf).await {
                    Ok(0) | Err(_) => break,
                    Ok(n) => {
                        if stream.write_all(&buf[..n]).await.is_err() {
                            break;
                        }
                    }
                }
            }
            let _ = stream.shutdown().await;
        })
    })
}

/// The open-op spec for the test's openable ALPN (`channels/echo/sub`,
/// opening the `alk/echo` data plane).
fn echo_open_spec() -> OperationSpec {
    OperationSpec::new(
        "channels/echo/sub",
        OperationType::Sub,
        Visibility::External,
        serde_json::json!({
            "type": "object",
            "properties": {},
            "additionalProperties": false
        }),
        serde_json::json!({
            "type": "object",
            "properties": { "channel_id": { "type": "integer" } }
        }),
        vec![],
        AccessControl::default(),
        None,
    )
    .with_channel_open(alkcall::registry::spec::ChannelOpenSpec::new("alk/echo"))
}

/// Spawn a WS server whose adapter declares the echo openable ALPN and
/// a small per-identity channel cap; returns the WS URL and the shared
/// policy (for the ledger assertions).
async fn spawn_openable_ws_server(
    registry: Arc<OperationRegistry>,
    cap: usize,
) -> (
    String,
    Arc<alkcall::channels::policy::PerIdentityChannelPolicy>,
) {
    use alkhttp::websocket::OpenableAlpns;

    let policy = Arc::new(alkcall::channels::policy::PerIdentityChannelPolicy::new(
        cap,
    ));
    let policy_dyn: Arc<dyn alkcall::channels::policy::ChannelLifecyclePolicy> =
        Arc::clone(&policy) as Arc<dyn alkcall::channels::policy::ChannelLifecyclePolicy>;

    let openables = OpenableAlpns(Arc::from([alkhttp::websocket::OpenableAlpn::new(
        echo_open_spec(),
        echo_open_handler(),
    )]));

    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            provider_with(vec![("tok-1", identity("alice", &[]))]),
            alkhttp::websocket::ws_bearer_auth,
        ))
        .layer(axum::Extension(openables))
        .layer(axum::Extension(alkhttp::websocket::ChannelsPolicy(
            policy_dyn,
        )))
        .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"), policy)
}

/// Drain channel-0 chunks until a complete envelope shows up (shared
/// with `call_and_await`'s loop but tolerant of interleaved
/// non-zero-channel chunks).
async fn await_envelope(ws: &mut WsClient, request_id: &str) -> EventEnvelope {
    let mut chunks = ChunkAssembler::new();
    let mut frames = FrameAssembler::new();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "timed out waiting for {request_id}"
        );
        if let Some(env) = frames.next_frame() {
            if env.id == request_id {
                return env;
            }
        }
        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
        match bin {
            Some(bytes) => {
                chunks.push(&bytes);
                while let Some((channel_id, payload)) = chunks.next_chunk() {
                    assert_eq!(channel_id, 0, "response must ride channel 0");
                    frames.push(&payload);
                }
            }
            None => panic!("ws closed unexpectedly while awaiting {request_id}"),
        }
    }
}

/// Unit 3 scenario 1+2 (review 006): the open op resolves with a
/// `channel_id`, the per-session openable is discoverable through
/// `services/list`, and data chunks flow both directions on the
/// channel — the `OpenHandler`'s `Connection` sees the bytes.
#[tokio::test]
async fn data_channel_open_discoverable_and_bytes_round_trip() {
    let (url, _policy) = spawn_openable_ws_server(
        echo_registry(),
        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
    )
    .await;
    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();

    // Discovery first: the fork's bootstrap discovery lists the
    // session's own openable (the F-06 shape).
    let env = call_and_await(&mut ws, "req-list", "services/list", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    let names: Vec<String> = env.payload["output"]["operations"]
        .as_array()
        .expect("operations array")
        .iter()
        .filter_map(|o| o["name"].as_str().map(String::from))
        .collect();
    assert!(
        names.contains(&"channels/echo/sub".to_string()),
        "per-session openable discoverable: {names:?}"
    );
    assert!(
        names.contains(&"channel/close".to_string()) && names.contains(&"op/register".to_string()),
        "generic channel ops + op/register served: {names:?}"
    );

    // The open op resolves with a channel_id.
    let env = call_and_await(
        &mut ws,
        "req-open",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    let channel_id = env.payload["output"]["channel_id"]
        .as_u64()
        .expect("channel_id in open response");
    assert!(channel_id > 0, "even id from the accept side: {channel_id}");

    // Data-plane echo: send one chunk on the data channel and read the
    // handler's echo back. Data-channel payloads are opaque (ADR-035:
    // the handler owns its framing) — the echo handler echoes
    // byte-for-byte per read.
    let payload = b"hello-data-plane";
    let framed = frame_data_channel(channel_id as u32, payload);
    ws.send_binary(framed).await;

    let echoed = read_data_channel_block(&mut ws, channel_id as u32).await;
    assert_eq!(echoed, payload, "handler echoed the bytes");
    ws.close().await;
}

/// Frame a raw data-channel chunk as a WS binary message: the 8-byte
/// chunk header (channel_id BE + length BE) + payload — the same wire
/// format channel 0 uses, the opaque data-plane unit (ADR-035).
fn frame_data_channel(channel_id: u32, block: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(8 + block.len());
    out.extend_from_slice(&channel_id.to_be_bytes());
    out.extend_from_slice(&(block.len() as u32).to_be_bytes());
    out.extend_from_slice(block);
    out
}

/// Read one echoed chunk's payload off a data channel, skipping
/// channel-0 traffic.
async fn read_data_channel_block(ws: &mut WsClient, channel_id: u32) -> Vec<u8> {
    let mut chunks = ChunkAssembler::new();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "timed out waiting for data-channel {channel_id}"
        );
        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
        match bin {
            Some(bytes) => {
                chunks.push(&bytes);
                while let Some((cid, payload)) = chunks.next_chunk() {
                    if cid == channel_id {
                        return payload;
                    }
                }
            }
            None => panic!("ws closed unexpectedly while awaiting data channel {channel_id}"),
        }
    }
}

/// Unit 3 scenario 2 (review 006): `channel/close` tears the handler
/// stream down and the opener ledger decrements (the policy count
/// drops) — plus the close op itself resolves `{"closed": true}`.
#[tokio::test]
async fn channel_close_resolves_and_decrements_the_opener_ledger() {
    let (url, policy) = spawn_openable_ws_server(
        echo_registry(),
        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
    )
    .await;
    let alice = identity("alice", &[]);
    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();

    let env = call_and_await(
        &mut ws,
        "req-open",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    let channel_id = env.payload["output"]["channel_id"].as_u64().unwrap() as u32;
    assert_eq!(policy.count_for(&alice), 1, "open incremented the ledger");

    let env = call_and_await(
        &mut ws,
        "req-close",
        "channel/close",
        serde_json::json!({ "channel_id": channel_id }),
    )
    .await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    assert_eq!(
        env.payload["output"]["closed"], true,
        "close op resolves: {}",
        env.payload["output"]
    );
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "ledger decrement never landed"
        );
        if policy.count_for(&alice) == 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
    ws.close().await;
}

/// Unit 3 scenario 3: the per-identity cap denies the open over the
/// limit with the `channel:` error prefix (the policy is the
/// `ChannelsPolicy` extension; the open wrapper consults it before
/// allocation).
#[tokio::test]
async fn channel_cap_denies_open_over_the_limit() {
    let (url, policy) = spawn_openable_ws_server(echo_registry(), 1).await;
    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();

    let env = call_and_await(
        &mut ws,
        "req-open-1",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    assert!(
        env.payload["output"]["channel_id"].is_u64(),
        "first open admitted"
    );

    let env = call_and_await(
        &mut ws,
        "req-open-2",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(
        env.r#type, EVENT_ERROR,
        "cap denial: got {} payload {}",
        env.r#type, env.payload
    );
    let error = &env.payload;
    assert!(
        error["code"]
            .as_str()
            .is_some_and(|c| c.starts_with("channel:")),
        "channel-prefixed denial code: {error}"
    );
    assert_eq!(policy.count_for(&identity("alice", &[])), 1);
    ws.close().await;
}

/// Unit 3 gate (the WS-24/WS-25 shape): `op/register` is served per
/// session and its collision policy binds the session fork. Announce
/// lands in the connection overlay (re-announce without `replace`
/// rejects with the overlay gate's `ALREADY_EXISTS`); announcing a name
/// the serving side itself registers is rejected with
/// `ALREADY_EXISTS` regardless of `replace` (review 005 G-03's gate,
/// through the WS path).
#[tokio::test]
async fn op_register_served_per_session_and_collision_is_already_exists() {
    use alkcall::registry::op_register::OpRegisterRequest;

    let addr = spawn_ws_server(
        echo_registry(),
        provider_with(vec![("tok-1", identity("alice", &[]))]),
    )
    .await;
    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    // A distinct name announces cleanly into the connection overlay.
    let spec = OperationSpec::new(
        "consumer/exec",
        OperationType::Query,
        Visibility::External,
        serde_json::json!({}),
        serde_json::json!({}),
        vec![],
        AccessControl::default(),
        None,
    );
    let request = OpRegisterRequest {
        spec,
        replace: false,
    };
    let frame = EventEnvelope::requested(
        "req-announce",
        serde_json::json!({
            "operationId": "op/register",
            "input": request.to_json(),
        }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;
    let env = await_envelope(&mut ws, &frame.id).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    assert_eq!(env.payload["output"]["registered"], true);

    // The announced op is discoverable via services/list-peers: the
    // peer entry keyed by the connection identity (alice) lists it
    // (UP-03, fixed in alkcall 0.3.1).
    let peers = call_and_await(
        &mut ws,
        "req-list-peers",
        "services/list-peers",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(peers.r#type, EVENT_RESPONDED, "got {}", peers.r#type);
    let peer_entry = peers.payload["output"]["peers"]
        .as_array()
        .expect("peers array")
        .iter()
        .find(|p| p["peer_id"] == "alice")
        .expect("alice peer entry present");
    let peer_op_names: Vec<&str> = peer_entry["operations"]
        .as_array()
        .expect("peer operations array")
        .iter()
        .filter_map(|o| o["name"].as_str())
        .collect();
    assert!(
        peer_op_names.contains(&"consumer/exec"),
        "announced op discoverable via services/list-peers: {peer_entry}"
    );

    // The announced op landed in the overlay: a second announce of the
    // same name without `replace` hits the overlay collision gate.
    let spec = OperationSpec::new(
        "consumer/exec",
        OperationType::Query,
        Visibility::External,
        serde_json::json!({}),
        serde_json::json!({}),
        vec![],
        AccessControl::default(),
        None,
    );
    let request = OpRegisterRequest {
        spec,
        replace: false,
    };
    let frame = EventEnvelope::requested(
        "req-reannounce",
        serde_json::json!({
            "operationId": "op/register",
            "input": request.to_json(),
        }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;
    let env = await_envelope(&mut ws, &frame.id).await;
    assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type);
    assert_eq!(
        env.payload["code"], "ALREADY_EXISTS",
        "overlay collision gate: {}",
        env.payload
    );

    // Collision: announcing the serving side's own op name rejects with
    // ALREADY_EXISTS regardless of replace (review 005 G-03's gate,
    // through the WS path).
    let spec = OperationSpec::new(
        "echo/run",
        OperationType::Query,
        Visibility::External,
        serde_json::json!({}),
        serde_json::json!({}),
        vec![],
        AccessControl::default(),
        None,
    );
    let request = OpRegisterRequest {
        spec,
        replace: true,
    };
    let frame = EventEnvelope::requested(
        "req-collide",
        serde_json::json!({
            "operationId": "op/register",
            "input": request.to_json(),
        }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;
    let env = await_envelope(&mut ws, &frame.id).await;
    assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type);
    assert_eq!(
        env.payload["code"], "ALREADY_EXISTS",
        "serving-registry collision rejected: {}",
        env.payload
    );
    ws.close().await;
}

/// Unit 3 scenario 4: a mid-open disconnect tears the session down
/// without leaking — the ledger count for the opener drops to zero
/// (the channels adapter's teardown decrements), the WS close is
/// observed, and no further open succeeds (the socket is gone).
#[tokio::test]
async fn disconnect_mid_open_tears_down_and_decrements() {
    let (url, policy) = spawn_openable_ws_server(
        echo_registry(),
        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
    )
    .await;
    let alice = identity("alice", &[]);
    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();

    let env = call_and_await(
        &mut ws,
        "req-open",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    assert!(env.payload["output"]["channel_id"].is_u64());
    assert_eq!(policy.count_for(&alice), 1);

    // Hard disconnect (drop without close frame — the pump sees EOF).
    drop(ws);
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        assert!(
            tokio::time::Instant::now() < deadline,
            "teardown never decremented the ledger"
        );
        if policy.count_for(&alice) == 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
}

/// Unit 3 scenario 5: a `TooLarge` chunk on a data channel survives —
/// the demux skips it and resyncs (alkcall's hardening, asserted
/// through the WS path). The oversized chunk is skipped by the demux
/// (TooLarge), the following chunks on other channels still route.
#[tokio::test]
async fn too_large_data_channel_chunk_survives_demux_resync() {
    use alkhttp::websocket::MAX_CHUNK_LEN;

    let (url, _policy) = spawn_openable_ws_server(
        echo_registry(),
        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
    )
    .await;
    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();

    let env = call_and_await(
        &mut ws,
        "req-open",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    let channel_id = env.payload["output"]["channel_id"].as_u64().unwrap() as u32;

    // An oversized declared chunk on the data channel: the demux skips
    // it (bounded by its 256 MiB budget) and stays in sync. The skip
    // consumes exactly the declared payload length off the transport,
    // so we must actually send that many filler bytes (16 MiB + 1 is
    // over MAX_CHUNK_LEN but under the WS message caps, sent as
    // separate WS messages).
    let oversized_len = MAX_CHUNK_LEN + 1;
    let mut header = Vec::with_capacity(8);
    header.extend_from_slice(&channel_id.to_be_bytes());
    header.extend_from_slice(&oversized_len.to_be_bytes());
    ws.send_binary(header).await;
    let filler_len = oversized_len as usize;
    let filler = vec![0u8; filler_len];
    let mut sent = 0usize;
    while sent < filler_len {
        let take = (filler_len - sent).min(256 * 1024);
        ws.send_binary(filler[sent..sent + take].to_vec()).await;
        sent += take;
    }

    // The next small chunk on the data channel still routes and the
    // handler echoes it — resync proven.
    let payload = b"after-skip";
    ws.send_binary(frame_data_channel(channel_id, payload))
        .await;
    let echoed = read_data_channel_block(&mut ws, channel_id).await;
    assert_eq!(echoed, payload, "demux resynced after the TooLarge skip");
    ws.close().await;
}

// --- Review 007 remediation gates (WS-29/30/32) ---

/// The announce frame the `op/register` gates send: one
/// `OpRegisterRequest` wrapped as a channel-0 chunk.
async fn announce_op(ws: &mut WsClient, request_id: &str, name: &str) -> EventEnvelope {
    use alkcall::registry::op_register::OpRegisterRequest;

    let spec = OperationSpec::new(
        name,
        OperationType::Query,
        Visibility::External,
        serde_json::json!({}),
        serde_json::json!({}),
        vec![],
        AccessControl::default(),
        None,
    );
    let request = OpRegisterRequest {
        spec,
        replace: false,
    };
    let frame = EventEnvelope::requested(
        request_id,
        serde_json::json!({
            "operationId": "op/register",
            "input": request.to_json(),
        }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;
    await_envelope(ws, &frame.id).await
}

/// Review 007 WS-29 acceptance (the builder path): a deployment's
/// `with_ws_op_register_acl` value is the `op/register` surface's ACL —
/// a peer whose identity lacks the required scope gets `FORBIDDEN` on
/// the announce, a peer with it announces cleanly. The permissive
/// default (any authenticated peer may announce) is pinned by the
/// Unit-3 gate above, whose registry route carries no override.
#[tokio::test]
async fn op_register_acl_builder_gates_the_announce_surface() {
    use alkhttp::server::HttpAdapter;

    let registry = echo_registry();
    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            match s.as_str() {
                "tok-1" => Some(identity("alice", &[])),
                "tok-2" => Some(identity("bobb", &["announcer"])),
                _ => None,
            }
        }
    }

    let adapter = HttpAdapter::new(
        Arc::new(StaticTok) as Arc<dyn IdentityProvider>,
        Arc::clone(&registry),
    )
    .with_ws_op_register_acl(AccessControl {
        required_scopes: vec!["announcer".to_string()],
        ..Default::default()
    });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = format!("ws://{}", listener.local_addr().unwrap());
    let app: axum::Router = adapter.router().clone();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });
    let url = format!("{addr}/alk/channels");

    // Scope-less peer: the announce is denied before the handler runs.
    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
    let env = call_and_await(&mut ws, "req-deny", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "session established");
    let env = announce_op(&mut ws, "req-announce-deny", "consumer/denied").await;
    assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type);
    assert_eq!(
        env.payload["code"], "FORBIDDEN",
        "ACL without the scope is denied: {}",
        env.payload
    );

    // A peer with the scope announces cleanly.
    let mut ws2 = WsClient::connect_authorized(&url, "tok-2").await.unwrap();
    let env = announce_op(&mut ws2, "req-announce-ok", "consumer/allowed").await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    assert_eq!(env.payload["output"]["registered"], true);
    ws.close().await;
    ws2.close().await;
}

/// Review 007 WS-29 acceptance (the extension path): the
/// `OpRegisterAcl` request extension on a bare-registry route carries
/// the same gate — `FORBIDDEN` without the scope, announce-ok with it.
#[tokio::test]
async fn op_register_acl_extension_gates_the_announce_surface() {
    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            provider_with(vec![
                ("tok-1", identity("alice", &[])),
                ("tok-2", identity("bobb", &["announcer"])),
            ]),
            alkhttp::websocket::ws_bearer_auth,
        ))
        .layer(axum::Extension(alkhttp::websocket::OpRegisterAcl(
            AccessControl {
                required_scopes: vec!["announcer".to_string()],
                ..Default::default()
            },
        )))
        .with_state(echo_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();
    });
    let url = format!("{addr}/alk/channels");

    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
    let env = announce_op(&mut ws, "req-announce-deny", "consumer/denied").await;
    assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type);
    assert_eq!(
        env.payload["code"], "FORBIDDEN",
        "extension ACL without the scope is denied: {}",
        env.payload
    );

    let mut ws2 = WsClient::connect_authorized(&url, "tok-2").await.unwrap();
    let env = announce_op(&mut ws2, "req-announce-ok", "consumer/allowed").await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    assert_eq!(env.payload["output"]["registered"], true);
    ws.close().await;
    ws2.close().await;
}

/// Review 007 WS-30 acceptance: a bare-registry route's
/// `SessionState` semaphore is built per request (`FromRef`) and
/// bounds nothing; the `SessionSlots` extension carries the shared
/// semaphore — two sessions over a cap of 1 reject with 503, and a
/// slot frees on session end (the WS-09 shape on the extension path).
#[tokio::test]
async fn session_slots_extension_bounds_a_bare_registry_route() {
    let slots = Arc::new(tokio::sync::Semaphore::new(1));
    let app = axum::Router::new()
        .route(
            "/alk/channels",
            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
        )
        .layer(axum::middleware::from_fn_with_state(
            provider_with(vec![("tok-1", identity("alice", &[]))]),
            alkhttp::websocket::ws_bearer_auth,
        ))
        .layer(axum::Extension(alkhttp::websocket::SessionSlots(
            Arc::clone(&slots),
        )))
        .with_state(echo_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();
    });
    let url = format!("{addr}/alk/channels");

    let mut first = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
    let env = call_and_await(&mut first, "req-1", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "first session admitted");

    let status = WsClient::connect_status(&url, Some("tok-1")).await;
    assert_eq!(status, Some(503), "second session rejected over the cap");

    first.close().await;
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let mut third = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
    let env = call_and_await(&mut third, "req-2", "echo/run", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "slot freed after session end");
    third.close().await;
}

/// Review 007 WS-32 acceptance: the built-in openables threading
/// (`HttpAdapter::with_ws_openable_alpns` → `RouterState` →
/// `SessionState` → hook) serves the data-channel surface — a session
/// on the built-in router discovers the openable via `services/list`,
/// opens the channel, and round-trips bytes on it. Every Unit-3 gate
/// rides the `OpenableAlpns` request-extension fallback; this is the
/// only gate through the builder path a real `HttpAdapter` deployment
/// uses.
#[tokio::test]
async fn builder_path_openables_serve_the_data_channel_surface() {
    use alkhttp::server::HttpAdapter;

    struct StaticTok;
    impl IdentityProvider for StaticTok {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let s = String::from_utf8_lossy(&token.raw).to_string();
            (s == "tok-1").then(|| identity("alice", &[]))
        }
    }

    let adapter =
        HttpAdapter::new(Arc::new(StaticTok), echo_registry()).with_ws_openable_alpns(vec![
            alkhttp::websocket::OpenableAlpn::new(echo_open_spec(), echo_open_handler()),
        ]);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = format!("ws://{}", listener.local_addr().unwrap());
    let app: axum::Router = adapter.router().clone();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
        .await
        .unwrap();

    let env = call_and_await(&mut ws, "req-list", "services/list", serde_json::json!({})).await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    let names: Vec<String> = env.payload["output"]["operations"]
        .as_array()
        .expect("operations array")
        .iter()
        .filter_map(|o| o["name"].as_str().map(String::from))
        .collect();
    assert!(
        names.contains(&"channels/echo/sub".to_string()),
        "builder-path openable discoverable: {names:?}"
    );

    let env = call_and_await(
        &mut ws,
        "req-open",
        "channels/echo/sub",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
    let channel_id = env.payload["output"]["channel_id"]
        .as_u64()
        .expect("channel_id");

    let payload = b"builder-path-bytes";
    ws.send_binary(frame_data_channel(channel_id as u32, payload))
        .await;
    let echoed = read_data_channel_block(&mut ws, channel_id as u32).await;
    assert_eq!(
        echoed, payload,
        "data channel round trip on the built-in surface"
    );
    ws.close().await;
}