// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! SFTP compliance regression suite.
//!
//! Per-case assertions that close coverage gaps in SFTP testing.
//! sftp_core covers core functionality. Tests here cover compliance
//! with the SFTP spec, lifecycle invariants under abnormal client
//! patterns (half-close, wedge, paused-drain), and pipelining shapes
//! the OpenSSH and paramiko coverage in sftp_core does not exercise.
//! Each case carries a CMPTST-NN identifier so a failure log line
//! points at a single named property.
//!
//! The shared-server entry test_sftp_compliance_suite spawns one
//! rustfs binary and runs CMPTST-01 through CMPTST-14 against the
//! same session. CMPTST-15 through CMPTST-23 cover read-only mode
//! and run under test_sftp_compliance_readonly with a separate rustfs
//! server started with RUSTFS_SFTP_READ_ONLY=true. The remaining
//! cases (CMPTST-24 through CMPTST-33) each spawn a dedicated rustfs
//! server because the property under test depends on a specific
//! per-server configuration (idle timeout, read-cache window) or a
//! dedicated TCP port.
//!
//! # Case index
//!
//! Shared server (test_sftp_compliance_suite):
//!
//! - CMPTST-01: medium-binary upload then download with SHA256 compare,
//!   single-shot PutObject path below the multipart boundary.
//! - CMPTST-02: zero-byte upload, download, and stat-size match.
//! - CMPTST-03: rm against a bucket path is rejected and the bucket
//!   still exists.
//! - CMPTST-04: rmdir against a non-empty bucket is rejected and the
//!   contained object survives.
//! - CMPTST-05: rmdir against a non-empty sub-directory is rejected
//!   and the inner object survives.
//! - CMPTST-06: open with a path-traversal pattern cannot leak a
//!   host file via SFTP read.
//! - CMPTST-07: read_dir of /.. either errors or returns a listing
//!   that contains no host system entries.
//! - CMPTST-08: rename across buckets writes the payload byte-for-byte
//!   at the destination and removes the source object.
//! - CMPTST-09: paths with embedded spaces round-trip through the
//!   russh-sftp client.
//! - CMPTST-10: read_link is rejected (S3 storage has no symlinks).
//! - CMPTST-11: SETSTAT on a path and FSETSTAT on a separate open
//!   handle both return ok (rsync, WinSCP transfer-success contract).
//! - CMPTST-12: rename to the same path is a no-op and the file stays
//!   in place with the original payload (no copy-then-delete data
//!   loss).
//! - CMPTST-13: nested-key upload creates the parent directory
//!   implicitly and three listing forms show the inner file
//!   (implicit-directory round-trip).
//! - CMPTST-14: OPEN, WRITE, FSETSTAT, CLOSE on the same write handle
//!   all return ok (WinSCP packet sequence, where FSETSTAT against an
//!   in-flight write handle must not error).
//!
//! Read-only spawn (test_sftp_compliance_readonly,
//! RUSTFS_SFTP_READ_ONLY=true):
//!
//! - CMPTST-15: put through SFTP is rejected.
//! - CMPTST-16: rm through SFTP is rejected.
//! - CMPTST-17: mkdir through SFTP is rejected.
//! - CMPTST-18: rmdir through SFTP is rejected.
//! - CMPTST-19: rename through SFTP is rejected.
//! - CMPTST-20: ls through SFTP is allowed and lists the seeded bucket.
//! - CMPTST-21: get through SFTP is allowed and returns the seeded
//!   payload byte-for-byte.
//! - CMPTST-22: SETSTAT on a path is rejected with PermissionDenied.
//! - CMPTST-23: FSETSTAT on a read handle is rejected with
//!   PermissionDenied.
//!
//! Standalone-server cases:
//!
//! - CMPTST-24: concurrent half-close burst does not leak server-side
//!   session tasks (TCP half-close mid-transfer must drain).
//! - CMPTST-25: wedge-kill watchdog kills sessions parked on the russh
//!   per-channel mpsc behind a CLOSE_WAIT socket.
//! - CMPTST-26: healthy idle session past the watchdog fast-kill
//!   threshold stays alive (procfs ESTABLISHED discriminator must not
//!   false-kill).
//! - CMPTST-27: sustained-read thrash, multi-GiB downloads on N
//!   parallel sessions all byte-identical to seed.
//! - CMPTST-28: 5 MB download intact under concurrent metadata storm
//!   on a parallel session.
//! - CMPTST-29: high-volume read-past-EOF pipelining completes inside
//!   the deadline and every read returns EOF.
//! - CMPTST-30: per-operation handler latency stays inside the
//!   ceiling under parallel pipelined sessions (ignored by default).
//! - CMPTST-31: server resilience under client paused-drain, byte-exact
//!   completion after a mid-transfer pause window.
//! - CMPTST-32: read-cache enabled regression, 8 MiB download
//!   byte-exact with the production cache window.
//! - CMPTST-33: read-cache disabled regression, 8 MiB download
//!   byte-exact with RUSTFS_SFTP_READ_CACHE_WINDOW_BYTES=0.
//! - CMPTST-34: OPEN with non-default FileAttributes followed by a
//!   payload that crosses the 5 MiB multipart boundary preserves the
//!   client-supplied mtime and permissions through the streaming
//!   CreateMultipartUpload path. HeadObject through aws-sdk-s3
//!   confirms the metadata reached the finalised S3 object.

use crate::common::rustfs_binary_path_with_features;
use crate::protocols::sftp_helpers::{
    AcceptAnyServerKey, ServerProcess, build_test_s3_client, connect_sftp_to, generate_host_key, sftp_read_full,
    wait_for_s3_ready,
};
use crate::protocols::test_env::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, ProtocolTestEnvironment};
use anyhow::{Result, anyhow};
use aws_sdk_s3::Client as S3Client;
use aws_sdk_s3::primitives::ByteStream;
use futures::stream::{FuturesUnordered, StreamExt};
use russh::client;
use russh_sftp::client::{Config, SftpSession};
use russh_sftp::protocol::{FileAttributes, OpenFlags, StatusCode};
use rustfs_config::ENV_SFTP_IDLE_TIMEOUT;
use rustfs_config::{
    ENV_CONSOLE_ENABLE, ENV_RUSTFS_ADDRESS, ENV_SFTP_ADDRESS, ENV_SFTP_ENABLE, ENV_SFTP_HOST_KEY_DIR, ENV_SFTP_PART_SIZE,
    ENV_SFTP_READ_CACHE_WINDOW_BYTES, ENV_SFTP_READ_ONLY,
};
use sha2::{Digest, Sha256};
use std::path::PathBuf;
use std::pin::Pin;
use std::process::Stdio;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWrite, AsyncWriteExt, BufReader, ReadBuf};
use tokio::net::TcpStream;
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::process::{Child, Command};
use tokio::time::sleep;
use tokio::time::timeout;
use tracing::info;

// Cross-case constants used by every spawn helper. Pinned to 5 MiB so
// the multipart boundary is deterministic across runs.
const PART_SIZE_ENV: &str = "5242880";

// Number of attempts the pipelining-style cases give the rustfs S3
// endpoint before failing the readiness wait.
const S3_READY_ATTEMPTS: u32 = 30;

// Logger level passed to spawned rustfs binaries that capture stdout
// for diagnostic dumps. Protocol-level info plus warn-everywhere keeps
// the log volume bounded while preserving the SFTP traces the
// failure-path dumps look for.
const PIPELINING_OBS_LOGGER_LEVEL: &str = "warn,rustfs_protocols=info,rustfs_protocols::sftp::diag=info";

// RUST_LOG is a stdlib-side env var with no canonical constant in the
// rustfs config crate. Left as a literal at the call sites that set it.

// Cross-case fixture parameters used by the pipelining cases that
// exercise the GUI-client traversal shape (a multi-MB fixture object
// next to a sub-directory containing several siblings).
const FIXTURE_SIZE: usize = 5 * 1024 * 1024;
const SUBDIR_FILE_COUNT: usize = 200;

// Pattern multiplier for deterministic seed payloads. Every case that
// seeds a multi-MB fixture and verifies via SHA256 reads from this.
const THRASH_PATTERN_MULTIPLIER: u8 = 13;

/// Build a fresh ProtocolTestEnvironment, generate a per-test ed25519
/// host key, and spawn a rustfs binary configured for SFTP compliance
/// testing on the given bind addresses. The caller owns both returned
/// values: the env keeps the temp directory alive (its Drop cleans it
/// up), and the ServerProcess wrapper guarantees a SIGKILL on every
/// path including panic unwind.
pub(crate) async fn spawn_compliance_rustfs(
    sftp_address: &str,
    s3_address: &str,
    read_only: bool,
) -> Result<(ProtocolTestEnvironment, ServerProcess)> {
    let env = ProtocolTestEnvironment::new().map_err(|e| anyhow!("{}", e))?;
    let host_key_dir = PathBuf::from(&env.temp_dir).join("sftp_host_keys");
    generate_host_key(&host_key_dir).await?;

    let binary_path = rustfs_binary_path_with_features(Some("ftps,webdav,sftp"));
    let host_key_dir_str = host_key_dir
        .to_str()
        .ok_or_else(|| anyhow!("host key dir path is not utf-8: {}", host_key_dir.display()))?;
    let child = Command::new(&binary_path)
        .env(ENV_CONSOLE_ENABLE, "false")
        .env(ENV_SFTP_ENABLE, "true")
        .env(ENV_SFTP_ADDRESS, sftp_address)
        .env(ENV_SFTP_HOST_KEY_DIR, host_key_dir_str)
        .env(ENV_SFTP_READ_ONLY, if read_only { "true" } else { "false" })
        .env(ENV_SFTP_PART_SIZE, PART_SIZE_ENV)
        .env(ENV_RUSTFS_ADDRESS, s3_address)
        .arg(&env.temp_dir)
        .spawn()?;
    Ok((env, ServerProcess::new(child)))
}

/// Build a fresh ProtocolTestEnvironment, generate a per-test ed25519
/// host key, and spawn a rustfs binary configured for the pipelining
/// regression cases. Same ownership contract as spawn_compliance_rustfs.
async fn spawn_pipelining_rustfs(sftp_address: &str, s3_address: &str) -> Result<(ProtocolTestEnvironment, ServerProcess)> {
    spawn_pipelining_rustfs_with_extras(sftp_address, s3_address, &[]).await
}

/// Variant of spawn_pipelining_rustfs that layers additional
/// environment variables onto the spawned rustfs binary. Pairs of
/// (name, value) are forwarded as Command::env calls in iteration
/// order so a later pair overrides an earlier one for the same name.
async fn spawn_pipelining_rustfs_with_extras(
    sftp_address: &str,
    s3_address: &str,
    extra_env: &[(&str, &str)],
) -> Result<(ProtocolTestEnvironment, ServerProcess)> {
    let env = ProtocolTestEnvironment::new().map_err(|e| anyhow!("{}", e))?;
    let host_key_dir = PathBuf::from(&env.temp_dir).join("sftp_host_keys");
    generate_host_key(&host_key_dir).await?;

    let binary_path = rustfs_binary_path_with_features(Some("ftps,webdav,sftp"));
    let host_key_dir_str = host_key_dir
        .to_str()
        .ok_or_else(|| anyhow!("host key dir path is not utf-8: {}", host_key_dir.display()))?;
    let mut cmd = Command::new(&binary_path);
    cmd.env(ENV_SFTP_ENABLE, "true")
        .env(ENV_SFTP_ADDRESS, sftp_address)
        .env(ENV_SFTP_HOST_KEY_DIR, host_key_dir_str)
        .env(ENV_SFTP_READ_ONLY, "false")
        .env(ENV_SFTP_PART_SIZE, PART_SIZE_ENV)
        .env(ENV_RUSTFS_ADDRESS, s3_address)
        .env(ENV_CONSOLE_ENABLE, "false")
        .env("RUSTFS_OBS_LOGGER_LEVEL", PIPELINING_OBS_LOGGER_LEVEL)
        .env("RUST_LOG", PIPELINING_OBS_LOGGER_LEVEL);
    for (k, v) in extra_env {
        cmd.env(k, v);
    }
    let child = cmd.stdout(Stdio::piped()).arg(&env.temp_dir).spawn()?;
    Ok((env, ServerProcess::new(child)))
}

/// Seed the bucket with one multi-MB fixture object plus a
/// sub-directory containing several small siblings. The fixture
/// shape mirrors the GUI-client traversal pattern.
async fn seed_pipelining_fixture(s3: &S3Client, bucket: &str, fixture_key: &str, subdir: &str) -> Result<Vec<u8>> {
    s3.create_bucket()
        .bucket(bucket)
        .send()
        .await
        .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;

    let payload: Vec<u8> = (0..FIXTURE_SIZE).map(|i| (i as u8).wrapping_mul(13)).collect();
    s3.put_object()
        .bucket(bucket)
        .key(fixture_key)
        .body(ByteStream::from(payload.clone()))
        .send()
        .await
        .map_err(|e| anyhow!("S3 PutObject {bucket}/{fixture_key} failed: {e:?}"))?;

    for i in 0..SUBDIR_FILE_COUNT {
        let key = format!("{subdir}/file_{i:04}.txt");
        let body = format!("sample-{i}");
        s3.put_object()
            .bucket(bucket)
            .key(&key)
            .body(ByteStream::from(body.into_bytes()))
            .send()
            .await
            .map_err(|e| anyhow!("S3 PutObject {bucket}/{key} failed: {e:?}"))?;
    }

    Ok(payload)
}

/// Seed a multi-GiB object into the rustfs S3 endpoint via multipart
/// upload. Each part is built in memory, uploaded, and dropped, so
/// peak memory stays bounded at one part_size regardless of total
/// fixture size. The byte at object offset p is
/// (p as u8).wrapping_mul(THRASH_PATTERN_MULTIPLIER) so the expected
/// SHA256 can be calculated independently without materialising the
/// fixture in memory.
async fn seed_large_via_multipart(s3: &S3Client, bucket: &str, key: &str, size_bytes: u64) -> Result<()> {
    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
    let part_size: usize = 5 * 1024 * 1024;
    let create = s3
        .create_multipart_upload()
        .bucket(bucket)
        .key(key)
        .send()
        .await
        .map_err(|e| anyhow!("CreateMultipartUpload failed: {e:?}"))?;
    let upload_id = create
        .upload_id
        .ok_or_else(|| anyhow!("CreateMultipartUpload returned no upload_id"))?;
    let mut parts = Vec::new();
    let mut offset: u64 = 0;
    let mut part_number: i32 = 1;
    while offset < size_bytes {
        let chunk = (part_size as u64).min(size_bytes - offset) as usize;
        let mut body = vec![0u8; chunk];
        for (i, b) in body.iter_mut().enumerate() {
            *b = ((offset + i as u64) as u8).wrapping_mul(THRASH_PATTERN_MULTIPLIER);
        }
        let r = s3
            .upload_part()
            .bucket(bucket)
            .key(key)
            .part_number(part_number)
            .upload_id(&upload_id)
            .body(ByteStream::from(body))
            .send()
            .await
            .map_err(|e| anyhow!("UploadPart {part_number} failed: {e:?}"))?;
        let etag = r.e_tag.ok_or_else(|| anyhow!("UploadPart {part_number} returned no ETag"))?;
        parts.push(CompletedPart::builder().part_number(part_number).e_tag(etag).build());
        offset += chunk as u64;
        part_number += 1;
    }
    let completed = CompletedMultipartUpload::builder().set_parts(Some(parts)).build();
    s3.complete_multipart_upload()
        .bucket(bucket)
        .key(key)
        .upload_id(&upload_id)
        .multipart_upload(completed)
        .send()
        .await
        .map_err(|e| anyhow!("CompleteMultipartUpload failed: {e:?}"))?;
    Ok(())
}

/// Calculate the SHA256 of a deterministic pattern object whose byte
/// at offset p is (p as u8).wrapping_mul(multiplier), without
/// materialising the full pattern in memory. The streaming form
/// keeps memory bounded for the multi-GiB sustained-read fixture.
fn calculate_pattern_sha256(size_bytes: u64, multiplier: u8) -> [u8; 32] {
    let mut hasher = Sha256::new();
    let chunk: usize = 1024 * 1024;
    let mut buf = vec![0u8; chunk];
    let mut written: u64 = 0;
    while written < size_bytes {
        let n = (chunk as u64).min(size_bytes - written) as usize;
        for (i, b) in buf[..n].iter_mut().enumerate() {
            *b = ((written + i as u64) as u8).wrapping_mul(multiplier);
        }
        hasher.update(&buf[..n]);
        written += n as u64;
    }
    hasher.finalize().into()
}

/// Streaming SHA256 download: read the SFTP file end-to-end with a
/// bounded scratch buffer so total client memory stays at one buffer
/// regardless of file size. Returns (bytes_read, sha256). The byte
/// count is the canary for a wedge: a partial-read failure appears
/// as fewer bytes than expected without raising an error from the
/// transport layer (the bytes simply stop arriving and the client's
/// timeout fires).
async fn streaming_sha256_download(sftp: &SftpSession, path: &str) -> Result<(u64, [u8; 32])> {
    let mut file = sftp
        .open_with_flags(path, OpenFlags::READ)
        .await
        .map_err(|e| anyhow!("OPEN {path} failed: {e:?}"))?;
    let mut hasher = Sha256::new();
    let mut buf = vec![0u8; 256 * 1024];
    let mut total: u64 = 0;
    loop {
        let n = file.read(&mut buf).await.map_err(|e| anyhow!("READ {path} failed: {e:?}"))?;
        if n == 0 {
            break;
        }
        hasher.update(&buf[..n]);
        total += n as u64;
    }
    let _ = file.shutdown().await;
    Ok((total, hasher.finalize().into()))
}

/// Bounded in-memory ring of the rustfs server's stdout. Spawned
/// from the test entry so the test can dump the last N lines on
/// failure for diagnostic purposes.
fn capture_server_stdout(child: &mut Child) -> Arc<tokio::sync::Mutex<Vec<String>>> {
    let buffer: Arc<tokio::sync::Mutex<Vec<String>>> = Arc::new(tokio::sync::Mutex::new(Vec::new()));
    if let Some(stdout) = child.stdout.take() {
        let buf_clone = Arc::clone(&buffer);
        tokio::spawn(async move {
            let reader = BufReader::new(stdout);
            let mut lines = reader.lines();
            while let Ok(Some(line)) = lines.next_line().await {
                let mut buf = buf_clone.lock().await;
                buf.push(line);
                if buf.len() > 5000 {
                    buf.drain(0..1000);
                }
            }
        });
    }
    buffer
}

/// Counters scraped from the spawned server's stdout. Each "SFTP session
/// task entered" log emits an enter. Each "SFTP session task finished"
/// or "SFTP session task panicked" log emits a finish. The session-
/// lifecycle cases (CMPTST-24, CMPTST-25, CMPTST-26) read both fields
/// to assert the watchdog killed silent sessions on the expected path.
#[derive(Default)]
struct SessionCounters {
    entered: AtomicUsize,
    finished: AtomicUsize,
}

impl SessionCounters {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            entered: AtomicUsize::new(0),
            finished: AtomicUsize::new(0),
        })
    }
}

/// Spawn a background task that reads the child stdout line-by-line and
/// increments the matching counter for every server-side session
/// lifecycle event. The task ends when stdout closes (i.e. when the
/// child is killed at teardown).
fn watch_session_lifecycle_events(child: &mut Child, counters: Arc<SessionCounters>) {
    let Some(stdout) = child.stdout.take() else {
        return;
    };
    tokio::spawn(async move {
        let mut reader = BufReader::new(stdout);
        let mut line = String::new();
        loop {
            line.clear();
            match reader.read_line(&mut line).await {
                Ok(0) => break,
                Ok(_) => {
                    if line.contains("SFTP session task entered") {
                        counters.entered.fetch_add(1, Ordering::Relaxed);
                    } else if line.contains("SFTP session task finished") || line.contains("SFTP session task panicked") {
                        counters.finished.fetch_add(1, Ordering::Relaxed);
                    }
                }
                Err(_) => break,
            }
        }
    });
}

/// Count TCP connections in CLOSE_WAIT against the given local port
/// by shelling out to ss -tn state CLOSE-WAIT. The check is
/// best-effort: if ss is missing on the host the function returns
/// Ok(None) and the caller skips the assertion. The contract is zero
/// CLOSE_WAIT entries attributable to the test.
#[cfg(target_os = "linux")]
async fn count_close_wait_on_port(port: u16) -> Result<Option<usize>> {
    let output = match Command::new("ss").args(["-tn", "state", "CLOSE-WAIT"]).output().await {
        Ok(o) => o,
        Err(_) => return Ok(None),
    };
    if !output.status.success() {
        return Ok(None);
    }
    let stdout = String::from_utf8_lossy(&output.stdout);
    let needle_local = format!(":{port} ");
    let needle_local_eol = format!(":{port}\n");
    let count = stdout
        .lines()
        .filter(|l| l.contains(&needle_local) || l.contains(needle_local_eol.trim_end()))
        .count();
    Ok(Some(count))
}

// CMPTST-01: medium-binary upload then download with SHA256 compare.
pub(crate) mod cmptst_01 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-01";

    // 300 KiB exercises a multi-buffer write through the russh-sftp
    // 256 KiB chunking boundary while staying under part_size, so the
    // upload takes the single-shot PutObject path rather than the
    // multipart path that sftp_core already covers.
    const MEDIUM_BINARY_SIZE: usize = 300 * 1024;

    pub(crate) async fn run_medium_binary_round_trip(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: medium-binary round-trip with SHA256 compare");
        let bucket = "complbucket1";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/medium.bin");
        let content: Vec<u8> = (0..MEDIUM_BINARY_SIZE).map(|i| (i as u8).wrapping_mul(13)).collect();
        let mut wf = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(&content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let read_back = sftp_read_full(sftp, &path).await?;
        if read_back.len() != content.len() {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} medium-binary round-trip length mismatch: expected {}, got {}",
                content.len(),
                read_back.len()
            ));
        }
        if Sha256::digest(&content) != Sha256::digest(&read_back) {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} medium-binary SHA256 mismatch"));
        }

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: medium-binary round-trip SHA256 match");
        Ok(())
    }
}

// CMPTST-02: zero-byte upload, download, and stat-size match.
pub(crate) mod cmptst_02 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-02";

    pub(crate) async fn run_zero_byte_round_trip(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: zero-byte round-trip");
        let bucket = "complzerobucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/zero.txt");
        let mut wf = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let read_back = sftp_read_full(sftp, &path).await?;
        if !read_back.is_empty() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} zero-byte read returned {} bytes", read_back.len()));
        }
        let meta = sftp.metadata(&path).await?;
        if meta.size != Some(0) {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} zero-byte stat reported size {:?}", meta.size));
        }

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: zero-byte round-trip");
        Ok(())
    }
}

// CMPTST-03: rm against a bucket path is rejected and the bucket still exists.
pub(crate) mod cmptst_03 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-03";

    pub(crate) async fn run_rm_on_bucket_path_rejected(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: rm on a bucket path is rejected");
        let bucket = "complrmbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let rm_result = sftp.remove_file(&bucket_path).await;
        if rm_result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} rm on a bucket path must error"));
        }

        let root_entries: Vec<String> = sftp.read_dir("/").await?.map(|e| e.file_name()).collect();
        if !root_entries.iter().any(|n| n == bucket) {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} bucket must still exist after rejected rm"));
        }

        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: rm on a bucket path rejected and the bucket survived");
        Ok(())
    }
}

// CMPTST-04: rmdir against a non-empty bucket is rejected and contents survive.
pub(crate) mod cmptst_04 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-04";

    pub(crate) async fn run_rmdir_nonempty_bucket_rejected(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: rmdir on a non-empty bucket is rejected");
        let bucket = "complfullbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let inner_path = format!("/{bucket}/keep.txt");
        let inner_content = b"keep me\n";
        let mut wf = sftp
            .open_with_flags(&inner_path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(inner_content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let rmdir_result = sftp.remove_dir(&bucket_path).await;
        if rmdir_result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} rmdir on a non-empty bucket must error"));
        }

        let entries: Vec<String> = sftp.read_dir(&bucket_path).await?.map(|e| e.file_name()).collect();
        if !entries.iter().any(|n| n == "keep.txt") {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} object inside the bucket must survive a rejected rmdir, entries were {entries:?}"
            ));
        }

        sftp.remove_file(&inner_path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: rmdir on a non-empty bucket rejected and contents still in place");
        Ok(())
    }
}

// CMPTST-05: rmdir against a non-empty sub-directory is rejected and contents survive.
pub(crate) mod cmptst_05 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-05";

    pub(crate) async fn run_rmdir_nonempty_subdir_rejected(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: rmdir on a non-empty sub-directory is rejected");
        let bucket = "complnedirbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let subdir_path = format!("/{bucket}/sub");
        sftp.create_dir(&subdir_path).await?;

        let inner_path = format!("/{bucket}/sub/inner.txt");
        let inner_content = b"persist\n";
        let mut wf = sftp
            .open_with_flags(&inner_path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(inner_content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let rmdir_result = sftp.remove_dir(&subdir_path).await;
        if rmdir_result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} rmdir on a non-empty sub-directory must error"));
        }

        let read_back = sftp_read_full(sftp, &inner_path).await?;
        if read_back != inner_content {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} inner object must survive a rejected sub-directory rmdir"
            ));
        }

        sftp.remove_file(&inner_path).await?;
        sftp.remove_dir(&subdir_path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: rmdir on a non-empty sub-directory rejected and inner object still in place");
        Ok(())
    }
}

// CMPTST-06: get with a path-traversal pattern cannot leak a host file.
pub(crate) mod cmptst_06 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-06";

    pub(crate) async fn run_path_traversal_get_rejected(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: get with path traversal is rejected");
        let traversal = sftp.open_with_flags("/../../../etc/passwd", OpenFlags::READ).await;
        if traversal.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} path traversal open must error"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: get with path traversal rejected");
        Ok(())
    }
}

// CMPTST-07: read_dir of /.. either errors or returns a listing that contains no host system entries.
pub(crate) mod cmptst_07 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-07";

    pub(crate) async fn run_dotdot_collapses_to_root(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read_dir of /.. does not expose host paths");
        if let Ok(entries) = sftp.read_dir("/..").await {
            let names: Vec<String> = entries.map(|e| e.file_name()).collect();
            for forbidden in ["etc", "bin", "usr", "lib", "var", "tmp", "root", "home", "proc", "sys", "dev"] {
                if names.iter().any(|n| n == forbidden) {
                    return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read_dir of /.. exposed host path {forbidden}"));
                }
            }
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: /.. did not expose host paths");
        Ok(())
    }
}

// CMPTST-08: rename across buckets writes the payload byte-for-byte at the destination and removes the source.
pub(crate) mod cmptst_08 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-08";

    pub(crate) async fn run_rename_cross_bucket(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: rename across buckets writes content at destination and removes source");
        let bucket_a = "complxbucketa";
        let bucket_b = "complxbucketb";
        let path_a = format!("/{bucket_a}");
        let path_b = format!("/{bucket_b}");
        sftp.create_dir(&path_a).await?;
        sftp.create_dir(&path_b).await?;

        let source = format!("/{bucket_a}/cross.txt");
        let dest = format!("/{bucket_b}/cross.txt");
        let content = b"cross-bucket payload\n";
        let mut wf = sftp
            .open_with_flags(&source, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        sftp.rename(&source, &dest).await?;

        let read_back = sftp_read_full(sftp, &dest).await?;
        if read_back != content {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} cross-bucket rename payload mismatch"));
        }
        let entries_a: Vec<String> = sftp.read_dir(&path_a).await?.map(|e| e.file_name()).collect();
        if entries_a.iter().any(|n| n == "cross.txt") {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} source object must be gone after cross-bucket rename, entries were {entries_a:?}"
            ));
        }

        sftp.remove_file(&dest).await?;
        sftp.remove_dir(&path_a).await?;
        sftp.remove_dir(&path_b).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: cross-bucket rename wrote content at destination and removed source");
        Ok(())
    }
}

// CMPTST-09: a path with embedded spaces round-trips via russh-sftp.
pub(crate) mod cmptst_09 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-09";

    pub(crate) async fn run_path_with_spaces_round_trip(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: path with embedded spaces round-trips");
        let bucket = "complspacebucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/file with spaces.txt");
        let content = b"spaces in the key\n";
        let mut wf = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let read_back = sftp_read_full(sftp, &path).await?;
        if read_back != content {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} path-with-spaces round-trip payload mismatch"));
        }

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: path with embedded spaces round-tripped");
        Ok(())
    }
}

// CMPTST-10: read_link is rejected (S3 storage has no symlinks).
pub(crate) mod cmptst_10 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-10";

    pub(crate) async fn run_readlink_rejected(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read_link is rejected");
        let result = sftp.read_link("/anything").await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read_link must error"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read_link rejected");
        Ok(())
    }
}

// CMPTST-11: SETSTAT on path and FSETSTAT on a separate handle both return ok.
pub(crate) mod cmptst_11 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-11";

    pub(crate) async fn run_setstat_after_put_returns_ok(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: SETSTAT on path and FSETSTAT on a separate handle both return ok");
        let bucket = "complsetstatbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/setstat.txt");
        let content = b"SETSTAT after put\n";
        let mut wf = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let path_attrs = FileAttributes {
            permissions: Some(0o644),
            mtime: Some(1_700_000_000),
            ..FileAttributes::default()
        };
        sftp.set_metadata(&path, path_attrs).await?;

        let mut read_handle = sftp.open_with_flags(&path, OpenFlags::READ).await?;
        let handle_attrs = FileAttributes {
            permissions: Some(0o600),
            mtime: Some(1_700_000_001),
            ..FileAttributes::default()
        };
        read_handle.set_metadata(handle_attrs).await?;
        read_handle.shutdown().await?;

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: SETSTAT on path and FSETSTAT on a separate handle both returned ok");
        Ok(())
    }
}

// CMPTST-12: rename to the same path leaves the file in place with original payload.
pub(crate) mod cmptst_12 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-12";

    pub(crate) async fn run_rename_same_path_keeps_file(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: rename to the same path leaves the file in place");
        let bucket = "complrenameselfbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/keep.txt");
        let content = b"do not lose me\n";
        let mut wf = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        sftp.rename(&path, &path).await?;

        let read_back = sftp_read_full(sftp, &path).await?;
        if read_back != content {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} same-path rename lost content"));
        }

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: same-path rename left the file in place");
        Ok(())
    }
}

// CMPTST-13: implicit-directory round-trip from a nested-key upload.
pub(crate) mod cmptst_13 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-13";

    pub(crate) async fn run_implicit_dir_round_trip(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: implicit-directory round-trip");
        let bucket = "compli4bucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let inner_path = format!("/{bucket}/implicit/file.txt");
        let content = b"implicit subdir payload\n";
        let mut wf = sftp
            .open_with_flags(&inner_path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;
        wf.write_all(content).await?;
        wf.flush().await?;
        wf.shutdown().await?;

        let implicit_dir = format!("/{bucket}/implicit");
        let entries_a: Vec<String> = sftp.read_dir(&implicit_dir).await?.map(|e| e.file_name()).collect();
        if !entries_a.iter().any(|n| n == "file.txt") {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} read_dir of the implicit sub-directory must list file.txt, got {entries_a:?}"
            ));
        }

        let entries_b: Vec<String> = sftp
            .read_dir(&format!("{implicit_dir}/"))
            .await?
            .map(|e| e.file_name())
            .collect();
        if !entries_b.iter().any(|n| n == "file.txt") {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} read_dir of the trailing-slash form must list file.txt, got {entries_b:?}"
            ));
        }

        let entries_c: Vec<String> = sftp.read_dir(&bucket_path).await?.map(|e| e.file_name()).collect();
        if !entries_c.iter().any(|n| n == "implicit") {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} read_dir of the bucket must list the implicit sub-directory entry, got {entries_c:?}"
            ));
        }

        let read_back = sftp_read_full(sftp, &inner_path).await?;
        if read_back != content {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} implicit-directory file payload mismatch"));
        }

        let stat = sftp.metadata(&inner_path).await?;
        if stat.size != Some(content.len() as u64) {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} implicit-directory file stat size mismatch"));
        }
        if !stat.file_type().is_file() {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} implicit-directory file stat must report a regular file"
            ));
        }

        sftp.remove_file(&inner_path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: implicit-directory round-trip");
        Ok(())
    }
}

// CMPTST-14: WinSCP-style OPEN, WRITE, FSETSTAT, CLOSE on the same handle returns ok.
pub(crate) mod cmptst_14 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-14";

    pub(crate) async fn run_winscp_setstat_shape_on_handle(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: OPEN + WRITE + FSETSTAT + CLOSE on the same handle returns ok");
        let bucket = "complwinscpbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/winscp.txt");
        let content = b"winscp packet sequence payload\n";
        let handle = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await?;

        let mut writer = handle;
        writer.write_all(content).await?;
        writer.flush().await?;

        let attrs = FileAttributes {
            permissions: Some(0o644),
            mtime: Some(1_700_000_002),
            ..FileAttributes::default()
        };
        writer.set_metadata(attrs).await?;

        writer.shutdown().await?;

        let read_back = sftp_read_full(sftp, &path).await?;
        if read_back != content {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} WinSCP packet-sequence payload mismatch"));
        }

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: OPEN + WRITE + FSETSTAT + CLOSE on the same handle returned ok");
        Ok(())
    }
}

// CMPTST-15: put through SFTP is rejected in read-only mode.
pub(crate) mod cmptst_15 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-15";

    pub(crate) async fn run_ro_put_rejected(sftp: &SftpSession, bucket: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects put");
        let path = format!("/{bucket}/blocked.txt");
        let result = sftp
            .open_with_flags(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE)
            .await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject open-for-write"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected put");
        Ok(())
    }
}

// CMPTST-16: rm through SFTP is rejected in read-only mode.
pub(crate) mod cmptst_16 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-16";

    pub(crate) async fn run_ro_rm_rejected(sftp: &SftpSession, bucket: &str, seeded_key: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects rm");
        let path = format!("/{bucket}/{seeded_key}");
        let result = sftp.remove_file(&path).await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject remove_file"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected rm");
        Ok(())
    }
}

// CMPTST-17: mkdir through SFTP is rejected in read-only mode.
pub(crate) mod cmptst_17 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-17";

    pub(crate) async fn run_ro_mkdir_rejected(sftp: &SftpSession) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects mkdir");
        let result = sftp.create_dir("/ronewbucket").await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject mkdir"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected mkdir");
        Ok(())
    }
}

// CMPTST-18: rmdir through SFTP is rejected in read-only mode.
pub(crate) mod cmptst_18 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-18";

    pub(crate) async fn run_ro_rmdir_rejected(sftp: &SftpSession, bucket: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects rmdir");
        let path = format!("/{bucket}");
        let result = sftp.remove_dir(&path).await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject remove_dir"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected rmdir");
        Ok(())
    }
}

// CMPTST-19: rename through SFTP is rejected in read-only mode.
pub(crate) mod cmptst_19 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-19";

    pub(crate) async fn run_ro_rename_rejected(sftp: &SftpSession, bucket: &str, seeded_key: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects rename");
        let from = format!("/{bucket}/{seeded_key}");
        let to = format!("/{bucket}/moved.txt");
        let result = sftp.rename(&from, &to).await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject rename"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected rename");
        Ok(())
    }
}

// CMPTST-20: ls through SFTP is allowed in read-only mode and lists the seeded bucket.
pub(crate) mod cmptst_20 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-20";

    pub(crate) async fn run_ro_ls_allowed(sftp: &SftpSession, bucket: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode allows ls");
        let entries: Vec<String> = sftp.read_dir("/").await?.map(|e| e.file_name()).collect();
        if !entries.iter().any(|n| n == bucket) {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must list buckets, expected {bucket}, got {entries:?}"
            ));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode allowed ls");
        Ok(())
    }
}

// CMPTST-21: get through SFTP is allowed in read-only mode and returns the seeded payload.
pub(crate) mod cmptst_21 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-21";

    pub(crate) async fn run_ro_get_allowed(sftp: &SftpSession, bucket: &str, seeded_key: &str, expected: &[u8]) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode allows get");
        let path = format!("/{bucket}/{seeded_key}");
        let read_back = sftp_read_full(sftp, &path).await?;
        if read_back != expected {
            return Err(anyhow!(
                "{COMPLIANCE_TEST_OUTPUT_ID} read-only mode get returned {} bytes, expected {}",
                read_back.len(),
                expected.len()
            ));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode allowed get and returned the seeded payload");
        Ok(())
    }
}

// CMPTST-22: SETSTAT on a path is rejected with PermissionDenied in read-only mode.
pub(crate) mod cmptst_22 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-22";

    pub(crate) async fn run_ro_setstat_rejected(sftp: &SftpSession, bucket: &str, seeded_key: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects SETSTAT on a path");
        let path = format!("/{bucket}/{seeded_key}");
        let attrs = FileAttributes {
            permissions: Some(0o600),
            mtime: Some(1_700_000_000),
            ..FileAttributes::default()
        };
        let result = sftp.set_metadata(&path, attrs).await;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject SETSTAT"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected SETSTAT on a path");
        Ok(())
    }
}

// CMPTST-23: FSETSTAT on a read handle is rejected with PermissionDenied in read-only mode.
pub(crate) mod cmptst_23 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-23";

    pub(crate) async fn run_ro_fsetstat_rejected(sftp: &SftpSession, bucket: &str, seeded_key: &str) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejects FSETSTAT on an open handle");
        let path = format!("/{bucket}/{seeded_key}");
        let mut handle = sftp.open_with_flags(&path, OpenFlags::READ).await?;
        let attrs = FileAttributes {
            permissions: Some(0o600),
            mtime: Some(1_700_000_001),
            ..FileAttributes::default()
        };
        let result = handle.set_metadata(attrs).await;
        handle.shutdown().await?;
        if result.is_ok() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read-only mode must reject FSETSTAT"));
        }
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: read-only mode rejected FSETSTAT on an open handle");
        Ok(())
    }
}

// CMPTST-24: concurrent half-close burst does not leak server-side session tasks.
#[cfg(target_os = "linux")]
pub(crate) mod cmptst_24 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-24";

    // Half-close zombie regression ports. Pair held distinct from the
    // other SFTP test entries so the half-close traffic stays off the
    // shared listener and so the assertion-time CLOSE_WAIT scan only
    // counts connections this entry opened.
    const HALF_CLOSE_SFTP_PORT: u16 = 9026;
    const HALF_CLOSE_SFTP_ADDRESS: &str = "127.0.0.1:9026";
    const HALF_CLOSE_S3_ADDRESS: &str = "127.0.0.1:9302";
    const HALF_CLOSE_S3_ENDPOINT: &str = "http://127.0.0.1:9302";
    const HALF_CLOSE_S3_READY_ATTEMPTS: u32 = 30;
    // Per-session deadline the spawned server uses. Short enough that
    // the post-fix kill path completes well inside the 30 s wait.
    const HALF_CLOSE_IDLE_TIMEOUT_SECS: u64 = 8;
    // Window the test waits after triggering the N half-close peers.
    // Long enough that the post-fix server-side deadline has fired and
    // the session task has finished.
    const HALF_CLOSE_WAIT_SECS: u64 = 30;
    // Concurrent half-close client count. FileZilla 3.66.5 was observed
    // at 17 parallel sessions in the real-world capture. Eight is
    // enough to reproduce the leak under the same shape and keeps test
    // runtime bounded.
    const HALF_CLOSE_PARALLEL_SESSIONS: usize = 8;
    // Fixture file size. Larger than one MAX_READ_LEN chunk so the
    // test session can complete one full READ before triggering the
    // half-close.
    const HALF_CLOSE_FIXTURE_BYTES: usize = 1024 * 1024;
    // One MAX_READ_LEN chunk. The case contract requires at least one
    // READ packet to complete before the half-close trigger.
    const HALF_CLOSE_FIRST_READ_BYTES: usize = 256 * 1024;

    // Shared flags between the test loop and the per-session
    // HalfClosableStream instance handed to the russh client. The
    // wrapper polls these flags from inside the russh I/O task to flip
    // the underlying TCP socket into the half-closed-write state and
    // to suspend further reads.
    struct HalfCloseControl {
        half_close_writes: AtomicBool,
        block_reads: AtomicBool,
    }

    impl HalfCloseControl {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                half_close_writes: AtomicBool::new(false),
                block_reads: AtomicBool::new(false),
            })
        }
    }

    /// Wrapper around a tokio::net::TcpStream split into owned halves
    /// so the test can request a one-sided shutdown (FIN on the write
    /// side, no further reads acknowledged) while the russh client
    /// remains the I/O owner. The control flags are toggled by the
    /// test loop after the first SFTP READ packet completes.
    ///
    /// The wrapper deliberately returns Poll::Pending after the FIN is
    /// on the wire instead of an io::Error: the russh client task must
    /// remain suspended on the wrapper rather than tearing the SSH
    /// session down, which would full-close the socket and reset the
    /// OS state the test is asserting against.
    struct HalfClosableStream {
        read: OwnedReadHalf,
        write: OwnedWriteHalf,
        control: Arc<HalfCloseControl>,
        write_shutdown_done: bool,
    }

    impl HalfClosableStream {
        fn from_tcp(stream: TcpStream, control: Arc<HalfCloseControl>) -> Self {
            let (read, write) = stream.into_split();
            Self {
                read,
                write,
                control,
                write_shutdown_done: false,
            }
        }
    }

    impl AsyncRead for HalfClosableStream {
        fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            if this.control.block_reads.load(Ordering::Relaxed) {
                return Poll::Pending;
            }
            Pin::new(&mut this.read).poll_read(cx, buf)
        }
    }

    impl AsyncWrite for HalfClosableStream {
        fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
            let this = self.get_mut();
            if this.control.half_close_writes.load(Ordering::Relaxed) {
                if !this.write_shutdown_done {
                    match Pin::new(&mut this.write).poll_shutdown(cx) {
                        Poll::Ready(Ok(())) => {
                            this.write_shutdown_done = true;
                        }
                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                        Poll::Pending => return Poll::Pending,
                    }
                }
                return Poll::Pending;
            }
            Pin::new(&mut this.write).poll_write(cx, buf)
        }

        fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            if this.control.half_close_writes.load(Ordering::Relaxed) && this.write_shutdown_done {
                return Poll::Pending;
            }
            Pin::new(&mut this.write).poll_flush(cx)
        }

        fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            Pin::new(&mut this.write).poll_shutdown(cx)
        }
    }

    /// Drive a single half-close session against the running server:
    /// open the TCP, hand the socket halves to a HalfClosableStream,
    /// run the SSH+SFTP handshake through russh, read one MAX_READ_LEN
    /// chunk, then flip the control flags and issue a follow-up SFTP
    /// request so the russh client task touches poll_write once and
    /// the wrapper has the chance to drive the TCP shutdown(SHUT_WR)
    /// syscall before the request is suspended.
    ///
    /// Returns the still-live russh client Handle and the SftpSession
    /// alongside the control handle. The caller holds them in a Vec to
    /// prevent the russh client task from being dropped, which would
    /// otherwise full-close the socket and mask the leak the test is
    /// probing for.
    async fn drive_half_close_session(
        address: &str,
        bucket: &str,
        seeded_key: &str,
    ) -> Result<(client::Handle<AcceptAnyServerKey>, SftpSession, Arc<HalfCloseControl>)> {
        let tcp = TcpStream::connect(address)
            .await
            .map_err(|e| anyhow!("TCP connect to {address} failed: {e}"))?;
        let control = HalfCloseControl::new();
        let stream = HalfClosableStream::from_tcp(tcp, Arc::clone(&control));
        let config = Arc::new(client::Config::default());
        let mut session = client::connect_stream(config, stream, AcceptAnyServerKey)
            .await
            .map_err(|e| anyhow!("russh connect_stream failed: {e}"))?;

        let auth = session
            .authenticate_password(DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY)
            .await
            .map_err(|e| anyhow!("russh password auth failed: {e}"))?;
        if !auth.success() {
            return Err(anyhow!("SFTP password auth rejected on half-close session"));
        }
        let channel = session
            .channel_open_session()
            .await
            .map_err(|e| anyhow!("channel open failed: {e}"))?;
        channel
            .request_subsystem(true, "sftp")
            .await
            .map_err(|e| anyhow!("subsystem request failed: {e}"))?;
        let sftp = SftpSession::new(channel.into_stream())
            .await
            .map_err(|e| anyhow!("SftpSession init failed: {e}"))?;

        let path = format!("/{bucket}/{seeded_key}");
        let mut file = sftp
            .open_with_flags(&path, OpenFlags::READ)
            .await
            .map_err(|e| anyhow!("SFTP open failed: {e}"))?;
        let mut buf = vec![0u8; HALF_CLOSE_FIRST_READ_BYTES];
        let mut read_total = 0usize;
        while read_total < HALF_CLOSE_FIRST_READ_BYTES {
            let n = file
                .read(&mut buf[read_total..])
                .await
                .map_err(|e| anyhow!("SFTP read failed: {e}"))?;
            if n == 0 {
                break;
            }
            read_total += n;
        }
        if read_total < HALF_CLOSE_FIRST_READ_BYTES {
            return Err(anyhow!(
                "first SFTP READ packet returned only {read_total} bytes, expected at least {HALF_CLOSE_FIRST_READ_BYTES}"
            ));
        }

        // Flip the half-close trigger before the next SFTP request goes
        // out and stop draining the receive side. The next inflight
        // request forces the russh client task to call poll_write on
        // the wrapper, which drives the underlying TCP shutdown(SHUT_WR)
        // and then suspends. After the FIN has been sent the russh task
        // remains parked rather than tearing the SSH session down, so
        // the OS-level socket stays in the half-closed state the leak
        // depends on.
        control.half_close_writes.store(true, Ordering::Relaxed);
        control.block_reads.store(true, Ordering::Relaxed);
        let _ = tokio::time::timeout(Duration::from_millis(750), file.metadata()).await;
        drop(file);

        Ok((session, sftp, control))
    }

    // Test orchestration:
    //
    // 1. Spawn rustfs with a short idle timeout (HALF_CLOSE_IDLE_TIMEOUT_SECS).
    // 2. Open HALF_CLOSE_PARALLEL_SESSIONS SFTP sessions over a custom
    //    HalfClosableStream that issues shutdown(SHUT_WR) on the read
    //    half mid-transfer and parks subsequent reads (returns Pending).
    // 3. Wait HALF_CLOSE_WAIT_SECS for the server-side idle timer to
    //    fire and the accept loop to drain finished session tasks.
    // 4. Issue dummy TCP connects to wake the accept loop's select so
    //    the JoinSet flushes finished tasks before the assertion runs.
    // 5. Assert the entered/finished session counters balance and that
    //    no CLOSE_WAIT sockets remain on the bind port (Linux ss(8)
    //    only; the assertion skips with a warn if ss is unavailable).
    pub(crate) async fn run_concurrent_half_close_no_leak() -> Result<()> {
        let env = ProtocolTestEnvironment::new().map_err(|e| anyhow!("{}", e))?;
        let host_key_dir = PathBuf::from(&env.temp_dir).join("sftp_host_keys");
        generate_host_key(&host_key_dir).await?;

        info!(
            "{COMPLIANCE_TEST_OUTPUT_ID}: starting half-close server on {} (idle_timeout={}s)",
            HALF_CLOSE_SFTP_ADDRESS, HALF_CLOSE_IDLE_TIMEOUT_SECS
        );
        let binary_path = rustfs_binary_path_with_features(Some("ftps,webdav,sftp"));
        let host_key_dir_str = host_key_dir
            .to_str()
            .ok_or_else(|| anyhow!("host key dir path is not utf-8: {}", host_key_dir.display()))?;
        let mut server_process = ServerProcess::new(
            Command::new(&binary_path)
                .env(ENV_SFTP_ENABLE, "true")
                .env(ENV_SFTP_ADDRESS, HALF_CLOSE_SFTP_ADDRESS)
                .env(ENV_SFTP_HOST_KEY_DIR, host_key_dir_str)
                .env(ENV_SFTP_READ_ONLY, "false")
                .env(ENV_SFTP_PART_SIZE, PART_SIZE_ENV)
                .env(ENV_SFTP_IDLE_TIMEOUT, HALF_CLOSE_IDLE_TIMEOUT_SECS.to_string())
                .env(ENV_RUSTFS_ADDRESS, HALF_CLOSE_S3_ADDRESS)
                // Disable the admin console listener to avoid port
                // contention with local dev-testing containers.
                .env(ENV_CONSOLE_ENABLE, "false")
                .env("RUSTFS_OBS_LOGGER_LEVEL", "rustfs_protocols=debug")
                .env("RUST_LOG", "rustfs_protocols=debug")
                .stdout(Stdio::piped())
                .arg(&env.temp_dir)
                .spawn()?,
        );
        let counters = SessionCounters::new();
        watch_session_lifecycle_events(server_process.child_mut(), Arc::clone(&counters));

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(HALF_CLOSE_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(HALF_CLOSE_S3_ENDPOINT);
            wait_for_s3_ready(&s3, HALF_CLOSE_S3_READY_ATTEMPTS).await?;

            let bucket = "halfclose";
            let seeded_key = "fixture.bin";
            s3.create_bucket()
                .bucket(bucket)
                .send()
                .await
                .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;
            let payload: Vec<u8> = (0..HALF_CLOSE_FIXTURE_BYTES).map(|i| (i as u8).wrapping_mul(7)).collect();
            s3.put_object()
                .bucket(bucket)
                .key(seeded_key)
                .body(ByteStream::from(payload))
                .send()
                .await
                .map_err(|e| anyhow!("S3 PutObject {bucket}/{seeded_key} failed: {e:?}"))?;

            let mut futs = Vec::with_capacity(HALF_CLOSE_PARALLEL_SESSIONS);
            for i in 0..HALF_CLOSE_PARALLEL_SESSIONS {
                let address = HALF_CLOSE_SFTP_ADDRESS.to_string();
                let bucket = bucket.to_string();
                let key = seeded_key.to_string();
                futs.push(tokio::spawn(async move {
                    drive_half_close_session(&address, &bucket, &key)
                        .await
                        .map_err(|e| anyhow!("session {i} setup failed: {e}"))
                }));
            }
            // Hold each (Handle, SftpSession, Control) tuple for the
            // full wait window so the OwnedRead/OwnedWriteHalf inside
            // the wrapper stay alive and the OS keeps each socket in
            // its half-closed state. Dropping any of them would trigger
            // a full-close on the socket, which would mask the leak by
            // waking the server's session task through a real EOF or
            // RST.
            let mut keepalive: Vec<(client::Handle<AcceptAnyServerKey>, SftpSession, Arc<HalfCloseControl>)> = Vec::new();
            for fut in futs {
                keepalive.push(fut.await??);
            }

            let entered_after_setup = counters.entered.load(Ordering::Relaxed);
            let finished_after_setup = counters.finished.load(Ordering::Relaxed);
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: {} half-close sessions established (server entered={}, finished={}). Waiting {} s for the watchdog to kill them",
                HALF_CLOSE_PARALLEL_SESSIONS, entered_after_setup, finished_after_setup, HALF_CLOSE_WAIT_SECS,
            );

            sleep(Duration::from_secs(HALF_CLOSE_WAIT_SECS)).await;

            // The accept loop drains finished session tasks at the top
            // of every iteration, which only runs when a new TCP accept
            // (or a shutdown signal) wakes the select. Issue a single
            // TCP connection so the loop iterates once and the JoinSet
            // drain emits the "SFTP session task finished" log for
            // every session that the per-session deadline has already
            // canceled. Without this, the counters under-report on a
            // quiet server.
            for _ in 0..3 {
                if let Ok(stream) = TcpStream::connect(HALF_CLOSE_SFTP_ADDRESS).await {
                    drop(stream);
                }
                sleep(Duration::from_millis(200)).await;
            }
            sleep(Duration::from_millis(500)).await;

            let entered = counters.entered.load(Ordering::Relaxed);
            let finished = counters.finished.load(Ordering::Relaxed);
            let outstanding = entered.saturating_sub(finished);
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: post-wait counters entered={} finished={} outstanding={}",
                entered, finished, outstanding
            );
            if outstanding > 1 {
                return Err(anyhow!(
                    "{COMPLIANCE_TEST_OUTPUT_ID} session-task balance contract failed: entered={entered} finished={finished} outstanding={outstanding}, expected at most 1"
                ));
            }

            match count_close_wait_on_port(HALF_CLOSE_SFTP_PORT).await? {
                Some(0) => info!("{COMPLIANCE_TEST_OUTPUT_ID}: zero CLOSE_WAIT entries against port {HALF_CLOSE_SFTP_PORT}"),
                Some(n) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} {n} CLOSE_WAIT entries against port {HALF_CLOSE_SFTP_PORT}, expected 0"
                    ));
                }
                None => info!("{COMPLIANCE_TEST_OUTPUT_ID}: ss(8) unavailable, skipping CLOSE_WAIT assertion"),
            }

            // Drop the keepalive vector now so the test process does
            // not leave the half-closed sockets dangling past the
            // assertion.
            drop(keepalive);
            info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: half-close burst did not leak server-side session tasks");
            Ok::<(), anyhow::Error>(())
        }
        .await;

        server_process.kill_and_wait().await;

        result
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_concurrent_half_close_no_leak()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-25: wedge-kill watchdog kills sessions parked behind a CLOSE_WAIT socket.
#[cfg(target_os = "linux")]
pub(crate) mod cmptst_25 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-25";

    // Wedge-kill regression ports. Distinct from the half-close ports
    // so the wedge-driving traffic stays off the half-close listener
    // and so the post-wait CLOSE_WAIT scan only counts sockets this
    // entry opened.
    const WEDGE_SFTP_PORT: u16 = 9027;
    const WEDGE_SFTP_ADDRESS: &str = "127.0.0.1:9027";
    const WEDGE_S3_ADDRESS: &str = "127.0.0.1:9303";
    const WEDGE_S3_ENDPOINT: &str = "http://127.0.0.1:9303";
    const WEDGE_S3_READY_ATTEMPTS: u32 = 30;

    // Idle timeout the spawned server uses. Set well above the wait
    // window so russh's own inactivity_timeout cannot kill any session
    // during the test. The contract: only the watchdog kills the
    // wedged session inside the wait window. Without the watchdog the
    // session leaks because the russh select! is parked outside its
    // own arms.
    const WEDGE_IDLE_TIMEOUT_SECS: u64 = 300;

    // Total wait window. Must exceed
    // WEDGE_FAST_KILL_SILENCE_SECS (30) + WEDGE_WATCHDOG_TICK_SECS (15)
    // = 45 s of worst-case watchdog detection latency, plus a 15 s
    // grace.
    // 90 s instead of 60 s gives a 30 s margin above the watchdog
    // worst-case cancel latency (FAST_KILL_SILENCE 30 s plus two
    // 15 s ticks = 60 s) so scheduler jitter on a busy CI host does
    // not flip the assertion.
    const WEDGE_WAIT_SECS: u64 = 90;

    // Concurrent wedged sessions. Mirrors the half-close case so
    // server-side bookkeeping counters move in the same magnitude
    // regardless of which case runs.
    const WEDGE_PARALLEL_SESSIONS: usize = 8;

    // Fixture file size. Large enough that 8 pipelined READ requests
    // of 256 KiB each fit inside it without overrunning end-of-file.
    const WEDGE_FIXTURE_BYTES: usize = 4 * 1024 * 1024;

    // Fixture chunk size requested by the test's pipelined READs.
    // Matches MAX_READ_LEN so the server's response is one full chunk
    // per request.
    const WEDGE_CHUNK_BYTES: u32 = 256 * 1024;

    // Pipelined READs sent in the window-exhaustion phase. Eight times
    // 256 KiB equals 2 MiB, which equals russh's default window_size,
    // so the server's stream.write_all parks at the SSH window the
    // moment the eighth response is queued.
    const WEDGE_WINDOW_FILL_READS: usize = 8;

    // Pipelined READs sent in the mpsc-fill phase, after the SSH
    // window has been exhausted. Above the russh server-side
    // channel_buffer_size = 100 default so the per-channel mpsc fills
    // and the session loop's chan.send().await parks. 200 picks a
    // comfortable margin without blowing up the test wire footprint
    // (200 times ~30 B per FXP_READ packet ~ 6 KiB).
    const WEDGE_MPSC_FILL_READS: usize = 200;

    // SFTPv3 packet type codes used by the raw-protocol path the wedge
    // driver follows. The driver hand-builds FXP_INIT, FXP_OPEN, and
    // FXP_READ packets via channel.data() rather than going through
    // russh-sftp's high-level File API because SftpSession serialises
    // reads (one outstanding request at a time) and the wedge requires
    // pipelining many READs without waiting for responses.
    const SSH_FXP_INIT: u8 = 1;
    const SSH_FXP_OPEN: u8 = 3;
    const SSH_FXP_READ: u8 = 5;
    // SSH_FXP_OPEN flags. READ-only access against the seeded fixture.
    const SSH_FXF_READ: u32 = 0x0000_0001;

    // Shared flags between the test loop and the per-session
    // WedgeStream. The wrapper polls these flags from inside the russh
    // I/O task to suspend wire reads (so the per-channel mpsc on the
    // server fills) and to land FIN on the wire (so the kernel reports
    // the socket in CLOSE_WAIT after the SFTP driver also stops
    // draining on its own).
    struct WedgeControl {
        block_reads: AtomicBool,
        half_close_writes: AtomicBool,
    }

    impl WedgeControl {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                block_reads: AtomicBool::new(false),
                half_close_writes: AtomicBool::new(false),
            })
        }
    }

    /// Wrapper around tokio::net::TcpStream for the wedge regression.
    /// Same shape as the half-close wrapper but its purpose is to keep
    /// the russh client task wedged once block_reads is set so the test
    /// can pile in further FXP_READ requests via the still-live write
    /// half. Once half_close_writes is set the wrapper drives
    /// shutdown(SHUT_WR) on the next poll_write, sending FIN to the
    /// server.
    struct WedgeStream {
        read: OwnedReadHalf,
        write: OwnedWriteHalf,
        control: Arc<WedgeControl>,
        write_shutdown_done: bool,
    }

    impl WedgeStream {
        fn from_tcp(stream: TcpStream, control: Arc<WedgeControl>) -> Self {
            let (read, write) = stream.into_split();
            Self {
                read,
                write,
                control,
                write_shutdown_done: false,
            }
        }
    }

    impl AsyncRead for WedgeStream {
        fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            if this.control.block_reads.load(Ordering::Relaxed) {
                return Poll::Pending;
            }
            Pin::new(&mut this.read).poll_read(cx, buf)
        }
    }

    impl AsyncWrite for WedgeStream {
        fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
            let this = self.get_mut();
            if this.control.half_close_writes.load(Ordering::Relaxed) {
                if !this.write_shutdown_done {
                    match Pin::new(&mut this.write).poll_shutdown(cx) {
                        Poll::Ready(Ok(())) => {
                            this.write_shutdown_done = true;
                        }
                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                        Poll::Pending => return Poll::Pending,
                    }
                }
                return Poll::Pending;
            }
            Pin::new(&mut this.write).poll_write(cx, buf)
        }

        fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            if this.control.half_close_writes.load(Ordering::Relaxed) && this.write_shutdown_done {
                return Poll::Pending;
            }
            Pin::new(&mut this.write).poll_flush(cx)
        }

        fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            Pin::new(&mut this.write).poll_shutdown(cx)
        }
    }

    /// Drive a single wedge session against the running server.
    ///
    /// Returns the still-live russh client Handle, channel, and control
    /// flags. The caller holds them in a Vec for the entire wait window
    /// so the underlying TCP socket stays in CLOSE_WAIT and the
    /// per-channel mpsc on the server stays full.
    async fn drive_wedge_session(
        address: &str,
        bucket: &str,
        seeded_key: &str,
    ) -> Result<(client::Handle<AcceptAnyServerKey>, russh::Channel<russh::client::Msg>, Arc<WedgeControl>)> {
        let tcp = TcpStream::connect(address)
            .await
            .map_err(|e| anyhow!("TCP connect to {address} failed: {e}"))?;
        let control = WedgeControl::new();
        let stream = WedgeStream::from_tcp(tcp, Arc::clone(&control));
        let config = Arc::new(client::Config::default());
        let mut session = client::connect_stream(config, stream, AcceptAnyServerKey)
            .await
            .map_err(|e| anyhow!("russh connect_stream failed: {e}"))?;

        let auth = session
            .authenticate_password(DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY)
            .await
            .map_err(|e| anyhow!("russh password auth failed: {e}"))?;
        if !auth.success() {
            return Err(anyhow!("SFTP password auth rejected on wedge session"));
        }
        let mut channel = session
            .channel_open_session()
            .await
            .map_err(|e| anyhow!("channel open failed: {e}"))?;
        channel
            .request_subsystem(true, "sftp")
            .await
            .map_err(|e| anyhow!("subsystem request failed: {e}"))?;

        // FXP_INIT (version 3). One u32 payload (the version).
        let init_pkt = build_sftp_init();
        channel
            .data(&init_pkt[..])
            .await
            .map_err(|e| anyhow!("FXP_INIT send failed: {e:?}"))?;
        // Drain the FXP_VERSION response so the channel is in steady
        // state before the wedge flags are toggled. The russh client
        // receive loop delivers it via channel.wait().
        let _ = wait_for_data(&mut channel).await?;

        // FXP_OPEN against the seeded fixture. Returns FXP_HANDLE with
        // the server-assigned handle string fed to every following READ.
        let path = format!("/{bucket}/{seeded_key}");
        let open_pkt = build_sftp_open(1, &path, SSH_FXF_READ);
        channel
            .data(&open_pkt[..])
            .await
            .map_err(|e| anyhow!("FXP_OPEN send failed: {e:?}"))?;
        let handle = parse_handle(&wait_for_data(&mut channel).await?)?;

        // One FXP_READ to confirm the path works end-to-end before the
        // wedge phase. Drain the response so subsequent reads do not
        // see stale FXP_DATA on the wire.
        let probe_read = build_sftp_read(2, &handle, 0, WEDGE_CHUNK_BYTES);
        channel
            .data(&probe_read[..])
            .await
            .map_err(|e| anyhow!("probe FXP_READ send failed: {e:?}"))?;
        let _ = wait_for_data(&mut channel).await?;

        // Wedge phase one: stop draining the wire on the client side,
        // then pipeline N FXP_READ packets that fill the SSH window.
        // The server queues FXP_DATA responses for each. The responses
        // leave the server's stream.write_all only as long as the SSH
        // receive window has slack. After WEDGE_WINDOW_FILL_READS
        // responses the server's next stream.write_all parks because
        // the client is no longer sending CHANNEL_WINDOW_ADJUST.
        control.block_reads.store(true, Ordering::Relaxed);
        let mut req_id = 3u32;
        for i in 0..WEDGE_WINDOW_FILL_READS {
            let offset = (i as u64) * WEDGE_CHUNK_BYTES as u64;
            let pkt = build_sftp_read(req_id, &handle, offset, WEDGE_CHUNK_BYTES);
            channel
                .data(&pkt[..])
                .await
                .map_err(|e| anyhow!("FXP_READ window-fill send failed at i={i}: {e:?}"))?;
            req_id = req_id.wrapping_add(1);
        }

        // Wedge phase two: pile in further FXP_READ packets while the
        // SFTP driver is parked on stream.write_all. Each arriving
        // CHANNEL_DATA pushes one entry into the server's per-channel
        // mpsc (default capacity 100). Once that mpsc fills, the
        // server's session loop's chan.send().await blocks. The
        // select! is then unreachable from the keepalive and
        // inactivity arms. This is the wedge.
        for i in 0..WEDGE_MPSC_FILL_READS {
            let offset = ((i % WEDGE_WINDOW_FILL_READS) as u64) * WEDGE_CHUNK_BYTES as u64;
            let pkt = build_sftp_read(req_id, &handle, offset, WEDGE_CHUNK_BYTES);
            // Best-effort: once the wire backs up the channel's send
            // buffer fills and channel.data().await yields. Bound the
            // wait so a stalled client side does not block the test.
            match tokio::time::timeout(Duration::from_millis(250), channel.data(&pkt[..])).await {
                Ok(Ok(())) => {}
                Ok(Err(e)) => return Err(anyhow!("FXP_READ mpsc-fill send failed at i={i}: {e:?}")),
                Err(_) => break,
            }
            req_id = req_id.wrapping_add(1);
        }

        // Phase three: trigger the FIN. Setting half_close_writes flips
        // the wrapper into shutdown(SHUT_WR) on the next poll_write.
        // One last FXP_READ drives that poll_write. After this point
        // the wrapper returns Pending forever on writes, so the russh
        // client task remains parked instead of tearing the SSH session
        // down.
        control.half_close_writes.store(true, Ordering::Relaxed);
        let trigger = build_sftp_read(req_id, &handle, 0, WEDGE_CHUNK_BYTES);
        let _ = tokio::time::timeout(Duration::from_millis(750), channel.data(&trigger[..])).await;

        Ok((session, channel, control))
    }

    /// Read one full SFTPv3 packet from the channel. The packet wire
    /// format is length(4) || type(1) || payload, so this accumulates
    /// inbound CHANNEL_DATA frames until the four-byte length prefix
    /// has been satisfied. Returns the full packet bytes including the
    /// length prefix.
    async fn wait_for_data(channel: &mut russh::Channel<russh::client::Msg>) -> Result<Vec<u8>> {
        use russh::ChannelMsg;
        let timeout_per_packet = Duration::from_secs(5);
        let mut buf: Vec<u8> = Vec::new();
        loop {
            if buf.len() >= 4 {
                let declared = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
                if buf.len() >= 4 + declared {
                    return Ok(buf);
                }
            }
            let msg = tokio::time::timeout(timeout_per_packet, channel.wait())
                .await
                .map_err(|_| anyhow!("timed out waiting for SFTP response (have {} bytes)", buf.len()))?
                .ok_or_else(|| anyhow!("channel closed before SFTP packet complete (have {} bytes)", buf.len()))?;
            match msg {
                ChannelMsg::Data { data } => buf.extend_from_slice(&data),
                ChannelMsg::Eof | ChannelMsg::Close => {
                    return Err(anyhow!("channel ended before SFTP packet complete (have {} bytes)", buf.len()));
                }
                _ => {}
            }
        }
    }

    fn build_sftp_init() -> Vec<u8> {
        // Length(4) || Type(1) || Version(4). Packet length excludes
        // the length field itself.
        let mut payload = Vec::with_capacity(9);
        payload.extend_from_slice(&5u32.to_be_bytes());
        payload.push(SSH_FXP_INIT);
        payload.extend_from_slice(&3u32.to_be_bytes());
        payload
    }

    fn build_sftp_open(req_id: u32, path: &str, flags: u32) -> Vec<u8> {
        // Length(4) || Type(1) || ReqId(4) || PathLen(4) || Path ||
        // Flags(4) || AttrFlags(4). SFTPv3 OPEN ends with a
        // FileAttributes block. An empty attrs (flags=0) is one u32.
        let mut body = Vec::new();
        body.push(SSH_FXP_OPEN);
        body.extend_from_slice(&req_id.to_be_bytes());
        body.extend_from_slice(&(path.len() as u32).to_be_bytes());
        body.extend_from_slice(path.as_bytes());
        body.extend_from_slice(&flags.to_be_bytes());
        body.extend_from_slice(&0u32.to_be_bytes()); // empty FileAttributes
        let mut pkt = Vec::with_capacity(4 + body.len());
        pkt.extend_from_slice(&(body.len() as u32).to_be_bytes());
        pkt.extend_from_slice(&body);
        pkt
    }

    fn build_sftp_read(req_id: u32, handle: &[u8], offset: u64, len: u32) -> Vec<u8> {
        // Length(4) || Type(1) || ReqId(4) || HandleLen(4) || Handle ||
        // Offset(8) || Len(4).
        let mut body = Vec::with_capacity(1 + 4 + 4 + handle.len() + 8 + 4);
        body.push(SSH_FXP_READ);
        body.extend_from_slice(&req_id.to_be_bytes());
        body.extend_from_slice(&(handle.len() as u32).to_be_bytes());
        body.extend_from_slice(handle);
        body.extend_from_slice(&offset.to_be_bytes());
        body.extend_from_slice(&len.to_be_bytes());
        let mut pkt = Vec::with_capacity(4 + body.len());
        pkt.extend_from_slice(&(body.len() as u32).to_be_bytes());
        pkt.extend_from_slice(&body);
        pkt
    }

    fn parse_handle(packet: &[u8]) -> Result<Vec<u8>> {
        // Wire layout: Length(4) || Type(1) || ReqId(4) || HandleLen(4)
        // || Handle. For FXP_HANDLE the type byte is 102.
        if packet.len() < 4 + 1 + 4 + 4 {
            return Err(anyhow!("SFTP open response too short: {} bytes", packet.len()));
        }
        let kind = packet[4];
        if kind != 102 {
            return Err(anyhow!("expected FXP_HANDLE (102), got type {kind} from FXP_OPEN reply"));
        }
        let handle_len = u32::from_be_bytes([packet[9], packet[10], packet[11], packet[12]]) as usize;
        if packet.len() < 13 + handle_len {
            return Err(anyhow!(
                "FXP_HANDLE truncated: declared {handle_len} bytes, packet has {} after header",
                packet.len().saturating_sub(13)
            ));
        }
        Ok(packet[13..13 + handle_len].to_vec())
    }

    // Test orchestration:
    //
    // 1. Spawn rustfs with a long idle_timeout (300 s) so the test
    //    isolates the watchdog kill path from the inactivity timer.
    // 2. Open WEDGE_PARALLEL_SESSIONS SFTP sessions over a custom
    //    WedgeStream that allows writes but parks reads after a flag
    //    flips. Hand-build raw FXP_INIT, FXP_OPEN, FXP_READ packets to
    //    fill the SSH per-channel window plus the per-channel mpsc on
    //    the server (WEDGE_WINDOW_FILL_READS + WEDGE_MPSC_FILL_READS),
    //    so the server's send loop parks on the mpsc.
    // 3. Issue shutdown(SHUT_WR) on the client side to drive the
    //    socket into CLOSE_WAIT.
    // 4. Wait WEDGE_WAIT_SECS for the watchdog (FAST_KILL_SILENCE 30 s
    //    plus two 15 s ticks worst-case = 60 s) to detect CLOSE_WAIT
    //    via /proc/net/tcp and cancel the parked session.
    // 5. Assert the session task counters balance and CLOSE_WAIT count
    //    is zero (ss(8) only; skips with a warn when ss is missing).
    pub(crate) async fn run_wedge_kill_after_silence_in_close_wait() -> Result<()> {
        let env = ProtocolTestEnvironment::new().map_err(|e| anyhow!("{}", e))?;
        let host_key_dir = PathBuf::from(&env.temp_dir).join("sftp_host_keys");
        generate_host_key(&host_key_dir).await?;

        info!(
            "{COMPLIANCE_TEST_OUTPUT_ID}: starting wedge server on {} (idle_timeout={}s; only the watchdog should kill)",
            WEDGE_SFTP_ADDRESS, WEDGE_IDLE_TIMEOUT_SECS
        );
        let binary_path = rustfs_binary_path_with_features(Some("ftps,webdav,sftp"));
        let host_key_dir_str = host_key_dir
            .to_str()
            .ok_or_else(|| anyhow!("host key dir path is not utf-8: {}", host_key_dir.display()))?;
        let mut server_process = ServerProcess::new(
            Command::new(&binary_path)
                .env(ENV_SFTP_ENABLE, "true")
                .env(ENV_SFTP_ADDRESS, WEDGE_SFTP_ADDRESS)
                .env(ENV_SFTP_HOST_KEY_DIR, host_key_dir_str)
                .env(ENV_SFTP_READ_ONLY, "false")
                .env(ENV_SFTP_PART_SIZE, PART_SIZE_ENV)
                .env(ENV_SFTP_IDLE_TIMEOUT, WEDGE_IDLE_TIMEOUT_SECS.to_string())
                .env(ENV_RUSTFS_ADDRESS, WEDGE_S3_ADDRESS)
                .env(ENV_CONSOLE_ENABLE, "false")
                .env("RUSTFS_OBS_LOGGER_LEVEL", "rustfs_protocols=debug")
                .env("RUST_LOG", "rustfs_protocols=debug")
                .stdout(Stdio::piped())
                .arg(&env.temp_dir)
                .spawn()?,
        );
        let counters = SessionCounters::new();
        watch_session_lifecycle_events(server_process.child_mut(), Arc::clone(&counters));

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(WEDGE_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(WEDGE_S3_ENDPOINT);
            wait_for_s3_ready(&s3, WEDGE_S3_READY_ATTEMPTS).await?;

            let bucket = "wedge";
            let seeded_key = "fixture.bin";
            s3.create_bucket()
                .bucket(bucket)
                .send()
                .await
                .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;
            let payload: Vec<u8> = (0..WEDGE_FIXTURE_BYTES).map(|i| (i as u8).wrapping_mul(11)).collect();
            s3.put_object()
                .bucket(bucket)
                .key(seeded_key)
                .body(ByteStream::from(payload))
                .send()
                .await
                .map_err(|e| anyhow!("S3 PutObject {bucket}/{seeded_key} failed: {e:?}"))?;

            let mut futs = Vec::with_capacity(WEDGE_PARALLEL_SESSIONS);
            for i in 0..WEDGE_PARALLEL_SESSIONS {
                let address = WEDGE_SFTP_ADDRESS.to_string();
                let bucket = bucket.to_string();
                let key = seeded_key.to_string();
                futs.push(tokio::spawn(async move {
                    drive_wedge_session(&address, &bucket, &key)
                        .await
                        .map_err(|e| anyhow!("wedge session {i} setup failed: {e}"))
                }));
            }
            let mut keepalive: Vec<(client::Handle<AcceptAnyServerKey>, russh::Channel<russh::client::Msg>, Arc<WedgeControl>)> =
                Vec::new();
            for fut in futs {
                keepalive.push(fut.await??);
            }

            let entered_after_setup = counters.entered.load(Ordering::Relaxed);
            let finished_after_setup = counters.finished.load(Ordering::Relaxed);
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: {} wedge sessions established (server entered={}, finished={}); waiting {} s for the watchdog kill path",
                WEDGE_PARALLEL_SESSIONS, entered_after_setup, finished_after_setup, WEDGE_WAIT_SECS,
            );

            sleep(Duration::from_secs(WEDGE_WAIT_SECS)).await;

            // Tickle the accept loop so JoinSet::try_join_next emits
            // the "SFTP session task finished" log lines for any
            // session the watchdog has canceled. Same mechanism the
            // half-close case uses. Mirrors the accept-loop drain
            // pattern in server.rs.
            for _ in 0..3 {
                if let Ok(stream) = TcpStream::connect(WEDGE_SFTP_ADDRESS).await {
                    drop(stream);
                }
                sleep(Duration::from_millis(200)).await;
            }
            sleep(Duration::from_millis(500)).await;

            let entered = counters.entered.load(Ordering::Relaxed);
            let finished = counters.finished.load(Ordering::Relaxed);
            let outstanding = entered.saturating_sub(finished);
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: post-wait counters entered={} finished={} outstanding={}",
                entered, finished, outstanding
            );
            if outstanding > 1 {
                return Err(anyhow!(
                    "{COMPLIANCE_TEST_OUTPUT_ID} session-task balance contract failed: entered={entered} finished={finished} outstanding={outstanding}, expected at most 1"
                ));
            }

            match count_close_wait_on_port(WEDGE_SFTP_PORT).await? {
                Some(0) => info!("{COMPLIANCE_TEST_OUTPUT_ID}: zero CLOSE_WAIT entries against port {WEDGE_SFTP_PORT}"),
                Some(n) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} {n} CLOSE_WAIT entries against port {WEDGE_SFTP_PORT}, expected 0"
                    ));
                }
                None => info!("{COMPLIANCE_TEST_OUTPUT_ID}: ss(8) unavailable, skipping CLOSE_WAIT assertion"),
            }

            drop(keepalive);
            info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: wedged sessions killed by the watchdog");
            Ok::<(), anyhow::Error>(())
        }
        .await;

        server_process.kill_and_wait().await;

        result
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_wedge_kill_after_silence_in_close_wait()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-26: healthy idle session past the watchdog fast-kill threshold stays alive.
pub(crate) mod cmptst_26 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-26";

    const IDLE_SFTP_PORT: u16 = 9028;
    const IDLE_SFTP_ADDRESS: &str = "127.0.0.1:9028";
    const IDLE_S3_ADDRESS: &str = "127.0.0.1:9304";
    const IDLE_S3_ENDPOINT: &str = "http://127.0.0.1:9304";
    const IDLE_S3_READY_ATTEMPTS: u32 = 30;

    // Idle timeout for the spawned server. 300 s sits well above the
    // wait window so russh's own inactivity_timeout cannot kill during
    // the test. The contract: a healthy idle session past the
    // watchdog's fast-kill threshold MUST stay alive. On Linux the
    // procfs probe sees ESTABLISHED and the decision function returns
    // Decision::Quiet. On non-Linux targets the fallback watchdog only
    // cancels at the far larger fallback ceiling, so the session
    // survives the wait window.
    const IDLE_TIMEOUT_SECS: u64 = 300;

    // Wait window. Must exceed the worst-case watchdog detection
    // latency on Linux: WEDGE_FAST_KILL_SILENCE_SECS (30) plus one to
    // two WEDGE_WATCHDOG_TICK_SECS ticks (15 each), so 45 to 60 s. Sits
    // well below WEDGE_FALLBACK_KILL_SILENCE_SECS (1800) so the fallback
    // path does not fire either. 90 s instead of 60 s: the case asserts
    // the watchdog does NOT false-kill, so a longer wait strengthens the
    // assertion. On Linux, if the procfs ESTABLISHED discriminator is
    // broken, more wait windows give it more chances to fire.
    const IDLE_WAIT_SECS: u64 = 90;

    pub(crate) async fn run_healthy_idle_session_above_fast_threshold() -> Result<()> {
        let env = ProtocolTestEnvironment::new().map_err(|e| anyhow!("{}", e))?;
        let host_key_dir = PathBuf::from(&env.temp_dir).join("sftp_host_keys");
        generate_host_key(&host_key_dir).await?;

        info!(
            "{COMPLIANCE_TEST_OUTPUT_ID}: starting idle-session server on {} (idle_timeout={}s)",
            IDLE_SFTP_ADDRESS, IDLE_TIMEOUT_SECS
        );
        let binary_path = rustfs_binary_path_with_features(Some("ftps,webdav,sftp"));
        let host_key_dir_str = host_key_dir
            .to_str()
            .ok_or_else(|| anyhow!("host key dir path is not utf-8: {}", host_key_dir.display()))?;
        let mut server_process = ServerProcess::new(
            Command::new(&binary_path)
                .env(ENV_SFTP_ENABLE, "true")
                .env(ENV_SFTP_ADDRESS, IDLE_SFTP_ADDRESS)
                .env(ENV_SFTP_HOST_KEY_DIR, host_key_dir_str)
                .env(ENV_SFTP_READ_ONLY, "false")
                .env(ENV_SFTP_PART_SIZE, PART_SIZE_ENV)
                .env(ENV_SFTP_IDLE_TIMEOUT, IDLE_TIMEOUT_SECS.to_string())
                .env(ENV_RUSTFS_ADDRESS, IDLE_S3_ADDRESS)
                .env(ENV_CONSOLE_ENABLE, "false")
                .env("RUSTFS_OBS_LOGGER_LEVEL", "rustfs_protocols=debug")
                .env("RUST_LOG", "rustfs_protocols=debug")
                .stdout(Stdio::piped())
                .arg(&env.temp_dir)
                .spawn()?,
        );
        let counters = SessionCounters::new();
        watch_session_lifecycle_events(server_process.child_mut(), Arc::clone(&counters));

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(IDLE_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(IDLE_S3_ENDPOINT);
            wait_for_s3_ready(&s3, IDLE_S3_READY_ATTEMPTS).await?;

            // Open one healthy SFTP session and drive a single
            // operation to stamp SessionDiag.last_activity_ms. The
            // watchdog measures silence from this moment.
            let (handle, sftp) = connect_sftp_to(IDLE_SFTP_ADDRESS).await?;
            let _ = sftp.canonicalize("/").await?;

            let entered_after_setup = counters.entered.load(Ordering::Relaxed);
            let finished_after_setup = counters.finished.load(Ordering::Relaxed);
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: idle session established (server entered={}, finished={}). Waiting {} s past the watchdog fast-kill threshold",
                entered_after_setup, finished_after_setup, IDLE_WAIT_SECS,
            );

            sleep(Duration::from_secs(IDLE_WAIT_SECS)).await;

            // Verify the session is still alive by driving another
            // operation. If the watchdog had killed the session during
            // the sleep, this canonicalize call would fail with a
            // closed-channel error.
            let final_realpath = sftp
                .canonicalize("/")
                .await
                .map_err(|e| anyhow!("post-wait canonicalize failed (likely watchdog false-kill): {e:?}"))?;
            if final_realpath != "/" {
                return Err(anyhow!(
                    "{COMPLIANCE_TEST_OUTPUT_ID} SFTP canonicalize returned unexpected result: {final_realpath:?}"
                ));
            }

            let entered_after_wait = counters.entered.load(Ordering::Relaxed);
            let finished_after_wait = counters.finished.load(Ordering::Relaxed);
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: post-wait counters entered={} finished={}",
                entered_after_wait, finished_after_wait,
            );

            // The contract: no session task ended during the wait
            // window. entered_after_wait may have grown if any ambient
            // probe traffic hit the listener. finished_after_wait must
            // equal finished_after_setup because no session ended.
            if finished_after_wait != finished_after_setup {
                return Err(anyhow!(
                    "{COMPLIANCE_TEST_OUTPUT_ID} watchdog false-killed a healthy idle session: finished went from {} to {} during the {} s wait",
                    finished_after_setup,
                    finished_after_wait,
                    IDLE_WAIT_SECS,
                ));
            }

            // Clean disconnect. The shutdown bumps finished by 1 after
            // this point but that is the expected end-of-test path,
            // not a watchdog kill.
            drop(sftp);
            let _ = handle.disconnect(russh::Disconnect::ByApplication, "test complete", "").await;

            info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: healthy idle session NOT killed by watchdog after {IDLE_WAIT_SECS} s");
            Ok::<(), anyhow::Error>(())
        }
        .await;

        server_process.kill_and_wait().await;

        result
    }

    // Excluded on Windows. Ambient operating-system traffic can inflate
    // the session counter and false-positive this healthy-idle assertion.
    // Linux and macOS run it.
    #[cfg(not(target_os = "windows"))]
    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_healthy_idle_session_above_fast_threshold()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-27: sustained-read thrash, multi-GiB downloads on N parallel sessions byte-identical to seed.
pub(crate) mod cmptst_27 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-27";

    const PIPE27_SFTP_PORT: u16 = 9035;
    const PIPE27_SFTP_ADDRESS: &str = "127.0.0.1:9035";
    const PIPE27_S3_ADDRESS: &str = "127.0.0.1:9311";
    const PIPE27_S3_ENDPOINT: &str = "http://127.0.0.1:9311";

    // Sustained-read thrash parameters. N parallel SFTP sessions each
    // download a multi-GiB object end-to-end and verify byte-exact
    // SHA256. The fixture is large enough to keep the SSH per-channel
    // window under sustained pressure. The per-session streaming
    // SHA256 keeps client-side memory bounded so the workload is not
    // memory-limited.
    //
    // Load-bearing assertions: byte-count and SHA256 match against the
    // seeded pattern. Both are independent of throughput, and both
    // fire under any silent corruption or short read.
    //
    // THRASH_DEADLINE_SECS is a no-progress safety floor only.
    // Aggregate throughput across N parallel sessions is bounded by
    // the SSH SFTP subsystem layer's per-channel serial handler
    // dispatch and the shared backend. The figure that comes back
    // varies by hardware. The deadline is set far above any realistic
    // completion time so it only trips when sessions stop progressing
    // entirely (a wedge), not when sessions are merely slow.
    const THRASH_PARALLEL: usize = 4;
    const THRASH_FIXTURE_DEFAULT_GIB: u64 = 5;
    const THRASH_DEADLINE_SECS: u64 = 3600;

    /// Returns the fixture size in bytes. Default 5 GiB. Override via
    /// RUSTFS_TEST_THRASH_FIXTURE_GIB so a memory-constrained CI runner
    /// can run the thrash case at 1 or 2 GiB without OOM-killing the
    /// linker or exhausting a tmpfs /tmp. The minimum that still keeps
    /// the SSH per-channel window under sustained pressure is around
    /// 512 MiB, but the env var accepts any positive integer GiB.
    fn thrash_fixture_bytes() -> u64 {
        let gib: u64 = std::env::var("RUSTFS_TEST_THRASH_FIXTURE_GIB")
            .ok()
            .and_then(|s| s.parse::<u64>().ok())
            .filter(|g| *g > 0)
            .unwrap_or(THRASH_FIXTURE_DEFAULT_GIB);
        gib * 1024 * 1024 * 1024
    }

    pub(crate) async fn run_multi_session_mixed_pipelining() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting sustained-read thrash server on {PIPE27_SFTP_ADDRESS}");
        let (_env, mut server_process) = spawn_pipelining_rustfs(PIPE27_SFTP_ADDRESS, PIPE27_S3_ADDRESS).await?;
        let server_log = capture_server_stdout(server_process.child_mut());

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(PIPE27_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(PIPE27_S3_ENDPOINT);
            wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;

            let bucket = "thrash";
            let key = "fixture.bin";
            s3.create_bucket()
                .bucket(bucket)
                .send()
                .await
                .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;

            let fixture_bytes = thrash_fixture_bytes();
            let gib = fixture_bytes / (1024 * 1024 * 1024);
            info!("{COMPLIANCE_TEST_OUTPUT_ID}: seeding {gib} GiB via multipart upload");
            let seed_t0 = Instant::now();
            seed_large_via_multipart(&s3, bucket, key, fixture_bytes).await?;
            info!("{COMPLIANCE_TEST_OUTPUT_ID}: seed complete in {:?}", seed_t0.elapsed());
            let expected_sha = calculate_pattern_sha256(fixture_bytes, THRASH_PATTERN_MULTIPLIER);

            let path = format!("/{bucket}/{key}");
            let mut handles = Vec::with_capacity(THRASH_PARALLEL);
            for session_idx in 0..THRASH_PARALLEL {
                let address = PIPE27_SFTP_ADDRESS.to_string();
                let path = path.clone();
                handles.push(tokio::spawn(async move {
                    let t0 = Instant::now();
                    let (_handle, sftp) = connect_sftp_to(&address).await?;
                    let (bytes, sha) = streaming_sha256_download(&sftp, &path).await?;
                    Ok::<(usize, u64, [u8; 32], Duration), anyhow::Error>((session_idx, bytes, sha, t0.elapsed()))
                }));
            }

            let overall = Duration::from_secs(THRASH_DEADLINE_SECS);
            let drained = timeout(overall, async {
                let mut results = Vec::with_capacity(THRASH_PARALLEL);
                for h in handles {
                    results.push(h.await.map_err(|e| anyhow!("worker join failed: {e}"))??);
                }
                Ok::<Vec<(usize, u64, [u8; 32], Duration)>, anyhow::Error>(results)
            })
            .await;
            let results = match drained {
                Ok(Ok(r)) => r,
                Ok(Err(e)) => return Err(e),
                Err(_) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} deadline exceeded: {THRASH_PARALLEL} sessions did not finish within {THRASH_DEADLINE_SECS} s"
                    ));
                }
            };

            for (idx, bytes, sha, elapsed) in &results {
                if *bytes != fixture_bytes {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} session {idx} truncated: read {bytes} bytes, expected {fixture_bytes} (elapsed {elapsed:?})",
                    ));
                }
                if sha != &expected_sha {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} session {idx} SHA256 mismatch (elapsed {elapsed:?})"
                    ));
                }
            }
            let slowest = results.iter().map(|r| r.3).max().unwrap_or_default();
            info!(
                "PASS {COMPLIANCE_TEST_OUTPUT_ID}: {THRASH_PARALLEL} parallel {gib} GiB downloads byte-identical (slowest {slowest:?})",
            );
            Ok::<(), anyhow::Error>(())
        }
        .await;

        if result.is_err() {
            let buf = server_log.lock().await;
            let lines: Vec<&String> = buf.iter().rev().take(200).collect();
            eprintln!("--- last {} lines of rustfs server stdout (oldest first) ---", lines.len());
            for line in lines.iter().rev() {
                eprintln!("{line}");
            }
            eprintln!("--- end rustfs stdout dump ---");
        }

        server_process.kill_and_wait().await;
        result
    }

    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_multi_session_mixed_pipelining()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-28: 5 MB download intact under concurrent metadata storm on a parallel session.
pub(crate) mod cmptst_28 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-28";

    const PIPE28_SFTP_PORT: u16 = 9029;
    const PIPE28_SFTP_ADDRESS: &str = "127.0.0.1:9029";
    const PIPE28_S3_ADDRESS: &str = "127.0.0.1:9305";
    const PIPE28_S3_ENDPOINT: &str = "http://127.0.0.1:9305";

    // Parameters. METADATA_STORM_OPS bounds the in-flight metadata
    // depth fired against the storm session. STORM_PARALLEL_SESSIONS
    // opens that many independent SFTP channels each running its own
    // storm. The download session runs alongside and must complete
    // within the per-session deadline.
    const METADATA_STORM_OPS: usize = 500;
    const STORM_PARALLEL_SESSIONS: usize = 4;
    const METADATA_STORM_DEADLINE_SECS: u64 = 20;

    pub(crate) async fn run_5mb_download_with_concurrent_metadata_ops() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting metadata-pressure server on {PIPE28_SFTP_ADDRESS}");
        let (_env, mut server_process) = spawn_pipelining_rustfs(PIPE28_SFTP_ADDRESS, PIPE28_S3_ADDRESS).await?;

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(PIPE28_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(PIPE28_S3_ENDPOINT);
            wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;

            let bucket = "pipe28";
            let fixture_key = "fixture.bin";
            let subdir = "siblings";
            let payload = seed_pipelining_fixture(&s3, bucket, fixture_key, subdir).await?;
            let expected_sha: [u8; 32] = Sha256::digest(&payload).into();

            let fixture_path = format!("/{bucket}/{fixture_key}");
            let subdir_path = format!("/{bucket}/{subdir}");

            let stop_flag = Arc::new(AtomicBool::new(false));
            let mut storm_handles = Vec::with_capacity(STORM_PARALLEL_SESSIONS);
            for storm_idx in 0..STORM_PARALLEL_SESSIONS {
                let storm_address = PIPE28_SFTP_ADDRESS.to_string();
                let storm_subdir = subdir_path.clone();
                let storm_flag = Arc::clone(&stop_flag);
                storm_handles.push(tokio::spawn(async move {
                    let (_handle, sftp) = connect_sftp_to(&storm_address).await?;
                    let sftp = Arc::new(sftp);
                    let mut pipeline: FuturesUnordered<_> = (0..METADATA_STORM_OPS)
                        .map(|i| {
                            let sftp = Arc::clone(&sftp);
                            let subdir = storm_subdir.clone();
                            let flag = Arc::clone(&storm_flag);
                            async move {
                                if flag.load(Ordering::Relaxed) {
                                    return Ok::<(), anyhow::Error>(());
                                }
                                if i % 2 == 0 {
                                    sftp.read_dir(&subdir)
                                        .await
                                        .map_err(|e| anyhow!("storm {storm_idx} READDIR failed: {e:?}"))?;
                                } else {
                                    let path = format!("{subdir}/file_{:04}.txt", i % SUBDIR_FILE_COUNT);
                                    sftp.metadata(&path)
                                        .await
                                        .map_err(|e| anyhow!("storm {storm_idx} STAT {path} failed: {e:?}"))?;
                                }
                                Ok::<(), anyhow::Error>(())
                            }
                        })
                        .collect();
                    while let Some(r) = pipeline.next().await {
                        r?;
                    }
                    Ok::<(), anyhow::Error>(())
                }));
            }

            let download_address = PIPE28_SFTP_ADDRESS.to_string();
            let download_path = fixture_path.clone();
            let download = tokio::spawn(async move {
                let (_handle, sftp) = connect_sftp_to(&download_address).await?;
                let bytes = sftp_read_full(&sftp, &download_path)
                    .await
                    .map_err(|e| anyhow!("download READ failed: {e:?}"))?;
                if bytes.len() != FIXTURE_SIZE {
                    return Err(anyhow!(
                        "download byte count mismatch: expected {FIXTURE_SIZE}, got {}",
                        bytes.len()
                    ));
                }
                let observed: [u8; 32] = Sha256::digest(&bytes).into();
                if observed != expected_sha {
                    return Err(anyhow!("download SHA256 mismatch on {download_path}"));
                }
                Ok(())
            });

            let overall = Duration::from_secs(METADATA_STORM_DEADLINE_SECS);
            let download_outcome = timeout(overall, download).await;
            stop_flag.store(true, Ordering::Relaxed);
            for handle in storm_handles {
                let _ = handle.await;
            }

            match download_outcome {
                Ok(Ok(Ok(()))) => {}
                Ok(Ok(Err(e))) => return Err(e),
                Ok(Err(e)) => return Err(anyhow!("download join failed: {e}")),
                Err(_elapsed) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} deadline exceeded: download did not finish within {METADATA_STORM_DEADLINE_SECS} s under metadata pressure (parallel storm sessions = {STORM_PARALLEL_SESSIONS}, in-flight depth per storm = {METADATA_STORM_OPS})"
                    ));
                }
            }
            info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: 5 MB download finished intact under concurrent metadata pressure");
            Ok::<(), anyhow::Error>(())
        }
        .await;

        server_process.kill_and_wait().await;
        result
    }

    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_5mb_download_with_concurrent_metadata_ops()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-29: high-volume read-past-EOF pipelining completes inside the deadline and every read returns EOF.
pub(crate) mod cmptst_29 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-29";

    const PIPE29_SFTP_PORT: u16 = 9030;
    const PIPE29_SFTP_ADDRESS: &str = "127.0.0.1:9030";
    const PIPE29_S3_ADDRESS: &str = "127.0.0.1:9306";
    const PIPE29_S3_ENDPOINT: &str = "http://127.0.0.1:9306";

    // Parameters. EOF_VOLUME_REQUEST_COUNT total reads, fanned out
    // across EOF_VOLUME_INFLIGHT_DEPTH file handles on a single
    // SftpSession. Each handle drives reads serially within itself,
    // but reads across handles run concurrently because the russh-sftp
    // client pipelines per-call response routing through a request-id
    // table. EOF_VOLUME_INFLIGHT_DEPTH is held below the server's
    // default handles-per-session cap (DEFAULT_HANDLES_PER_SESSION = 64
    // in crates/protocols/src/sftp/constants.rs) so the test never
    // trips the cap-exceeded surface, which has its own dedicated
    // coverage.
    const EOF_VOLUME_FIXTURE_BYTES: usize = 1024;
    const EOF_VOLUME_REQUEST_COUNT: usize = 10_000;
    const EOF_VOLUME_INFLIGHT_DEPTH: usize = 50;
    const EOF_VOLUME_DEADLINE_SECS: u64 = 30;

    pub(crate) async fn run_read_past_eof_volume() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting EOF-volume server on {PIPE29_SFTP_ADDRESS}");
        let (_env, mut server_process) = spawn_pipelining_rustfs(PIPE29_SFTP_ADDRESS, PIPE29_S3_ADDRESS).await?;

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(PIPE29_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(PIPE29_S3_ENDPOINT);
            wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;

            let bucket = "pipe29";
            let key = "tiny.bin";
            s3.create_bucket()
                .bucket(bucket)
                .send()
                .await
                .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;
            let payload: Vec<u8> = (0..EOF_VOLUME_FIXTURE_BYTES).map(|i| i as u8).collect();
            s3.put_object()
                .bucket(bucket)
                .key(key)
                .body(ByteStream::from(payload.clone()))
                .send()
                .await
                .map_err(|e| anyhow!("S3 PutObject {bucket}/{key} failed: {e:?}"))?;

            let path = format!("/{bucket}/{key}");
            let (_handle, sftp) = connect_sftp_to(PIPE29_SFTP_ADDRESS).await?;
            let sftp = Arc::new(sftp);
            // Open a fan of independent file handles. russh-sftp's File
            // requires &mut self for read, so concurrent reads need
            // separate handles. Reads stack as in-flight FXP packets
            // on the same channel because the underlying SftpSession
            // pipelines through its request-id table.
            let mut handle_setup: FuturesUnordered<_> = (0..EOF_VOLUME_INFLIGHT_DEPTH)
                .map(|_| {
                    let sftp = Arc::clone(&sftp);
                    let path = path.clone();
                    async move {
                        let mut file = sftp
                            .open_with_flags(&path, OpenFlags::READ)
                            .await
                            .map_err(|e| anyhow!("OPEN {path} failed: {e:?}"))?;
                        file.seek(std::io::SeekFrom::Start((EOF_VOLUME_FIXTURE_BYTES as u64) + 1024))
                            .await
                            .map_err(|e| anyhow!("SEEK past EOF failed: {e:?}"))?;
                        Ok::<_, anyhow::Error>(file)
                    }
                })
                .collect();
            let mut files = Vec::with_capacity(EOF_VOLUME_INFLIGHT_DEPTH);
            while let Some(r) = handle_setup.next().await {
                files.push(r?);
            }

            let overall = Duration::from_secs(EOF_VOLUME_DEADLINE_SECS);
            let reads_per_handle = EOF_VOLUME_REQUEST_COUNT / EOF_VOLUME_INFLIGHT_DEPTH;
            let drained = timeout(overall, async {
                let mut pipeline: FuturesUnordered<_> = files
                    .into_iter()
                    .map(|mut file| async move {
                        let mut scratch = [0u8; 64];
                        for i in 0..reads_per_handle {
                            match file.read(&mut scratch).await {
                                Ok(0) => {}
                                Ok(n) => {
                                    return Err(anyhow!(
                                        "{COMPLIANCE_TEST_OUTPUT_ID} read {i} returned {n} bytes past EOF; expected 0"
                                    ));
                                }
                                Err(e) => {
                                    return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} read {i} returned an error: {e}"));
                                }
                            }
                        }
                        let _ = file.shutdown().await;
                        Ok::<(), anyhow::Error>(())
                    })
                    .collect();
                while let Some(r) = pipeline.next().await {
                    r?;
                }
                Ok(())
            })
            .await;

            match drained {
                Ok(Ok(())) => {}
                Ok(Err(e)) => return Err(e),
                Err(_elapsed) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} deadline exceeded: {EOF_VOLUME_REQUEST_COUNT} EOF reads spread across {EOF_VOLUME_INFLIGHT_DEPTH} handles did not finish within {EOF_VOLUME_DEADLINE_SECS} s"
                    ));
                }
            }
            info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: {EOF_VOLUME_REQUEST_COUNT} read-past-EOF requests completed inside the deadline");
            Ok::<(), anyhow::Error>(())
        }
        .await;

        server_process.kill_and_wait().await;
        result
    }

    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_read_past_eof_volume()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-30: per-operation handler latency stays inside the ceiling under parallel pipelined sessions.
pub(crate) mod cmptst_30 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-30";

    const PIPE30_SFTP_PORT: u16 = 9031;
    const PIPE30_SFTP_ADDRESS: &str = "127.0.0.1:9031";
    const PIPE30_S3_ADDRESS: &str = "127.0.0.1:9307";
    const PIPE30_S3_ENDPOINT: &str = "http://127.0.0.1:9307";

    // Parameters. The per-operation ceiling is 1 s.
    // LATENCY_INFLIGHT_DEPTH per-session pipelines metadata operations
    // the same way the GUI-client traversal shape does, so a single
    // slow handler shows up against the ceiling instead of being
    // averaged into a passing aggregate.
    const LATENCY_PARALLEL: usize = 8;
    const LATENCY_ITERATIONS: usize = 20;
    const LATENCY_INFLIGHT_DEPTH: usize = 50;
    const LATENCY_PER_OP_CEILING_MILLIS: u64 = 1_000;
    const LATENCY_OVERALL_DEADLINE_SECS: u64 = 120;

    /// One observation: the wall-clock latency for one operation paired
    /// with a static label naming the op type so the failure log
    /// identifies which client-visible category of work produced the
    /// worst sample.
    type LatencyObservation = (Duration, &'static str);

    /// One worker outcome: the worst metadata-op observation (STAT or
    /// READDIR) and the worst fixture-read observation tracked
    /// independently. The ceiling assertion is metadata-only because a
    /// multi-MB SFTP READ inherently round-trips per MAX_READ_LEN chunk
    /// and a 5 MB transfer at ~100 ms per round trip lands well above
    /// the metadata ceiling without representing a wedge regression.
    struct WorkerWorst {
        metadata: LatencyObservation,
        read: Duration,
    }

    /// Worker: open one SFTP session and drive several batches of
    /// deeply-pipelined metadata operations interleaved with full
    /// fixture reads. Each batch fires LATENCY_INFLIGHT_DEPTH
    /// concurrent metadata futures on a single channel, mirroring
    /// GUI-client pipelining. Returns the worst metadata-op
    /// observation alongside the worst fixture-read wall-clock so the
    /// caller can assert against each surface independently.
    async fn cmptst30_worker(address: &str, fixture_path: String, subdir_path: String) -> Result<WorkerWorst> {
        let (_handle, sftp) = connect_sftp_to(address).await?;
        let sftp = Arc::new(sftp);
        let mut worst_meta: LatencyObservation = (Duration::ZERO, "init");
        let mut worst_read: Duration = Duration::ZERO;
        for _ in 0..LATENCY_ITERATIONS {
            let mut pipeline: FuturesUnordered<_> = (0..LATENCY_INFLIGHT_DEPTH)
                .map(|i| {
                    let sftp = Arc::clone(&sftp);
                    let fixture_path = fixture_path.clone();
                    let subdir_path = subdir_path.clone();
                    async move {
                        let t = Instant::now();
                        let op: &'static str = if i % 3 == 0 {
                            sftp.metadata(&fixture_path)
                                .await
                                .map_err(|e| anyhow!("STAT {fixture_path} failed: {e:?}"))?;
                            "metadata-fixture"
                        } else if i % 3 == 1 {
                            sftp.read_dir(&subdir_path)
                                .await
                                .map_err(|e| anyhow!("READDIR {subdir_path} failed: {e:?}"))?;
                            "readdir-subdir"
                        } else {
                            let path = format!("{subdir_path}/file_{:04}.txt", i % SUBDIR_FILE_COUNT);
                            sftp.metadata(&path).await.map_err(|e| anyhow!("STAT {path} failed: {e:?}"))?;
                            "metadata-sibling"
                        };
                        Ok::<LatencyObservation, anyhow::Error>((t.elapsed(), op))
                    }
                })
                .collect();
            while let Some(r) = pipeline.next().await {
                let observation = r?;
                if observation.0 > worst_meta.0 {
                    worst_meta = observation;
                }
            }

            let t = Instant::now();
            let bytes = sftp_read_full(&sftp, &fixture_path)
                .await
                .map_err(|e| anyhow!("READ {fixture_path} failed: {e:?}"))?;
            let elapsed = t.elapsed();
            if elapsed > worst_read {
                worst_read = elapsed;
            }
            if bytes.len() != FIXTURE_SIZE {
                return Err(anyhow!(
                    "READ {fixture_path} byte count mismatch: expected {FIXTURE_SIZE}, got {}",
                    bytes.len()
                ));
            }
        }
        Ok(WorkerWorst {
            metadata: worst_meta,
            read: worst_read,
        })
    }

    pub(crate) async fn run_handler_latency_under_backend_pressure() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting handler-latency server on {PIPE30_SFTP_ADDRESS}");
        let (_env, mut server_process) = spawn_pipelining_rustfs(PIPE30_SFTP_ADDRESS, PIPE30_S3_ADDRESS).await?;

        let result = async {
            ProtocolTestEnvironment::wait_for_port_ready(PIPE30_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(PIPE30_S3_ENDPOINT);
            wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;

            let bucket = "pipe30";
            let fixture_key = "fixture.bin";
            let subdir = "siblings";
            let _ = seed_pipelining_fixture(&s3, bucket, fixture_key, subdir).await?;

            let fixture_path = format!("/{bucket}/{fixture_key}");
            let subdir_path = format!("/{bucket}/{subdir}");

            let mut handles = Vec::with_capacity(LATENCY_PARALLEL);
            for session_idx in 0..LATENCY_PARALLEL {
                let address = PIPE30_SFTP_ADDRESS.to_string();
                let fixture_path = fixture_path.clone();
                let subdir_path = subdir_path.clone();
                handles.push(tokio::spawn(async move {
                    cmptst30_worker(&address, fixture_path, subdir_path)
                        .await
                        .map_err(|e| anyhow!("session {session_idx}: {e}"))
                }));
            }

            let overall = Duration::from_secs(LATENCY_OVERALL_DEADLINE_SECS);
            let drained = timeout(overall, async {
                let mut worst_meta: LatencyObservation = (Duration::ZERO, "init");
                let mut worst_read = Duration::ZERO;
                for handle in handles {
                    let session = handle.await.map_err(|e| anyhow!("worker join failed: {e}"))??;
                    if session.metadata.0 > worst_meta.0 {
                        worst_meta = session.metadata;
                    }
                    if session.read > worst_read {
                        worst_read = session.read;
                    }
                }
                Ok::<(LatencyObservation, Duration), anyhow::Error>((worst_meta, worst_read))
            })
            .await;

            let (worst_meta, worst_read) = match drained {
                Ok(Ok(p)) => p,
                Ok(Err(e)) => return Err(e),
                Err(_elapsed) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} deadline exceeded: workers did not finish within {LATENCY_OVERALL_DEADLINE_SECS} s"
                    ));
                }
            };
            let ceiling = Duration::from_millis(LATENCY_PER_OP_CEILING_MILLIS);
            if worst_meta.0 > ceiling {
                return Err(anyhow!(
                    "{COMPLIANCE_TEST_OUTPUT_ID} metadata ceiling exceeded: worst metadata op {} ms on '{}' > {} ms (worst fixture read {} ms; depth={LATENCY_INFLIGHT_DEPTH})",
                    worst_meta.0.as_millis(),
                    worst_meta.1,
                    ceiling.as_millis(),
                    worst_read.as_millis(),
                ));
            }
            info!(
                "PASS {COMPLIANCE_TEST_OUTPUT_ID}: worst metadata op {} ms on '{}' (ceiling {} ms; worst fixture read {} ms; depth={LATENCY_INFLIGHT_DEPTH})",
                worst_meta.0.as_millis(),
                worst_meta.1,
                ceiling.as_millis(),
                worst_read.as_millis(),
            );
            Ok::<(), anyhow::Error>(())
        }
        .await;

        server_process.kill_and_wait().await;
        result
    }

    #[ignore]
    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_handler_latency_under_backend_pressure()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-31: server resilience under client paused-drain, byte-exact completion after a mid-transfer pause.
pub(crate) mod cmptst_31 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-31";

    const PIPE31_SFTP_PORT: u16 = 9032;
    const PIPE31_SFTP_ADDRESS: &str = "127.0.0.1:9032";
    const PIPE31_S3_ADDRESS: &str = "127.0.0.1:9308";
    const PIPE31_S3_ENDPOINT: &str = "http://127.0.0.1:9308";

    // Parameters. Single SFTP session, multi-MB seed, and a
    // deterministic mid-transfer pause on the client-side TCP read
    // half. The pause lets the rustfs server fill its kernel TCP send
    // buffer and exhaust the SSH per-channel recipient_window_size,
    // which is the load-bearing precondition for russh-sftp's
    // stream.flush().await to park inside the per-channel response
    // loop. Pause duration is long enough that the watchdog and any
    // russh keepalives can't reach the parked task before the test
    // observes the symptom.
    const PAUSE31_FIXTURE_BYTES: u64 = 200 * 1024 * 1024;
    const PAUSE31_PRE_PAUSE_BYTES: u64 = 4 * 1024 * 1024;
    const PAUSE31_PAUSE_SECS: u64 = 25;
    const PAUSE31_RESUME_DEADLINE_SECS: u64 = 120;
    const PAUSE31_OVERALL_DEADLINE_SECS: u64 = 240;

    /// Control flag flipped by the test loop to pause the underlying
    /// TCP read half on the client side. Used to deplete the SSH
    /// recipient_window_size on the server side and force
    /// stream.flush() to park.
    struct PauseControl {
        paused: AtomicBool,
    }

    impl PauseControl {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                paused: AtomicBool::new(false),
            })
        }
    }

    /// Wrapper around tokio::net::TcpStream split halves. poll_read
    /// returns Pending while the control flag is set, simulating a
    /// slow-drain client (the FileZilla / Cyberduck shape). poll_write
    /// is unmodified so the russh client can keep sending FXP requests
    /// while the response side is throttled.
    struct PausableStream {
        read: OwnedReadHalf,
        write: OwnedWriteHalf,
        control: Arc<PauseControl>,
    }

    impl PausableStream {
        fn new(stream: TcpStream, control: Arc<PauseControl>) -> Self {
            let (read, write) = stream.into_split();
            Self { read, write, control }
        }
    }

    impl AsyncRead for PausableStream {
        fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            if this.control.paused.load(Ordering::Relaxed) {
                // Return Pending. Spawn a 100 ms-delayed task to wake
                // the future so the runtime re-polls and observes the
                // pause flag once it clears. The 100 ms interval caps
                // wake-up latency after the test releases the pause.
                let waker = cx.waker().clone();
                tokio::spawn(async move {
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    waker.wake();
                });
                return Poll::Pending;
            }
            Pin::new(&mut this.read).poll_read(cx, buf)
        }
    }

    impl AsyncWrite for PausableStream {
        fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
            Pin::new(&mut self.get_mut().write).poll_write(cx, buf)
        }

        fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Pin::new(&mut self.get_mut().write).poll_flush(cx)
        }

        fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Pin::new(&mut self.get_mut().write).poll_shutdown(cx)
        }
    }

    /// Connect to the server through a PausableStream so the test loop
    /// can pause the client-side TCP read half mid-transfer. Returns
    /// the russh client handle, the SFTP session, and the control
    /// handle the caller uses to flip the pause state.
    async fn connect_pausable_sftp(
        address: &str,
    ) -> Result<(client::Handle<AcceptAnyServerKey>, SftpSession, Arc<PauseControl>)> {
        let tcp = TcpStream::connect(address)
            .await
            .map_err(|e| anyhow!("TcpStream::connect {address} failed: {e}"))?;
        let control = PauseControl::new();
        let stream = PausableStream::new(tcp, Arc::clone(&control));
        let config = Arc::new(client::Config::default());
        let mut session = client::connect_stream(config, stream, AcceptAnyServerKey)
            .await
            .map_err(|e| anyhow!("russh connect_stream failed: {e}"))?;
        let auth = session
            .authenticate_password(DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY)
            .await
            .map_err(|e| anyhow!("authenticate_password failed: {e}"))?;
        if !auth.success() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} password auth rejected"));
        }
        let channel = session
            .channel_open_session()
            .await
            .map_err(|e| anyhow!("channel_open_session failed: {e}"))?;
        channel
            .request_subsystem(true, "sftp")
            .await
            .map_err(|e| anyhow!("request_subsystem failed: {e}"))?;
        // 60 s per-request timeout (default 10 s) so the 25 s test
        // pause does not trip the russh-sftp client's own request
        // timer before the server's flush parking can be observed.
        // The wedge mechanism the case is designed to surface is
        // server side. A client-side request timer firing first masks
        // it.
        let sftp = SftpSession::new_with_config(
            channel.into_stream(),
            Config {
                request_timeout_secs: 60,
                ..Config::default()
            },
        )
        .await
        .map_err(|e| anyhow!("SftpSession::new_with_config failed: {e}"))?;
        Ok((session, sftp, control))
    }

    // Test orchestration in three phases:
    //
    // 1. Pre-pause: drain PAUSE31_PRE_PAUSE_BYTES from the server
    //    normally to confirm the read path is healthy.
    // 2. Pause: flip the PausableStream pause flag and sleep for
    //    PAUSE31_PAUSE_SECS. With reads parked the kernel TCP receive
    //    buffer fills, the SSH per-channel recipient_window depletes,
    //    and the server-side stream.flush() parks inside the response
    //    loop. The pause is intentionally longer than the watchdog
    //    fast-kill threshold so this case proves the watchdog does
    //    NOT kill a session that is parked on flush (only sessions
    //    parked on the russh select! mpsc, see CMPTST-25).
    // 3. Resume: clear the pause flag, drain the remaining bytes
    //    inside PAUSE31_RESUME_DEADLINE_SECS, and SHA-compare against
    //    the seeded fixture to prove byte correctness.
    pub(crate) async fn run_paused_drain_provokes_flush_park() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting paused-drain wedge probe server on {PIPE31_SFTP_ADDRESS}");
        let (_env, mut server_process) = spawn_pipelining_rustfs(PIPE31_SFTP_ADDRESS, PIPE31_S3_ADDRESS).await?;
        let server_log = capture_server_stdout(server_process.child_mut());

        let result: Result<()> = async {
            ProtocolTestEnvironment::wait_for_port_ready(PIPE31_SFTP_PORT, 30)
                .await
                .map_err(|e| anyhow!("{}", e))?;

            let s3 = build_test_s3_client(PIPE31_S3_ENDPOINT);
            wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;

            let bucket = "pause31";
            let key = "fixture.bin";
            s3.create_bucket()
                .bucket(bucket)
                .send()
                .await
                .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;

            let mib = PAUSE31_FIXTURE_BYTES / (1024 * 1024);
            info!("{COMPLIANCE_TEST_OUTPUT_ID}: seeding {mib} MiB via multipart upload");
            let seed_t0 = Instant::now();
            seed_large_via_multipart(&s3, bucket, key, PAUSE31_FIXTURE_BYTES).await?;
            info!("{COMPLIANCE_TEST_OUTPUT_ID}: seed complete in {:?}", seed_t0.elapsed());
            let expected_sha = calculate_pattern_sha256(PAUSE31_FIXTURE_BYTES, THRASH_PATTERN_MULTIPLIER);

            let path = format!("/{bucket}/{key}");
            let (_handle, sftp, control) = connect_pausable_sftp(PIPE31_SFTP_ADDRESS).await?;
            let mut file = sftp
                .open_with_flags(&path, OpenFlags::READ)
                .await
                .map_err(|e| anyhow!("OPEN {path} failed: {e:?}"))?;

            let mut hasher = Sha256::new();
            let mut buf = vec![0u8; 256 * 1024];
            let mut total: u64 = 0;

            // Drain enough bytes that the SSH window has had time to
            // be reset and the connection is in steady state.
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: pre-pause drain up to {} bytes",
                PAUSE31_PRE_PAUSE_BYTES
            );
            while total < PAUSE31_PRE_PAUSE_BYTES {
                let n = file.read(&mut buf).await.map_err(|e| anyhow!("pre-pause READ failed: {e:?}"))?;
                if n == 0 {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} pre-pause drain short: got {total} bytes before EOF, expected at least {PAUSE31_PRE_PAUSE_BYTES}"
                    ));
                }
                hasher.update(&buf[..n]);
                total += n as u64;
            }
            info!("{COMPLIANCE_TEST_OUTPUT_ID}: pre-pause drain complete, drained {total} bytes; flipping pause flag");

            // Pause client-side reads. The server will keep pushing
            // FXP_DATA responses for in-flight FXP_READ requests until
            // its kernel TCP send buffer fills and the SSH
            // recipient_window_size is exhausted. From there
            // stream.flush() is expected to park inside russh-sftp's
            // per-channel response loop.
            control.paused.store(true, Ordering::Relaxed);
            let pause_t0 = Instant::now();

            // Spawn the read continuation. It will block at the first
            // file.read() call once the in-buffer SSH stream is
            // drained.
            let read_handle = tokio::spawn(async move {
                let mut hasher = hasher;
                let mut buf = buf;
                let mut total = total;
                while total < PAUSE31_FIXTURE_BYTES {
                    let n = file.read(&mut buf).await.map_err(|e| anyhow!("post-pause READ failed: {e:?}"))?;
                    if n == 0 {
                        break;
                    }
                    hasher.update(&buf[..n]);
                    total += n as u64;
                }
                let _ = file.shutdown().await;
                let sha: [u8; 32] = hasher.finalize().into();
                Ok::<(u64, [u8; 32]), anyhow::Error>((total, sha))
            });

            // Hold the pause for the configured window.
            tokio::time::sleep(Duration::from_secs(PAUSE31_PAUSE_SECS)).await;
            let pause_elapsed = pause_t0.elapsed();
            info!(
                "{COMPLIANCE_TEST_OUTPUT_ID}: pause window elapsed {pause_elapsed:?}; releasing pause flag"
            );
            control.paused.store(false, Ordering::Relaxed);

            // Read continuation must complete inside the resume
            // deadline. If it does not, the server-side flush did not
            // unwedge after the SSH window was replenished.
            let resume_outcome = timeout(Duration::from_secs(PAUSE31_RESUME_DEADLINE_SECS), read_handle).await;
            let (final_total, observed_sha) = match resume_outcome {
                Ok(join_result) => match join_result {
                    Ok(Ok(p)) => p,
                    Ok(Err(e)) => return Err(e),
                    Err(e) => return Err(anyhow!("read continuation join failed: {e}")),
                },
                Err(_) => {
                    return Err(anyhow!(
                        "{COMPLIANCE_TEST_OUTPUT_ID} resume deadline exceeded: read continuation did not finish within {PAUSE31_RESUME_DEADLINE_SECS} s after the pause flag was released"
                    ));
                }
            };
            if final_total != PAUSE31_FIXTURE_BYTES {
                return Err(anyhow!(
                    "{COMPLIANCE_TEST_OUTPUT_ID} final byte count mismatch: read {final_total} bytes, expected {PAUSE31_FIXTURE_BYTES}"
                ));
            }
            if observed_sha != expected_sha {
                return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} SHA256 mismatch on {path}"));
            }

            info!(
                "PASS {COMPLIANCE_TEST_OUTPUT_ID}: server delivered {final_total} bytes byte-exact across a {PAUSE31_PAUSE_SECS} s client paused-drain"
            );
            Ok(())
        }
        .await;

        if result.is_err() {
            let buf = server_log.lock().await;
            let lines: Vec<&String> = buf.iter().rev().take(200).collect();
            eprintln!("--- last {} lines of rustfs server stdout (oldest first) ---", lines.len());
            for line in lines.iter().rev() {
                eprintln!("{line}");
            }
            eprintln!("--- end rustfs stdout dump ---");
        }

        let _ = timeout(Duration::from_secs(PAUSE31_OVERALL_DEADLINE_SECS), async {
            server_process.kill_and_wait().await;
        })
        .await;
        result
    }

    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_paused_drain_provokes_flush_park()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-32: read-cache enabled regression, 8 MiB download byte-exact with the production cache window.
pub(crate) mod cmptst_32 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-32";

    const PIPE32_SFTP_PORT: u16 = 9033;
    const PIPE32_SFTP_ADDRESS: &str = "127.0.0.1:9033";
    const PIPE32_S3_ADDRESS: &str = "127.0.0.1:9309";
    const PIPE32_S3_ENDPOINT: &str = "http://127.0.0.1:9309";

    pub(crate) async fn run_read_cache_enabled_round_trip() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting read-cache enabled run on {PIPE32_SFTP_ADDRESS}");
        let extras = [(ENV_SFTP_READ_CACHE_WINDOW_BYTES, "1048576")];
        let (_env, mut server_process) =
            spawn_pipelining_rustfs_with_extras(PIPE32_SFTP_ADDRESS, PIPE32_S3_ADDRESS, &extras).await?;
        let server_log = capture_server_stdout(server_process.child_mut());

        let result = run_read_cache_byte_correctness(
            PIPE32_SFTP_PORT,
            PIPE32_SFTP_ADDRESS,
            PIPE32_S3_ENDPOINT,
            "pipe32",
            COMPLIANCE_TEST_OUTPUT_ID,
        )
        .await;

        if result.is_err() {
            let buf = server_log.lock().await;
            let lines: Vec<&String> = buf.iter().rev().take(200).collect();
            eprintln!("--- last {} lines of rustfs server stdout (oldest first) ---", lines.len());
            for line in lines.iter().rev() {
                eprintln!("{line}");
            }
            eprintln!("--- end rustfs stdout dump ---");
        }

        let _ = timeout(Duration::from_secs(READ_CACHE_DEADLINE_SECS), async {
            server_process.kill_and_wait().await;
        })
        .await;
        result
    }

    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_read_cache_enabled_round_trip()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-33: read-cache disabled regression, 8 MiB download byte-exact with RUSTFS_SFTP_READ_CACHE_WINDOW_BYTES=0.
pub(crate) mod cmptst_33 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-33";

    const PIPE33_SFTP_PORT: u16 = 9034;
    const PIPE33_SFTP_ADDRESS: &str = "127.0.0.1:9034";
    const PIPE33_S3_ADDRESS: &str = "127.0.0.1:9310";
    const PIPE33_S3_ENDPOINT: &str = "http://127.0.0.1:9310";

    pub(crate) async fn run_read_cache_disabled_round_trip() -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: starting read-cache disabled run on {PIPE33_SFTP_ADDRESS}");
        let extras = [(ENV_SFTP_READ_CACHE_WINDOW_BYTES, "0")];
        let (_env, mut server_process) =
            spawn_pipelining_rustfs_with_extras(PIPE33_SFTP_ADDRESS, PIPE33_S3_ADDRESS, &extras).await?;
        let server_log = capture_server_stdout(server_process.child_mut());

        let result = run_read_cache_byte_correctness(
            PIPE33_SFTP_PORT,
            PIPE33_SFTP_ADDRESS,
            PIPE33_S3_ENDPOINT,
            "pipe33",
            COMPLIANCE_TEST_OUTPUT_ID,
        )
        .await;

        if result.is_err() {
            let buf = server_log.lock().await;
            let lines: Vec<&String> = buf.iter().rev().take(200).collect();
            eprintln!("--- last {} lines of rustfs server stdout (oldest first) ---", lines.len());
            for line in lines.iter().rev() {
                eprintln!("{line}");
            }
            eprintln!("--- end rustfs stdout dump ---");
        }

        let _ = timeout(Duration::from_secs(READ_CACHE_DEADLINE_SECS), async {
            server_process.kill_and_wait().await;
        })
        .await;
        result
    }

    #[tokio::test]
    async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        crate::common::init_logging();
        run_read_cache_disabled_round_trip()
            .await
            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })
    }
}

// CMPTST-34: OPEN-time client attrs preservation across the streaming
// multipart write path. The payload crosses the 5 MiB part-size
// boundary so the driver transitions Buffering -> Streaming and
// finalises via CompleteMultipartUpload. The OPEN-supplied mtime and
// permissions must reach the resulting object as x-amz-meta-mtime and
// x-amz-meta-mode. The S3 client connects to the same rustfs process
// the shared-server suite already drives.
pub(crate) mod cmptst_34 {
    use super::*;

    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-34";
    const REQUESTED_MTIME: u32 = 1_715_000_010;
    const REQUESTED_MODE: u32 = 0o600;

    pub(crate) async fn run_open_attrs_round_trip_multipart(sftp: &SftpSession, s3: &S3Client) -> Result<()> {
        info!("{COMPLIANCE_TEST_OUTPUT_ID}: OPEN with mtime + mode, multi-part payload, streaming path");
        let bucket = "complopenattrsmpbucket";
        let bucket_path = format!("/{bucket}");
        sftp.create_dir(&bucket_path).await?;

        let path = format!("/{bucket}/attr-mp.bin");
        // 6 MiB exceeds the 5 MiB part-size boundary so the streaming
        // path runs at least one full UploadPart before the CLOSE-time
        // CompleteMultipartUpload finalises the object.
        let payload = vec![0xA5u8; 6 * 1024 * 1024];

        let client_attrs = FileAttributes {
            mtime: Some(REQUESTED_MTIME),
            atime: Some(REQUESTED_MTIME),
            permissions: Some(REQUESTED_MODE),
            ..FileAttributes::default()
        };

        let mut writer = sftp
            .open_with_flags_and_attributes(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE, client_attrs)
            .await?;
        writer.write_all(&payload).await?;
        writer.flush().await?;
        writer.shutdown().await?;

        let head = s3
            .head_object()
            .bucket(bucket)
            .key("attr-mp.bin")
            .send()
            .await
            .map_err(|e| anyhow!("S3 HeadObject failed: {e:?}"))?;
        let content_length = head.content_length().unwrap_or(0);
        if content_length != payload.len() as i64 {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} unexpected size: got {content_length} bytes"));
        }
        let metadata = head
            .metadata()
            .ok_or_else(|| anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} HeadObject returned no metadata map"))?;
        let mtime_value = metadata
            .get("mtime")
            .ok_or_else(|| anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mtime key missing on the object"))?;
        if mtime_value != &REQUESTED_MTIME.to_string() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mtime mismatch: got {mtime_value}"));
        }
        let mode_value = metadata
            .get("mode")
            .ok_or_else(|| anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mode key missing on the object"))?;
        if mode_value != &REQUESTED_MODE.to_string() {
            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mode mismatch: got {mode_value}"));
        }

        sftp.remove_file(&path).await?;
        sftp.remove_dir(&bucket_path).await?;
        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: multipart upload preserved mtime + mode end to end");
        Ok(())
    }
}

// Shared parameters for CMPTST-32 (cache enabled) and CMPTST-33 (cache
// disabled). Both cases seed the same fixture and download it
// end-to-end, then assert byte-count and SHA256 against the
// deterministic seed pattern. The sole difference between the two cases
// is the value of RUSTFS_SFTP_READ_CACHE_WINDOW_BYTES passed to the
// server. Backend call-count assertions are covered at the unit-test
// layer in crates/protocols/src/sftp/read.rs against a DummyBackend
// with explicit response queues. The e2e cases here exist to verify
// byte-correctness under both cache modes against a real ecstore
// backend, since that is the operator-visible regression risk.
const READ_CACHE_FIXTURE_BYTES: u64 = 8 * 1024 * 1024;
const READ_CACHE_DEADLINE_SECS: u64 = 120;

/// Shared body for CMPTST-32 and CMPTST-33. Waits for the SFTP port
/// to come up, seeds the fixture via multipart upload, downloads it
/// end-to-end via streaming SHA256, and asserts byte-count plus
/// SHA256 equality against the deterministic seed pattern. The two
/// cases differ only in the cache window the server was spawned with,
/// which is recorded in case_name for log triage.
async fn run_read_cache_byte_correctness(
    sftp_port: u16,
    sftp_address: &str,
    s3_endpoint: &str,
    bucket: &str,
    case_name: &str,
) -> Result<()> {
    ProtocolTestEnvironment::wait_for_port_ready(sftp_port, 30)
        .await
        .map_err(|e| anyhow!("{}", e))?;

    let s3 = build_test_s3_client(s3_endpoint);
    wait_for_s3_ready(&s3, S3_READY_ATTEMPTS).await?;

    let key = "fixture.bin";
    s3.create_bucket()
        .bucket(bucket)
        .send()
        .await
        .map_err(|e| anyhow!("S3 CreateBucket {bucket} failed: {e:?}"))?;

    info!("{case_name}: seeding {} MiB fixture", READ_CACHE_FIXTURE_BYTES / (1024 * 1024));
    seed_large_via_multipart(&s3, bucket, key, READ_CACHE_FIXTURE_BYTES).await?;
    let expected_sha = calculate_pattern_sha256(READ_CACHE_FIXTURE_BYTES, THRASH_PATTERN_MULTIPLIER);

    let path = format!("/{bucket}/{key}");
    let (_handle, sftp) = connect_sftp_to(sftp_address).await?;
    let download_t0 = Instant::now();
    let (bytes, sha) = streaming_sha256_download(&sftp, &path).await?;
    info!("{case_name}: download finished in {:?}", download_t0.elapsed());

    if bytes != READ_CACHE_FIXTURE_BYTES {
        return Err(anyhow!(
            "{case_name} byte-count mismatch: read {bytes} bytes, expected {READ_CACHE_FIXTURE_BYTES}"
        ));
    }
    if sha != expected_sha {
        return Err(anyhow!("{case_name} SHA256 mismatch on {path}"));
    }

    info!("PASS {case_name}: {} MiB downloaded byte-exact", bytes / (1024 * 1024));
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn cmptst29_eof_status_matches_protocol_constant() {
        // Compile-time check that the protocol enum the suite depends
        // on is still part of the russh-sftp surface. If the dependency
        // ships a breaking rename the assertion below catches it before
        // the end-to-end test runs.
        let code = StatusCode::Eof;
        assert_eq!(code as u32, 1);
    }
}