test(websocket): axum-path idle-progress mirror tests (WS-13)

Mirrors the WS-13 progress semantics onto the axum upgrade path with
three integration tests over the real HttpAdapter surface (also the
COV-11b dark-knob gate — with_ws_idle_timeout had no test caller):
the forever-dribble client is evicted with 1001 despite arriving
messages; a productive session round-trips calls across many windows
without eviction; the None knob never evicts a dribbling client.

tests/ws_upgrade_session.rs

@@ -758,3 +758,163 @@ async fn session_cap_rejects_over_limit_with_503_and_frees_slots_on_end() {
     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;
+}