use std::{
    io::{Read, Write},
    process::{Command, Stdio},
    thread,
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use testcontainers::core::error::{ClientError, WaitContainerError};
use testcontainers::{GenericImage, ImageExt, TestcontainersError, runners::SyncRunner};

use super::{
    container_process::{
        docker_text, docker_text_with_timeout, docker_text_with_timeout_and_limits,
    },
    container_selection::Selection,
};

const IMAGE: &str = "wireguard-admin-alpine:3.22.7";
const IMAGE_ID: &str = "sha256:f41cd6b848d8280d4d86bcaceea962e1b3e5a73e9496c6f9d7929470d0e874d5";
const NETWORK_IPV4: &str = "192.168.253.0/24";
const CONTAINER_STARTUP_TIMEOUT: Duration = Duration::from_secs(12 * 60);
const FIXTURE_COMMAND_TIMEOUT_SECONDS: &str = "15";
const FIXTURE_TRANSPORT_TIMEOUT: Duration = Duration::from_secs(20);
const DRIVER_TRANSPORT_TIMEOUT: Duration = Duration::from_secs(180);
const COMMAND_STDOUT_LIMIT: usize = 1024;
const NAMESPACE_STDOUT_LIMIT: usize = 128;
const COMMAND_STDERR_LIMIT: usize = 1024;
const MAX_PHASE_REQUEST_BYTES: usize = 4096;

#[derive(Clone, Debug)]
pub struct ProbeNames {
    pub network: String,
    pub server: String,
    pub baseline: String,
    pub client: String,
}

impl ProbeNames {
    pub fn unique() -> Self {
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        let suffix = format!("{}-{nonce:x}", std::process::id());
        Self {
            network: format!("wga-m83-net-{suffix}"),
            server: format!("wga-m83-server-{suffix}"),
            baseline: format!("wga-m85-baseline-{suffix}"),
            client: format!("wga-m83-client-{suffix}"),
        }
    }
}

pub fn run_probe(names: &ProbeNames) -> Result<(), &'static str> {
    let selection = Selection::from_process_environment()?;
    verify_daemon(&selection)?;
    verify_cached_image(&selection)?;

    let network = OwnedNetwork::create(&selection, names.network.clone())?;
    let server = start_fixture_container(&selection, names.server.as_str(), &network.name)?;
    let baseline = start_fixture_container(&selection, names.baseline.as_str(), &network.name)?;
    let client = start_fixture_container(&selection, names.client.as_str(), &network.name)?;

    verify_distinct_namespaces(&server, &baseline)?;
    verify_distinct_namespaces(&server, &client)?;
    verify_distinct_namespaces(&baseline, &client)?;
    verify_wireguard_and_ipv4(&server)?;
    verify_wireguard_and_ipv4(&client)?;

    run_application_smoke(&server, &baseline, &client)?;

    client.remove()?;
    baseline.remove()?;
    server.remove()?;
    network.remove()
}

fn run_application_smoke(
    server: &OwnedContainer,
    baseline_client: &OwnedContainer,
    client: &OwnedContainer,
) -> Result<(), &'static str> {
    exec_smoke_command(
        server,
        &["ip", "link", "set", "dev", "wg-m83-probe", "up"],
        "Fixture WireGuard link setup failed",
    )?;
    exec_smoke_command(
        server,
        &[
            "ip",
            "-4",
            "address",
            "add",
            "10.20.0.1/24",
            "dev",
            "wg-m83-probe",
        ],
        "Fixture IPv4 address setup failed",
    )?;

    let driver = std::env::var_os("WIREGUARD_ADMIN_TEST_DRIVER")
        .ok_or("musl test driver path was not explicitly selected")?;
    let driver = std::path::PathBuf::from(driver);
    let metadata = std::fs::symlink_metadata(&driver)
        .map_err(|_| "musl test driver could not be inspected")?;
    if metadata.file_type().is_symlink()
        || !metadata.is_file()
        || metadata.len() > 256 * 1024 * 1024
    {
        return Err("musl test driver is not a bounded regular file");
    }

    let endpoint = format!("--host={}", server.selection.endpoint);
    let remote_driver = "/tmp/wga-m8-driver";
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            driver
                .to_str()
                .ok_or("musl test driver path is not UTF-8")?,
            &format!("{}:{remote_driver}", server.container().id()),
        ],
        Duration::from_secs(180),
    )
    .map_err(|category| match category {
        "Docker command timed out" => "Application fixture driver staging timed out",
        _ => "Application fixture driver staging failed",
    })?;
    exec_smoke_command(
        server,
        &["chmod", "0700", remote_driver],
        "Application fixture driver permission setup failed",
    )?;

    let command = vec![
        endpoint.as_str(),
        "exec",
        "-i",
        "--env=WIREGUARD_ADMIN_FIXTURE_ROOT=/tmp/wireguard-admin-fixture",
        server.container().id(),
        "/bin/busybox",
        "timeout",
        "-s",
        "KILL",
        "150",
        remote_driver,
        "--ignored",
        "--exact",
        "container_driver::alpine_scenario_driver",
        "--test-threads=1",
        "--nocapture",
    ];
    let empty = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    if !empty.contains("phase=observe status=complete interface=wg-m83-probe peers=0") {
        return Err("real runtime reader did not report the initial empty peer set");
    }
    let baseline = invoke_driver(&command, b"BASELINE\nEND\n")?;
    let baseline_key = phase_public_key(&baseline, "baseline_apply")?;
    let unconnected = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    let unconnected_peer = unconnected
        .lines()
        .find(|line| line.starts_with(&format!("peer={baseline_key} ")))
        .ok_or("real runtime reader omitted the unconnected baseline peer")?;
    if !peer_has_exact_ipv4_routes(unconnected_peer, "10.20.0.10/32")
        || !unconnected_peer.contains("handshake=none")
        || !unconnected_peer.contains("received=0")
        || !unconnected_peer.contains("transmitted=0")
    {
        return Err("real runtime reader did not report the baseline pre-handshake state");
    }
    let baseline_temp = tempfile::Builder::new()
        .prefix("wga-m85-baseline-")
        .tempdir()
        .map_err(|_| "private baseline staging directory could not be created")?;
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(baseline_temp.path(), std::fs::Permissions::from_mode(0o700))
        .map_err(|_| "private baseline staging directory permissions failed")?;
    let baseline_file = baseline_temp.path().join("baseline-client.conf");
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            &format!(
                "{}:/tmp/wireguard-admin-fixture/baseline-client.conf",
                server.container().id()
            ),
            baseline_file
                .to_str()
                .ok_or("private baseline path was not UTF-8")?,
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "synthetic baseline client transfer failed")?;
    let baseline_metadata = std::fs::symlink_metadata(&baseline_file)
        .map_err(|_| "synthetic baseline client file could not be inspected")?;
    use std::os::unix::fs::MetadataExt;
    if baseline_metadata.file_type().is_symlink()
        || !baseline_metadata.is_file()
        || baseline_metadata.len() > 64 * 1024
        || baseline_metadata.mode() & 0o777 != 0o600
    {
        return Err("synthetic baseline client was not a bounded mode-0600 regular file");
    }
    let baseline_bytes = std::fs::read(&baseline_file)
        .map_err(|_| "synthetic baseline client file could not be read")?;
    let (baseline_address, baseline_endpoint) = client_connection_material(&baseline_bytes)?;
    const REMOTE_BASELINE_DELIVERY: &str = "/tmp/wgb85.conf";
    const REMOTE_BASELINE_SETCONF: &str = "/tmp/wgb85-set.conf";
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            baseline_file
                .to_str()
                .ok_or("private baseline path was not UTF-8")?,
            &format!(
                "{}:{REMOTE_BASELINE_DELIVERY}",
                baseline_client.container().id()
            ),
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "synthetic baseline client transfer failed")?;
    verify_private_container_file(baseline_client, REMOTE_BASELINE_DELIVERY)?;
    strip_client_config(
        baseline_client,
        REMOTE_BASELINE_DELIVERY,
        REMOTE_BASELINE_SETCONF,
    )?;
    verify_private_container_file(baseline_client, REMOTE_BASELINE_SETCONF)?;
    setup_client_interface(
        baseline_client,
        "wg-m85-baseline",
        REMOTE_BASELINE_SETCONF,
        &baseline_address,
        &baseline_endpoint,
    )?;
    exec_success(
        baseline_client,
        ["ping", "-4", "-c", "3", "-W", "2", "10.20.0.1"],
    )
    .map_err(|_| "synthetic baseline IPv4 tunnel traffic failed")?;
    let mut baseline_observed = None;
    for _ in 0..10 {
        let observation = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        let peer = observation
            .lines()
            .find(|line| line.starts_with(&format!("peer={baseline_key} ")));
        if peer.is_some_and(|line| {
            peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
                && line.contains("handshake=present")
                && counter_is_positive(line, "received=")
                && counter_is_positive(line, "transmitted=")
        }) {
            baseline_observed = Some(observation);
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if baseline_observed.is_none() {
        return Err("real runtime reader did not observe the baseline peer and traffic");
    }

    // Create, Commit and first delivery share this one Application session.
    // Only fixed phase requests cross stdin; key/config bytes remain in private
    // fixture files and are transferred separately by docker cp.
    let report = docker_exchange_report(
        &command,
        b"CREATE\tM8 smoke client\nCOMMIT\nDELIVER\nEND\n",
        DRIVER_TRANSPORT_TIMEOUT,
        2048,
        1024,
    )
    .map_err(|category| match category {
        "Docker command timed out" => {
            "Application fixture session timed out; in-container completion is uncertain"
        }
        "In-container fixture command timed out" => {
            "Application fixture driver exceeded its in-container deadline"
        }
        "Docker command output exceeded its limit" => {
            "Application fixture driver output exceeded its limit"
        }
        _ => "Application fixture driver protocol failed",
    })?;
    let phase_report = format!("{}\n{}", report.stdout, report.stderr);
    if !driver_ran_exactly_one_test(&phase_report) {
        return Err("Application fixture driver selection did not execute exactly one test");
    }
    if report.failure_category.is_some() {
        return Err(driver_phase_failure_category(&phase_report));
    }
    for marker in [
        "phase=fixture_bootstrap status=ok",
        "phase=create status=ok public_key=",
        "phase=commit status=ok",
        "phase=delivery status=ok activation_warning=true",
        "phase=end status=ok",
    ] {
        if !phase_report.contains(marker) {
            return Err("Application fixture report omitted a required creation phase");
        }
    }
    let public_key = phase_public_key(&phase_report, "create")?;
    let baseline_archive = phase_archive_name(&phase_report, "commit")?;

    let temp = tempfile::Builder::new()
        .prefix("wga-m85-delivery-")
        .tempdir()
        .map_err(|_| "private delivery staging directory could not be created")?;
    std::fs::set_permissions(temp.path(), std::fs::Permissions::from_mode(0o700))
        .map_err(|_| "private delivery staging directory permissions failed")?;
    let delivered = temp.path().join("delivered.conf");
    let endpoint = format!("--host={}", server.selection.endpoint);
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            &format!(
                "{}:/tmp/wireguard-admin-fixture/first-delivery.conf",
                server.container().id()
            ),
            delivered
                .to_str()
                .ok_or("private delivery path was not UTF-8")?,
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "first delivery transfer from server fixture failed")?;
    let metadata = std::fs::symlink_metadata(&delivered)
        .map_err(|_| "first delivery file could not be inspected")?;
    if metadata.file_type().is_symlink()
        || !metadata.is_file()
        || metadata.len() > 64 * 1024
        || metadata.mode() & 0o777 != 0o600
    {
        return Err("first delivery file was not a bounded mode-0600 regular file");
    }
    let bytes = std::fs::read(&delivered).map_err(|_| "first delivery file could not be read")?;
    let (address, endpoint_addr) = client_connection_material(&bytes)?;
    // wg-quick strip derives an interface name from the config basename and
    // rejects names longer than Linux's 15-character interface limit.
    let remote_delivery = "/tmp/wgm85.conf";
    let remote_setconf = "/tmp/wgm85-set.conf";
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            delivered
                .to_str()
                .ok_or("private client path was not UTF-8")?,
            &format!("{}:{remote_delivery}", client.container().id()),
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "first delivery transfer to client fixture failed")?;
    verify_private_container_file(client, remote_delivery)?;
    verify_container_copy_matches(
        client,
        remote_delivery,
        &temp.path().join("delivered-copy.conf"),
        &bytes,
        &endpoint,
    )?;
    strip_client_config(client, remote_delivery, remote_setconf)?;
    verify_private_container_file(client, remote_setconf)?;
    setup_client_interface(
        client,
        "wg-m85-client",
        remote_setconf,
        &address,
        &endpoint_addr,
    )?;

    let before = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    if !before.contains("phase=observe status=complete interface=wg-m83-probe peers=1")
        || !before.lines().any(|line| {
            line.starts_with(&format!("peer={baseline_key} "))
                && peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
        })
        || before.contains(&format!("peer={public_key} "))
    {
        return Err("real runtime reader did not report only the baseline peer before Apply");
    }
    for _ in 0..3 {
        expect_fixture_ping_failure(client, "10.20.0.1")?;
    }
    let apply = invoke_driver(
        &command,
        b"APPLY\t/tmp/wireguard-admin-fixture/server.conf\nEND\n",
    )?;
    if !apply.contains("phase=fixture_apply status=verified") {
        return Err("fixture Apply did not pass real runtime peer verification");
    }
    let post_apply = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    let baseline_post_apply = post_apply
        .lines()
        .find(|line| line.starts_with(&format!("peer={baseline_key} ")))
        .filter(|line| peer_has_exact_ipv4_routes(line, "10.20.0.10/32"))
        .and_then(peer_counters)
        .ok_or("real runtime reader omitted the baseline peer after new-client Apply")?;
    // Reapply the private client peer configuration after the fixture server
    // setconf. This explicitly reestablishes a handshake from the baseline
    // client instead of trusting counters from the pre-Apply session.
    eprintln!("phase=baseline_reconnect status=started reason=fixture_server_setconf");
    exec_success(
        baseline_client,
        ["wg", "setconf", "wg-m85-baseline", REMOTE_BASELINE_SETCONF],
    )
    .map_err(|_| "baseline client fixture reconnect failed after new-client Apply")?;
    eprintln!("phase=baseline_reconnect status=complete");
    let (transmitted, received) = bounded_ping_summary(baseline_client, "10.20.0.1")?;
    if received == 0 {
        report_baseline_failure(
            server,
            &command,
            baseline_key,
            "after baseline fixture reconnect",
        )?;
        eprintln!(
            "baseline ping after fixture reconnect: transmitted={transmitted} received={received}"
        );
        return Err(
            "pre-existing baseline client stopped after new-client Apply and fixture reconnect",
        );
    }
    let mut baseline_after_apply = None;
    for _ in 0..10 {
        let report = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        if let Some(line) = report
            .lines()
            .find(|line| line.starts_with(&format!("peer={baseline_key} ")))
            && peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
            && line.contains("handshake=present")
            && peer_counters(line).is_some_and(|after| {
                after.0 > baseline_post_apply.0 && after.1 > baseline_post_apply.1
            })
        {
            baseline_after_apply = Some(report);
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if baseline_after_apply.is_none() {
        return Err(
            "real runtime reader did not observe fresh baseline traffic after new-client Apply",
        );
    }
    let before_traffic = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    let pre_traffic_peer = before_traffic
        .lines()
        .find(|line| line.starts_with(&format!("peer={public_key} ")))
        .ok_or("real runtime reader omitted the applied peer before traffic")?;
    let pre_traffic_counters = peer_counters(pre_traffic_peer)
        .ok_or("real runtime reader counters were unavailable before traffic")?;
    exec_success(client, ["ping", "-4", "-c", "3", "-W", "2", "10.20.0.1"])
        .map_err(|_| "client IPv4 tunnel traffic failed after fixture Apply")?;
    let mut observed = None;
    for _ in 0..10 {
        let report = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        if let Some(line) = report
            .lines()
            .find(|line| line.starts_with(&format!("peer={public_key} ")))
            && peer_has_exact_ipv4_routes(line, &address)
            && line.contains("handshake=present")
            && peer_counters(line).is_some_and(|after| {
                after.0 > pre_traffic_counters.0 && after.1 > pre_traffic_counters.1
            })
        {
            observed = Some(report);
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if observed.is_none() {
        let last = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        if let Some(peer) = last
            .lines()
            .find(|line| line.starts_with(&format!("peer={public_key} ")))
        {
            eprintln!("Rustix reader final new-peer sample: {peer}");
        }
    }
    let observed = observed.ok_or("real runtime reader did not observe the client handshake")?;
    let peer = observed
        .lines()
        .find(|line| line.starts_with(&format!("peer={public_key} ")))
        .ok_or("real runtime reader omitted the created client")?;
    if !peer_has_exact_ipv4_routes(peer, &address)
        || !peer.contains("handshake=present")
        || !counter_is_positive(peer, "received=")
        || !counter_is_positive(peer, "transmitted=")
        || !endpoint_addr.contains(":51820")
    {
        return Err("real runtime reader reported incomplete IPv4 tunnel state");
    }

    // Commit the deletion through the real Application while the fixture's
    // running interface still has the peer. The request carries only the
    // public identity and address; key/config material stays in private files.
    let pre_delete_counters = peer_counters(peer)
        .ok_or("real runtime reader counters were unavailable before deletion")?;
    let delete_request = format!("DELETE\t{public_key}\t{address}\nEND\n");
    let deletion = invoke_driver(&command, delete_request.as_bytes())?;
    if !deletion.contains(&format!(
        "phase=delete status=saved public_key={public_key} current_record=removed backup_record=retained address_reserved=true"
    )) {
        return Err("Application deletion did not preserve the backup and address reservation");
    }
    let restore_archive = phase_archive_name(&deletion, "delete")?;

    let saved_but_live = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    if !saved_but_live.contains("phase=observe status=complete interface=wg-m83-probe peers=2")
        || !saved_but_live.lines().any(|line| {
            line.starts_with(&format!("peer={public_key} "))
                && peer_has_exact_ipv4_routes(line, &address)
        })
        || !saved_but_live.lines().any(|line| {
            line.starts_with(&format!("peer={baseline_key} "))
                && peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
        })
    {
        return Err("saved deletion changed the live peer set before explicit Apply");
    }
    exec_success(client, ["ping", "-4", "-c", "2", "-W", "2", "10.20.0.1"])
        .map_err(|_| "deleted client's live IPv4 access stopped before explicit Apply")?;
    exec_success(
        baseline_client,
        ["ping", "-4", "-c", "3", "-W", "2", "10.20.0.1"],
    )
    .map_err(|_| "pre-existing baseline client stopped before explicit Apply")?;
    let mut continued_access = None;
    for _ in 0..10 {
        let report = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        let live_peer = report
            .lines()
            .find(|line| line.starts_with(&format!("peer={public_key} ")));
        let baseline_peer = report
            .lines()
            .find(|line| line.starts_with(&format!("peer={baseline_key} ")));
        if live_peer.is_some_and(|line| {
            peer_counters(line).is_some_and(|after| {
                after.0 > pre_delete_counters.0 && after.1 > pre_delete_counters.1
            })
        }) && baseline_peer.is_some_and(|line| {
            peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
                && line.contains("handshake=present")
                && counter_is_positive(line, "received=")
                && counter_is_positive(line, "transmitted=")
        }) {
            continued_access = Some(report);
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if continued_access.is_none() {
        return Err("fresh tunnel traffic did not prove access persisted after saved deletion");
    }

    let apply_request =
        format!("APPLY\t/tmp/wireguard-admin-fixture/server.conf\t{address}\nEND\n");
    let apply = invoke_driver(&command, apply_request.as_bytes())?;
    if !apply.contains("phase=fixture_apply status=verified address_released=true") {
        return Err("explicit fixture Apply did not verify deletion or release the address");
    }
    let after_apply = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    let baseline_after_apply = after_apply
        .lines()
        .find(|line| line.starts_with(&format!("peer={baseline_key} ")))
        .filter(|line| peer_has_exact_ipv4_routes(line, "10.20.0.10/32"));
    if !after_apply.contains("phase=observe status=complete interface=wg-m83-probe peers=1")
        || after_apply.contains(&format!("peer={public_key} "))
        || baseline_after_apply.is_none()
    {
        return Err(
            "real runtime reader did not verify deleted peer removal and baseline retention",
        );
    }
    let baseline_after_apply = baseline_after_apply
        .and_then(peer_counters)
        .ok_or("baseline counters were unavailable after deletion Apply")?;
    // These are new attempts after the successful runtime verification. A
    // separate baseline reconnect and successful probe guard against setup
    // failure after the full server peer-set replacement.
    for _ in 0..3 {
        expect_fixture_ping_failure(client, "10.20.0.1")?;
    }
    eprintln!("phase=baseline_reconnect status=started reason=deletion_apply");
    exec_success(
        baseline_client,
        ["wg", "setconf", "wg-m85-baseline", REMOTE_BASELINE_SETCONF],
    )
    .map_err(|_| "baseline client fixture reconnect failed after deletion Apply")?;
    eprintln!("phase=baseline_reconnect status=complete reason=deletion_apply");
    let (transmitted, received) = bounded_ping_summary(baseline_client, "10.20.0.1")?;
    if received == 0 {
        report_baseline_failure(
            server,
            &command,
            baseline_key,
            "after deletion Apply baseline reconnect",
        )?;
        eprintln!(
            "baseline ping after deletion Apply: transmitted={transmitted} received={received}"
        );
        return Err(
            "pre-existing baseline client stopped after explicit Apply and fixture reconnect",
        );
    }
    let mut baseline_verified = false;
    for _ in 0..10 {
        let report = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        if report.contains("phase=observe status=complete interface=wg-m83-probe peers=1")
            && !report.contains(&format!("peer={public_key} "))
            && report.lines().any(|line| {
                line.starts_with(&format!("peer={baseline_key} "))
                    && peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
                    && line.contains("handshake=present")
                    && peer_counters(line).is_some_and(|after| {
                        after.0 > baseline_after_apply.0 && after.1 > baseline_after_apply.1
                    })
            })
        {
            baseline_verified = true;
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if !baseline_verified {
        return Err(
            "real runtime reader did not verify fresh baseline traffic after deletion Apply",
        );
    }

    // Restore the exact pre-delete archive emitted by this Application Commit.
    // It contains the complete prior client set and vault while runtime is
    // currently baseline-only. Restore must replace saved files only.
    let restore_request = format!("RESTORE\t{restore_archive}\t{address}\nEND\n");
    let restore = invoke_driver(&command, restore_request.as_bytes())?;
    if !restore.contains(&format!(
        "phase=restore status=confirmed archive={restore_archive} reason=pre-commit precautionary_backup=pre-restore settings=true server=true vault=true target_client_present=true pending_address_reserved=false runtime_unchanged=true"
    )) {
        return Err("password-free restore did not verify the exact archive and precautionary backup");
    }
    let saved_restored_live = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    if !saved_restored_live.contains("phase=observe status=complete interface=wg-m83-probe peers=1")
        || saved_restored_live.contains(&format!("peer={public_key} "))
        || !saved_restored_live.lines().any(|line| {
            line.starts_with(&format!("peer={baseline_key} "))
                && peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
        })
    {
        return Err("restore changed the live peer set before explicit fixture Apply");
    }
    for _ in 0..3 {
        expect_fixture_ping_failure(client, "10.20.0.1")?;
    }

    let restore_apply = invoke_driver(
        &command,
        format!("APPLY\t/tmp/wireguard-admin-fixture/server.conf\t{address}\nEND\n").as_bytes(),
    )?;
    if !restore_apply.contains("phase=fixture_apply status=verified address_released=true") {
        return Err(
            "explicit fixture Apply after restore did not verify the restored saved peer set",
        );
    }
    let before_restored_traffic = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    let restored_before = before_restored_traffic
        .lines()
        .find(|line| line.starts_with(&format!("peer={public_key} ")))
        .filter(|line| peer_has_exact_ipv4_routes(line, &address))
        .and_then(peer_counters)
        .ok_or("real runtime reader omitted the restored peer after explicit Apply")?;
    eprintln!("phase=client_reconnect status=started reason=restore_apply");
    exec_success(client, ["wg", "setconf", "wg-m85-client", remote_setconf])
        .map_err(|_| "restored client fixture reconnect failed after restore Apply")?;
    eprintln!("phase=client_reconnect status=complete reason=restore_apply");
    let (transmitted, received) = bounded_ping_summary(client, "10.20.0.1")?;
    if received == 0 {
        return Err("restored client did not renew IPv4 tunnel traffic after explicit Apply");
    }
    let mut restored_traffic = false;
    for _ in 0..10 {
        let report = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        if report.contains("phase=observe status=complete interface=wg-m83-probe peers=2")
            && report.lines().any(|line| {
                line.starts_with(&format!("peer={public_key} "))
                    && peer_has_exact_ipv4_routes(line, &address)
                    && line.contains("handshake=present")
                    && peer_counters(line).is_some_and(|after| {
                        after.0 > restored_before.0 && after.1 > restored_before.1
                    })
            })
        {
            restored_traffic = true;
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if !restored_traffic {
        eprintln!("restored client ping summary: transmitted={transmitted} received={received}");
        return Err("real runtime reader did not observe fresh restored-client IPv4 traffic");
    }

    // Restoring the earlier baseline archive removes the restored client from
    // saved state while it remains active in the kernel. Its address must stay
    // reserved until another explicit, successfully verified fixture Apply.
    let reservation_restore_request = format!("RESTORE\t{baseline_archive}\t{address}\nEND\n");
    let reservation_restore = invoke_driver(&command, reservation_restore_request.as_bytes())?;
    if !reservation_restore.contains(&format!(
        "phase=restore status=confirmed archive={baseline_archive} reason=pre-commit precautionary_backup=pre-restore settings=true server=true vault=true target_client_present=false pending_address_reserved=true runtime_unchanged=true"
    )) {
        return Err("restore released a pending client address before explicit Apply");
    }
    let live_after_second_restore = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    if !live_after_second_restore
        .contains("phase=observe status=complete interface=wg-m83-probe peers=2")
        || !live_after_second_restore.contains(&format!("peer={public_key} "))
    {
        return Err("second restore changed the live peer set before explicit fixture Apply");
    }
    exec_success(client, ["wg", "setconf", "wg-m85-client", remote_setconf])
        .map_err(|_| "restored client fixture reconnect failed before reservation Apply")?;
    exec_success(client, ["ping", "-4", "-c", "2", "-W", "2", "10.20.0.1"])
        .map_err(|_| "restored client access did not remain live before reservation Apply")?;
    let release_apply = invoke_driver(
        &command,
        format!("APPLY\t/tmp/wireguard-admin-fixture/server.conf\t{address}\nEND\n").as_bytes(),
    )?;
    if !release_apply.contains("phase=fixture_apply status=verified address_released=true") {
        return Err("explicit fixture Apply did not verify reservation release after restore");
    }
    let released_runtime = invoke_driver(&command, b"OBSERVE\nEND\n")?;
    if !released_runtime.contains("phase=observe status=complete interface=wg-m83-probe peers=1")
        || released_runtime.contains(&format!("peer={public_key} "))
    {
        return Err("explicit Apply did not remove the restored client after reservation restore");
    }
    for _ in 0..3 {
        expect_fixture_ping_failure(client, "10.20.0.1")?;
    }

    let imported = invoke_driver(&command, b"IMPORT_REISSUE\nEND\n")?;
    for marker in [
        "phase=import_reissue status=verified matched=true",
        "conflict_visible=true invalid_visible=true originals_unchanged=true",
        "wrong_password_blocked=true conflict_reissue_blocked=true reissue_mode=0600",
    ] {
        if !imported.contains(marker) {
            return Err("Application import/reissue scenario omitted a required safety assertion");
        }
    }
    let import_temp = tempfile::Builder::new()
        .prefix("wga-m85-import-reissue-")
        .tempdir()
        .map_err(|_| "private import/reissue staging directory could not be created")?;
    std::fs::set_permissions(import_temp.path(), std::fs::Permissions::from_mode(0o700))
        .map_err(|_| "private import/reissue directory permissions failed")?;
    let reissued = import_temp.path().join("reissued-client.conf");
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            &format!(
                "{}:/tmp/wireguard-admin-fixture/import-reissue/reissued-client.conf",
                server.container().id()
            ),
            reissued
                .to_str()
                .ok_or("private reissue path was not UTF-8")?,
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "password-authorized imported client transfer failed")?;
    verify_private_host_file(&reissued, 64 * 1024)?;
    let reissued_bytes = std::fs::read(&reissued)
        .map_err(|_| "password-authorized imported client file could not be read")?;
    let (imported_address, imported_endpoint) = client_connection_material(&reissued_bytes)?;
    const REMOTE_REISSUED: &str = "/tmp/wgb85-imported.conf";
    const REMOTE_REISSUED_SETCONF: &str = "/tmp/wgb85-imported-set.conf";
    docker_text_with_timeout(
        &[
            &endpoint,
            "cp",
            reissued
                .to_str()
                .ok_or("private reissue path was not UTF-8")?,
            &format!("{}:{REMOTE_REISSUED}", client.container().id()),
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "password-authorized imported client transfer failed")?;
    verify_private_container_file(client, REMOTE_REISSUED)?;
    verify_container_copy_matches(
        client,
        REMOTE_REISSUED,
        &import_temp.path().join("client-copy.conf"),
        &reissued_bytes,
        &endpoint,
    )?;
    strip_client_config(client, REMOTE_REISSUED, REMOTE_REISSUED_SETCONF)?;
    verify_private_container_file(client, REMOTE_REISSUED_SETCONF)?;
    exec_success(baseline_client, ["ip", "link", "del", "wg-m85-baseline"])
        .map_err(|_| "baseline client fixture did not stop before imported-key reissue")?;
    let _ = exec_success(client, ["ip", "link", "del", "wg-m85-client"]);
    setup_client_interface(
        client,
        "wg-m85-client",
        REMOTE_REISSUED_SETCONF,
        &imported_address,
        &imported_endpoint,
    )?;
    exec_success(client, ["ping", "-4", "-c", "3", "-W", "2", "10.20.0.1"])
        .map_err(|_| "reissued imported client did not establish IPv4 tunnel traffic")?;
    let mut imported_traffic = false;
    for _ in 0..10 {
        let observation = invoke_driver(&command, b"OBSERVE\nEND\n")?;
        if observation.lines().any(|line| {
            line.starts_with(&format!("peer={baseline_key} "))
                && peer_has_exact_ipv4_routes(line, "10.20.0.10/32")
                && line.contains("handshake=present")
                && counter_is_positive(line, "received=")
                && counter_is_positive(line, "transmitted=")
        }) {
            imported_traffic = true;
            break;
        }
        thread::sleep(Duration::from_millis(250));
    }
    if !imported_traffic {
        return Err("real runtime reader did not observe traffic from the reissued imported file");
    }

    let failure = invoke_driver(&command, b"FAILURE_PREVENTS_APPLICATION\nEND\n")?;
    for marker in [
        "phase=reader_failures status=verified missing_interface=rejected deadline=timeout empty_or_partial=false",
        "phase=commit_backup_failure status=verified saved_files_unchanged=true draft_preserved=true runtime_unchanged=true archive_path_restored=true",
        "phase=restart_backup_failure status=verified apply_called=false restart_submitted=false address_reserved=true historical_state=unchanged runtime_unchanged=true archive_path_restored=true",
        "phase=failure_prevents_application status=complete",
    ] {
        if !failure.contains(marker) {
            return Err("backup-failure scenario omitted a required safety assertion");
        }
    }
    Ok(())
}

fn invoke_driver(command: &[&str], request: &[u8]) -> Result<String, &'static str> {
    let report = docker_exchange_report(command, request, DRIVER_TRANSPORT_TIMEOUT, 2048, 1024)
        .map_err(|_| "Application fixture phase exchange failed")?;
    let combined = format!("{}\n{}", report.stdout, report.stderr);
    if report.failure_category.is_some() {
        let progress = combined
            .split_whitespace()
            .filter(|part| {
                part.starts_with("phase=")
                    || part.starts_with("status=")
                    || part.starts_with("stage=")
                    || part.starts_with("step=")
                    || part.starts_with("settings_match=")
                    || part.starts_with("server_match=")
                    || part.starts_with("vault_match=")
                    || part.starts_with("target_presence_match=")
                    || part.starts_with("reservations_match=")
                    || part.starts_with("actual_reservation_count=")
                    || part.starts_with("expected_reservation_count=")
                    || part.starts_with("actual_target_reserved=")
                    || part.starts_with("expected_target_reserved=")
            })
            .collect::<Vec<_>>()
            .join(" ");
        eprintln!("Fixture driver reported bounded phase status: {progress}");
        return Err(driver_phase_failure_category(&combined));
    }
    if !driver_ran_exactly_one_test(&combined) {
        return Err("Application fixture phase driver failed");
    }
    Ok(combined)
}

fn phase_public_key<'a>(report: &'a str, phase: &str) -> Result<&'a str, &'static str> {
    let marker = format!("phase={phase} status=verified public_key=");
    let create_marker = format!("phase={phase} status=ok public_key=");
    report
        .split_once(&marker)
        .or_else(|| report.split_once(&create_marker))
        .and_then(|(_, tail)| tail.split_whitespace().next())
        .ok_or("fixture phase report omitted its public identity")
}

fn phase_archive_name<'a>(report: &'a str, phase: &str) -> Result<&'a str, &'static str> {
    let completed_status = match phase {
        "commit" => "ok",
        "delete" => "saved",
        _ => return Err("Application phase archive identity was invalid"),
    };
    let prefix = format!("phase={phase} ");
    let line = report
        .lines()
        .find(|line| {
            if !line.starts_with(&prefix) {
                return false;
            }
            let mut statuses = line
                .split_whitespace()
                .filter_map(|field| field.strip_prefix("status="));
            statuses.next() == Some(completed_status) && statuses.next().is_none()
        })
        .ok_or("Application phase omitted its archive identity")?;
    let mut names = line
        .split_whitespace()
        .filter_map(|field| field.strip_prefix("backup_name="));
    let name = names
        .next()
        .ok_or("Application phase omitted its archive identity")?;
    if names.next().is_some()
        || name.is_empty()
        || name.len() > 128
        || !name
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || b"-_.".contains(&byte))
        || !name.ends_with(".tar")
    {
        return Err("Application phase archive identity was invalid");
    }
    Ok(name)
}

fn verify_private_container_file(
    container: &OwnedContainer,
    path: &str,
) -> Result<(), &'static str> {
    let mode = exec_small(container, ["stat", "-c", "%a", path])?;
    if mode == "600" {
        Ok(())
    } else {
        Err("secret-bearing fixture config was not mode 0600 after transfer")
    }
}

fn verify_container_copy_matches(
    container: &OwnedContainer,
    remote_path: &str,
    local_path: &std::path::Path,
    expected_bytes: &[u8],
    endpoint: &str,
) -> Result<(), &'static str> {
    docker_text_with_timeout(
        &[
            endpoint,
            "cp",
            &format!("{}:{remote_path}", container.container().id()),
            local_path
                .to_str()
                .ok_or("private client verification path was not UTF-8")?,
        ],
        Duration::from_secs(20),
    )
    .map_err(|_| "client copy could not be returned for byte verification")?;
    let metadata = std::fs::symlink_metadata(local_path)
        .map_err(|_| "client copy verification file could not be inspected")?;
    use std::os::unix::fs::MetadataExt;
    if metadata.file_type().is_symlink()
        || !metadata.is_file()
        || metadata.len() != expected_bytes.len() as u64
        || metadata.mode() & 0o777 != 0o600
    {
        return Err("client copy verification file was not a private regular file");
    }
    let copied_bytes =
        std::fs::read(local_path).map_err(|_| "client copy verification file could not be read")?;
    if copied_bytes == expected_bytes {
        Ok(())
    } else {
        Err("client copy did not match the first-delivery bytes")
    }
}

fn verify_private_host_file(
    path: &std::path::Path,
    maximum_bytes: u64,
) -> Result<(), &'static str> {
    use std::os::unix::fs::MetadataExt;
    let metadata = std::fs::symlink_metadata(path)
        .map_err(|_| "private transferred client file could not be inspected")?;
    if metadata.file_type().is_symlink()
        || !metadata.is_file()
        || metadata.len() > maximum_bytes
        || metadata.mode() & 0o777 != 0o600
    {
        return Err("private transferred client file was not a bounded mode-0600 regular file");
    }
    Ok(())
}

fn strip_client_config(
    client: &OwnedContainer,
    source_path: &str,
    destination_path: &str,
) -> Result<(), &'static str> {
    // wg-quick strip parses the copied file and does not apply its hooks. The
    // fixed wrapper redirects private output to a mode-0600 file; config text is
    // never passed as an argument or emitted to the test logs.
    const STRIP_SCRIPT: &str = "umask 077; wg-quick strip \"$1\" > \"$2\" && chmod 600 \"$2\"";
    exec_success(
        client,
        [
            "/bin/busybox",
            "sh",
            "-c",
            STRIP_SCRIPT,
            "wga-strip",
            source_path,
            destination_path,
        ],
    )
    .map_err(|_| "copied client configuration could not be safely stripped")
}

fn setup_client_interface(
    client: &OwnedContainer,
    interface: &str,
    setconf_path: &str,
    address: &str,
    endpoint: &str,
) -> Result<(), &'static str> {
    if interface != "wg-m85-baseline" && interface != "wg-m85-client" {
        return Err("client fixture interface name was not recognized");
    }
    let endpoint_route = endpoint_host_route(endpoint)?;
    exec_success(
        client,
        ["ip", "link", "add", interface, "type", "wireguard"],
    )
    .map_err(|_| "client fixture WireGuard interface setup failed")?;
    // The endpoint has a more-specific route through the underlay before the
    // default route is sent through WireGuard.
    exec_success(
        client,
        [
            "ip",
            "-4",
            "route",
            "replace",
            &endpoint_route,
            "dev",
            "eth0",
        ],
    )
    .map_err(|_| "server endpoint underlay route could not be preserved")?;
    exec_success(client, ["wg", "setconf", interface, setconf_path])
        .map_err(|_| "delivered client configuration was rejected by WireGuard")?;
    exec_success(
        client,
        ["ip", "-4", "address", "add", address, "dev", interface],
    )
    .map_err(|_| "delivered IPv4 address could not be configured")?;
    exec_success(client, ["ip", "link", "set", "dev", interface, "up"])
        .map_err(|_| "client fixture WireGuard link setup failed")?;
    exec_success(
        client,
        ["ip", "-4", "route", "replace", "default", "dev", interface],
    )
    .map_err(|_| "delivered full-tunnel IPv4 route could not be configured")
}

fn endpoint_host_route(endpoint: &str) -> Result<String, &'static str> {
    let (host, port) = endpoint
        .rsplit_once(':')
        .ok_or("delivered client endpoint was malformed")?;
    if port != "51820" {
        return Err("delivered client endpoint did not use the fixture port");
    }
    let address = host
        .parse::<std::net::Ipv4Addr>()
        .map_err(|_| "delivered client endpoint was not IPv4")?;
    Ok(format!("{address}/32"))
}

fn expect_fixture_ping_failure(
    container: &OwnedContainer,
    destination: &str,
) -> Result<(), &'static str> {
    require_zero_ping_replies(bounded_ping_summary(container, destination))
}

fn require_zero_ping_replies(
    summary: Result<(u32, u32), &'static str>,
) -> Result<(), &'static str> {
    let (transmitted, received) = summary?;
    if transmitted == 3 && received == 0 {
        Ok(())
    } else if received > 0 {
        Err("client still reached the tunnel during a negative reachability probe")
    } else {
        Err("negative reachability probe did not transmit three packets")
    }
}

fn bounded_ping_summary(
    container: &OwnedContainer,
    destination: &str,
) -> Result<(u32, u32), &'static str> {
    // Keep ping's packet-loss summary even when it exits nonzero; the final
    // `true` only preserves that bounded output for diagnosis, while the
    // parsed receive count below remains a strict reachability assertion.
    const PING_SCRIPT: &str = "ping -q -4 -c 3 -W 2 \"$1\"; /bin/busybox true";
    let output = exec_bounded(
        container,
        &[
            "/bin/busybox",
            "sh",
            "-c",
            PING_SCRIPT,
            "wga-ping",
            destination,
        ],
        COMMAND_STDOUT_LIMIT,
        COMMAND_STDERR_LIMIT,
        "Baseline reachability probe failed",
    )?;
    let output = String::from_utf8(output).map_err(|_| "Baseline ping output was invalid")?;
    let summary = output
        .lines()
        .find(|line| line.contains("packets transmitted") && line.contains("packets received"))
        .ok_or("Baseline ping summary was absent")?;
    parse_ping_summary(summary)
}

fn parse_ping_summary(summary: &str) -> Result<(u32, u32), &'static str> {
    let mut fields = summary.split(',');
    let transmitted = fields
        .next()
        .filter(|field| field.contains("packets transmitted"))
        .and_then(|field| field.split_whitespace().next())
        .and_then(|value| value.parse::<u32>().ok())
        .ok_or("Baseline ping summary was malformed")?;
    let received = fields
        .next()
        .filter(|field| field.contains("packets received"))
        .and_then(|field| field.split_whitespace().next())
        .and_then(|value| value.parse::<u32>().ok())
        .ok_or("Baseline ping summary was malformed")?;
    Ok((transmitted, received))
}

fn report_baseline_failure(
    server: &OwnedContainer,
    command: &[&str],
    baseline_key: &str,
    stage: &str,
) -> Result<(), &'static str> {
    let report = invoke_driver(command, b"OBSERVE\nEND\n")?;
    let peer = report
        .lines()
        .find(|line| line.starts_with(&format!("peer={baseline_key} ")));
    if let Some(peer) = peer {
        let routes = peer
            .split_whitespace()
            .find_map(|part| part.strip_prefix("routes="))
            .unwrap_or("unavailable");
        let handshake = peer
            .split_whitespace()
            .find_map(|part| part.strip_prefix("handshake="))
            .unwrap_or("unavailable");
        let received = peer
            .split_whitespace()
            .find_map(|part| part.strip_prefix("received="))
            .unwrap_or("unavailable");
        let transmitted = peer
            .split_whitespace()
            .find_map(|part| part.strip_prefix("transmitted="))
            .unwrap_or("unavailable");
        eprintln!(
            "baseline peer {stage}: routes={routes} handshake={handshake} received={received} transmitted={transmitted}"
        );
    } else {
        eprintln!("baseline peer {stage}: absent from complete runtime observation");
    }

    let endpoints = exec_bounded(
        server,
        &["wg", "show", "wg-m83-probe", "endpoints"],
        COMMAND_STDOUT_LIMIT,
        COMMAND_STDERR_LIMIT,
        "Baseline endpoint observation failed",
    )?;
    let endpoints =
        String::from_utf8(endpoints).map_err(|_| "Baseline endpoint observation was invalid")?;
    let endpoint_learned = endpoints.lines().any(|line| {
        line.split_whitespace()
            .next()
            .is_some_and(|identity| identity == baseline_key)
            && line
                .split_whitespace()
                .nth(1)
                .is_some_and(|endpoint| endpoint != "(none)")
    });
    eprintln!("baseline server endpoint learned={endpoint_learned}");
    Ok(())
}

fn counter_is_positive(peer: &str, field: &str) -> bool {
    peer.split_whitespace()
        .find_map(|part| part.strip_prefix(field))
        .and_then(|value| value.parse::<u64>().ok())
        .is_some_and(|value| value > 0)
}

fn peer_counters(peer: &str) -> Option<(u64, u64)> {
    let received = peer
        .split_whitespace()
        .find_map(|part| part.strip_prefix("received="))?
        .parse::<u64>()
        .ok()?;
    let transmitted = peer
        .split_whitespace()
        .find_map(|part| part.strip_prefix("transmitted="))?
        .parse::<u64>()
        .ok()?;
    Some((received, transmitted))
}

fn client_connection_material(bytes: &[u8]) -> Result<(String, String), &'static str> {
    let text = std::str::from_utf8(bytes).map_err(|_| "delivered client file was not UTF-8")?;
    let mut section = "";
    let mut addresses = Vec::new();
    let mut endpoint = None;
    let mut ipv4_routes = std::collections::BTreeSet::new();
    for line in text.lines() {
        let trimmed = line.trim();
        if trimmed.starts_with('[') && trimmed.ends_with(']') {
            section = trimmed;
        }
        if let Some((name, value)) = trimmed.split_once('=') {
            match (section, name.trim()) {
                ("[Interface]", "Address") => {
                    for item in value.split(',') {
                        let address = item
                            .trim()
                            .parse::<ipnet::IpNet>()
                            .map_err(|_| "delivered client Address was malformed")?;
                        if let ipnet::IpNet::V4(network) = address {
                            addresses.push(network);
                        }
                    }
                }
                ("[Peer]", "Endpoint") => endpoint = Some(value.trim().to_owned()),
                ("[Peer]", "AllowedIPs") => {
                    for item in value.split(',') {
                        let route = item
                            .trim()
                            .parse::<ipnet::IpNet>()
                            .map_err(|_| "delivered client AllowedIPs was malformed")?;
                        if let ipnet::IpNet::V4(network) = route {
                            ipv4_routes.insert(network);
                        }
                    }
                }
                _ => {}
            }
        }
    }
    let expected_route = "0.0.0.0/0"
        .parse::<ipnet::Ipv4Net>()
        .map_err(|_| "expected IPv4 full-tunnel route was invalid")?;
    if ipv4_routes != [expected_route].into_iter().collect() {
        return Err("delivered client file did not contain the expected IPv4 full-tunnel route");
    }
    let endpoint = endpoint.ok_or("delivered client endpoint was absent")?;
    let _ = endpoint_host_route(&endpoint)?;
    if addresses.len() != 1 || addresses[0].prefix_len() != 32 {
        return Err("delivered client did not contain exactly one IPv4 /32 address");
    }
    Ok((addresses[0].to_string(), endpoint))
}

fn peer_has_exact_ipv4_routes(peer: &str, expected: &str) -> bool {
    let Some(routes) = peer
        .split_whitespace()
        .find_map(|part| part.strip_prefix("routes="))
    else {
        return false;
    };
    let actual = routes
        .split(',')
        .filter_map(|route| route.parse::<ipnet::IpNet>().ok())
        .filter_map(|route| match route {
            ipnet::IpNet::V4(route) => Some(route),
            ipnet::IpNet::V6(_) => None,
        })
        .collect::<std::collections::BTreeSet<_>>();
    let Ok(expected) = expected.parse::<ipnet::Ipv4Net>() else {
        return false;
    };
    actual == [expected].into_iter().collect()
}

#[derive(Debug, Eq, PartialEq)]
struct DockerExchangeReport {
    stdout: String,
    stderr: String,
    failure_category: Option<&'static str>,
}

fn docker_exchange_report(
    arguments: &[&str],
    request: &[u8],
    command_deadline: Duration,
    stdout_limit: usize,
    stderr_limit: usize,
) -> Result<DockerExchangeReport, &'static str> {
    if request.len() > MAX_PHASE_REQUEST_BYTES {
        return Err("fixture phase request exceeded its limit");
    }
    let mut child = Command::new("docker")
        .args(arguments)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .map_err(|_| "Docker CLI could not be started")?;
    let mut stdin = child.stdin.take().ok_or("Docker stdin unavailable")?;
    let stdout = child.stdout.take().ok_or("Docker stdout unavailable")?;
    let stderr = child.stderr.take().ok_or("Docker stderr unavailable")?;
    let stdout_reader = thread::spawn(move || read_bounded(stdout, stdout_limit));
    let stderr_reader = thread::spawn(move || read_bounded(stderr, stderr_limit));
    let deadline = Instant::now() + command_deadline;
    if stdin.write_all(request).is_err() {
        let _ = child.kill();
        let _ = child.wait();
        let _ = stdout_reader.join();
        let _ = stderr_reader.join();
        return Err("Docker fixture request write failed");
    }
    drop(stdin);
    let status = loop {
        match child.try_wait() {
            Ok(Some(status)) => break status,
            Ok(None) if Instant::now() >= deadline => {
                let _ = child.kill();
                let _ = child.wait();
                let _ = stdout_reader.join();
                let _ = stderr_reader.join();
                return Err("Docker command timed out");
            }
            Ok(None) => thread::sleep(Duration::from_millis(20)),
            Err(_) => {
                let _ = child.kill();
                let _ = child.wait();
                let _ = stdout_reader.join();
                let _ = stderr_reader.join();
                return Err("Docker command wait failed");
            }
        }
    };
    let (stdout, stdout_exceeded) = stdout_reader
        .join()
        .map_err(|_| "Docker stdout reader failed")?
        .map_err(|_| "Docker stdout read failed")?;
    let (stderr, stderr_exceeded) = stderr_reader
        .join()
        .map_err(|_| "Docker stderr reader failed")?
        .map_err(|_| "Docker stderr read failed")?;
    if stdout_exceeded || stderr_exceeded {
        return Err("Docker command output exceeded its limit");
    }
    let stdout = String::from_utf8(stdout).map_err(|_| "Docker returned non-UTF-8 output")?;
    let stderr = String::from_utf8(stderr).map_err(|_| "Docker returned non-UTF-8 output")?;
    let failure_category = (!status.success()).then(|| match status.code() {
        Some(124 | 137 | 143) => "In-container fixture command timed out",
        _ => "Docker command returned a nonzero status",
    });
    Ok(DockerExchangeReport {
        stdout,
        stderr,
        failure_category,
    })
}

fn read_bounded(mut reader: impl Read, limit: usize) -> std::io::Result<(Vec<u8>, bool)> {
    let mut captured = Vec::new();
    let mut buffer = [0_u8; 1024];
    let mut exceeded = false;
    loop {
        let count = reader.read(&mut buffer)?;
        if count == 0 {
            return Ok((captured, exceeded));
        }
        let remaining = limit.saturating_sub(captured.len());
        captured.extend_from_slice(&buffer[..count.min(remaining)]);
        exceeded |= count > remaining;
    }
}

fn driver_ran_exactly_one_test(output: &str) -> bool {
    output.contains("running 1 test")
        && output.contains("test container_driver::alpine_scenario_driver")
}

fn driver_phase_failure_category(output: &str) -> &'static str {
    if let Some(stage) = output.lines().find_map(|line| {
        line.split_once("phase=fixture_apply status=failed stage=")
            .map(|(_, tail)| tail.trim())
    }) {
        return match stage {
            "Preparation" => "Application fixture Apply failed during preparation",
            "Backup" => "Application fixture Apply failed during backup",
            "Service" => "Application fixture Apply command failed",
            "RuntimeRead" => "Application fixture Apply runtime observation failed",
            "Verification" => "Application fixture Apply peer verification failed",
            "Persistence" => "Application fixture Apply saved-state publication failed",
            _ => "Application fixture Apply failed at an unknown stage",
        };
    }
    let failed_phase = output
        .split_whitespace()
        .collect::<Vec<_>>()
        .windows(2)
        .filter_map(|pair| {
            (pair[1] == "status=started")
                .then(|| pair[0].strip_prefix("phase="))
                .flatten()
        })
        .next_back();
    match failed_phase {
        Some("create") => "Application fixture Create phase failed",
        Some("commit") => "Application fixture Commit phase failed",
        Some("delivery") => "Application fixture delivery phase failed",
        Some("delete") => "Application fixture saved deletion phase failed",
        Some("baseline_apply") => "Application fixture baseline setup failed",
        Some("fixture_apply") => "Application fixture apply phase failed",
        Some("failure_prevents_application") => "Application fixture backup-failure phase failed",
        Some("reader_failures") => "Application fixture runtime-reader failure checks failed",
        Some("commit_backup_failure") => "Application fixture Commit backup-failure check failed",
        Some("restart_backup_failure") => "Application fixture restart backup-failure check failed",
        Some("end") => "Application fixture protocol termination failed",
        _ => "Application fixture bootstrap failed",
    }
}

fn missing_driver_phase_category(output: &str) -> Option<&'static str> {
    let has = |marker: &str| output.split_once(marker).is_some();
    if !has("phase=fixture_bootstrap status=ok") {
        return Some("Application fixture report omitted bootstrap completion");
    }
    if !has("phase=create status=ok public_key=") {
        return Some("Application fixture report omitted Create completion");
    }
    if !has("phase=commit status=ok") {
        return Some("Application fixture report omitted Commit completion");
    }
    if !has("phase=delivery status=ok activation_warning=true") {
        return Some("Application fixture report omitted first delivery or activation warning");
    }
    if !has("current_record=removed backup_record=retained address_reserved=true") {
        return Some("Application fixture report omitted saved-deletion retention checks");
    }
    if !has("phase=restore status=confirmed")
        || !has("target_client_present=true pending_address_reserved=false")
        || !has("target_client_present=false pending_address_reserved=true")
        || !has("precautionary_backup=pre-restore")
    {
        return Some("Application fixture report omitted password-free restore separation checks");
    }
    if !has("phase=fixture_apply status=verified address_released=true") {
        return Some("Application fixture report omitted verified fixture apply");
    }
    if !has("phase=end status=ok") {
        return Some("Application fixture report omitted protocol termination");
    }
    None
}

fn exec_smoke_command<const N: usize>(
    container: &OwnedContainer,
    arguments: &[&str; N],
    category: &'static str,
) -> Result<(), &'static str> {
    exec_bounded(
        container,
        arguments,
        COMMAND_STDOUT_LIMIT,
        COMMAND_STDERR_LIMIT,
        category,
    )
    .map(|_| ())
}

/// Cleans exact resources only after their name and per-worker owner label match.
/// The caller must stop and reap the worker first, preventing later creation.
pub fn cleanup_confirmed_probe_resources(
    selection: &Selection,
    names: &ProbeNames,
    owner: &str,
    budget: Duration,
) -> Vec<&'static str> {
    let deadline = Instant::now() + budget;
    let mut failures = Vec::new();

    for (kind, name) in [
        (ResourceKind::Container, names.client.as_str()),
        (ResourceKind::Container, names.baseline.as_str()),
        (ResourceKind::Container, names.server.as_str()),
        (ResourceKind::Network, names.network.as_str()),
    ] {
        if let Err(category) = remove_if_confirmed_owned(selection, kind, name, owner, deadline) {
            failures.push(category);
        }
    }
    failures
}

#[derive(Clone, Copy)]
enum ResourceKind {
    Container,
    Network,
}

fn remove_if_confirmed_owned(
    selection: &Selection,
    kind: ResourceKind,
    name: &str,
    owner: &str,
    deadline: Instant,
) -> Result<(), &'static str> {
    let timeout = deadline
        .checked_duration_since(Instant::now())
        .filter(|remaining| !remaining.is_zero())
        .ok_or("cleanup budget expired")?;
    let endpoint = format!("--host={}", selection.endpoint);
    let (resource, format) = match kind {
        ResourceKind::Container => (
            "container",
            "--format={{.Id}} {{index .Config.Labels \"wireguard-admin.m8_3_owner\"}}",
        ),
        ResourceKind::Network => (
            "network",
            "--format={{.Id}} {{index .Labels \"wireguard-admin.m8_3_owner\"}}",
        ),
    };
    let output = match docker_text_with_timeout(
        &[&endpoint, resource, "inspect", format, name],
        timeout.min(Duration::from_secs(5)),
    ) {
        Ok(output) => output,
        Err("Docker named resource was not found") => return Ok(()),
        Err(_) => return Err("owned resource status could not be confirmed"),
    };
    let id = owned_resource_id(&output, owner)
        .ok_or("resource name exists without the expected ownership label")?;

    let timeout = deadline
        .checked_duration_since(Instant::now())
        .filter(|remaining| !remaining.is_zero())
        .ok_or("cleanup budget expired")?;
    let removal = match kind {
        ResourceKind::Container => docker_text_with_timeout(
            &[&endpoint, "container", "rm", "--force", &id],
            timeout.min(Duration::from_secs(5)),
        ),
        ResourceKind::Network => docker_text_with_timeout(
            &[&endpoint, "network", "rm", &id],
            timeout.min(Duration::from_secs(5)),
        ),
    };
    removal
        .map(|_| ())
        .map_err(|_| "confirmed owned resource could not be removed")
}

fn owned_resource_id(output: &str, expected_owner: &str) -> Option<String> {
    let (id, actual_owner) = output.trim().split_once(' ')?;
    if actual_owner != expected_owner
        || id.len() < 12
        || id.len() > 64
        || !id.bytes().all(|byte| byte.is_ascii_hexdigit())
    {
        return None;
    }
    Some(id.to_owned())
}

fn verify_daemon(selection: &Selection) -> Result<(), &'static str> {
    let output = docker_text(&["--host", &selection.endpoint, "info", "--format={{.ID}}"])?;
    if output.trim() != selection.daemon_id {
        return Err("selected daemon ID does not match the live daemon");
    }
    Ok(())
}

fn verify_cached_image(selection: &Selection) -> Result<(), &'static str> {
    let endpoint = format!("--host={}", selection.endpoint);
    let image = "--format={{.Id}} {{.Os}}/{{.Architecture}}";
    let output = docker_text(&[&endpoint, "image", "inspect", image, IMAGE])?;
    if output.trim() != format!("{IMAGE_ID} linux/amd64") {
        return Err("cached Alpine fixture image identity does not match the pin");
    }
    Ok(())
}

fn start_fixture_container(
    selection: &Selection,
    name: &str,
    network: &str,
) -> Result<OwnedContainer, &'static str> {
    let owner = std::process::id().to_string();
    let container = fixture_container_request(name, network, owner.clone()).start();
    let container = match container {
        Ok(container) => container,
        Err(error) => {
            let category = testcontainers_start_failure_category(&error);
            match partial_start_status(find_owned_container(selection, name, &owner)) {
                PartialStartStatus::ConfirmedOwned(id) => {
                    let endpoint = format!("--host={}", selection.endpoint);
                    if docker_text(&[&endpoint, "container", "rm", "--force", &id]).is_err() {
                        eprintln!(
                            "Testcontainers cleanup incomplete; owned container may remain: {name}"
                        );
                    }
                }
                PartialStartStatus::Uncertain => eprintln!(
                    "Testcontainers start failed; resource status remains uncertain after unconfirmed exact-name inspection: {name}"
                ),
            }
            return Err(category);
        }
    };

    Ok(OwnedContainer {
        name: name.to_owned(),
        selection: selection.clone(),
        container: Some(container),
    })
}

fn fixture_container_request(
    name: &str,
    network: &str,
    owner: String,
) -> testcontainers::ContainerRequest<GenericImage> {
    GenericImage::new("wireguard-admin-alpine", "3.22.7")
        .with_entrypoint("/bin/sh")
        .with_cmd(["-c", "while :; do sleep 1; done"])
        .with_container_name(name)
        .with_network(network)
        .with_cap_add("NET_ADMIN")
        .with_label("wireguard-admin.m8_3_owner", owner)
        .with_startup_timeout(CONTAINER_STARTUP_TIMEOUT)
}

enum PartialStartStatus {
    ConfirmedOwned(String),
    Uncertain,
}

fn partial_start_status(inspection: Result<Option<String>, &'static str>) -> PartialStartStatus {
    match inspection {
        Ok(Some(id)) => PartialStartStatus::ConfirmedOwned(id),
        Ok(None) | Err(_) => PartialStartStatus::Uncertain,
    }
}

fn testcontainers_start_failure_category(error: &TestcontainersError) -> &'static str {
    match error {
        TestcontainersError::Client(ClientError::Init(_)) => {
            "Testcontainers Docker client initialization failed"
        }
        TestcontainersError::Client(ClientError::CreateContainer(_)) => {
            "Testcontainers container creation failed"
        }
        TestcontainersError::Client(ClientError::StartContainer(_)) => {
            "Testcontainers container start failed"
        }
        TestcontainersError::WaitContainer(WaitContainerError::StartupTimeout) => {
            "Testcontainers container readiness timed out"
        }
        TestcontainersError::WaitContainer(_) => "Testcontainers container readiness failed",
        TestcontainersError::Client(ClientError::Configuration(_)) => {
            "Testcontainers Docker client configuration failed"
        }
        TestcontainersError::Client(ClientError::InvalidDockerHost(_)) => {
            "Testcontainers Docker host configuration failed"
        }
        TestcontainersError::Client(ClientError::PullImage { .. }) => {
            "Testcontainers container creation not-found image fallback failed"
        }
        TestcontainersError::Client(ClientError::BuildImage { .. }) => {
            "Testcontainers fixture image build failed"
        }
        TestcontainersError::Client(ClientError::PortMapping(_)) => {
            "Testcontainers container port mapping failed"
        }
        TestcontainersError::Client(ClientError::ListContainers(_)) => {
            "Testcontainers container listing failed"
        }
        TestcontainersError::Client(ClientError::RemoveContainer(_)) => {
            "Testcontainers container removal failed"
        }
        TestcontainersError::Client(ClientError::StopContainer(_)) => {
            "Testcontainers container stop failed"
        }
        TestcontainersError::Client(ClientError::PauseContainer(_)) => {
            "Testcontainers container pause failed"
        }
        TestcontainersError::Client(ClientError::UnpauseContainer(_)) => {
            "Testcontainers container resume failed"
        }
        TestcontainersError::Client(ClientError::InspectContainer(_)) => {
            "Testcontainers container inspection failed"
        }
        TestcontainersError::Client(ClientError::CreateNetwork(_)) => {
            "Testcontainers network creation failed"
        }
        TestcontainersError::Client(ClientError::InspectNetwork(_)) => {
            "Testcontainers network inspection failed"
        }
        TestcontainersError::Client(ClientError::ListNetworks(_)) => {
            "Testcontainers network listing failed"
        }
        TestcontainersError::Client(ClientError::RemoveNetwork(_)) => {
            "Testcontainers network removal failed"
        }
        TestcontainersError::Client(ClientError::InitExec(_)) => {
            "Testcontainers exec initialization failed"
        }
        TestcontainersError::Client(ClientError::InspectExec(_)) => {
            "Testcontainers exec inspection failed"
        }
        TestcontainersError::Client(ClientError::UploadToContainerError(_)) => {
            "Testcontainers fixture upload failed"
        }
        TestcontainersError::Client(ClientError::CopyToContainerError(_)) => {
            "Testcontainers fixture preparation failed"
        }
        TestcontainersError::Client(ClientError::CopyFromContainerError(_)) => {
            "Testcontainers fixture download failed"
        }
        TestcontainersError::PortNotExposed { .. } => {
            "Testcontainers container configuration failed"
        }
        TestcontainersError::MissingInfo(_) => "Testcontainers container information unavailable",
        TestcontainersError::Exec(_) => "Testcontainers container execution failed",
        TestcontainersError::Io(_) => "Testcontainers host I/O failed",
        TestcontainersError::Other(_) => "Testcontainers container startup failed",
    }
}

fn find_owned_container(
    selection: &Selection,
    name: &str,
    expected_owner: &str,
) -> Result<Option<String>, &'static str> {
    let endpoint = format!("--host={}", selection.endpoint);
    let format = "--format={{.Id}} {{index .Config.Labels \"wireguard-admin.m8_3_owner\"}}";
    let output = docker_text(&[&endpoint, "container", "inspect", format, name])?;
    let Some((id, owner)) = output.trim().split_once(' ') else {
        return Ok(None);
    };
    if owner != expected_owner
        || id.len() < 12
        || id.len() > 64
        || !id.bytes().all(|byte| byte.is_ascii_hexdigit())
    {
        return Ok(None);
    }
    Ok(Some(id.to_owned()))
}

fn verify_distinct_namespaces(
    server: &OwnedContainer,
    client: &OwnedContainer,
) -> Result<(), &'static str> {
    let server_namespace = exec_small(server, ["readlink", "/proc/self/ns/net"])?;
    let client_namespace = exec_small(client, ["readlink", "/proc/self/ns/net"])?;
    if server_namespace.is_empty() || server_namespace == client_namespace {
        return Err("server and client do not have distinct network namespaces");
    }
    Ok(())
}

fn verify_wireguard_and_ipv4(container: &OwnedContainer) -> Result<(), &'static str> {
    exec_success(
        container,
        ["ip", "link", "add", "wg-m83-probe", "type", "wireguard"],
    )?;
    exec_success(
        container,
        ["ip", "-4", "addr", "add", "192.0.2.1/32", "dev", "lo"],
    )?;
    const ROUTE: &str = "198.51.100.0/24";
    exec_success(container, ["ip", "-4", "route", "add", "blackhole", ROUTE])?;
    let routes = exec_small(container, ["ip", "-4", "route", "show", "exact", ROUTE])?;
    if !ipv4_blackhole_route_present(&routes, ROUTE) {
        return Err("IPv4 route mutation was not visible in the fixture namespace");
    }
    exec_success(container, ["ip", "-4", "route", "del", "blackhole", ROUTE])?;
    let routes = exec_small(container, ["ip", "-4", "route", "show", "exact", ROUTE])?;
    if !ipv4_route_absent(&routes, ROUTE) {
        return Err("IPv4 fixture route remained after deletion");
    }
    Ok(())
}

fn ipv4_blackhole_route_present(routes: &str, prefix: &str) -> bool {
    let expected = format!("blackhole {prefix}");
    routes.lines().any(|line| line.starts_with(&expected))
}

fn ipv4_route_absent(routes: &str, prefix: &str) -> bool {
    !routes
        .lines()
        .any(|line| line.split_whitespace().any(|part| part == prefix))
}

fn exec_success<const N: usize>(
    container: &OwnedContainer,
    arguments: [&str; N],
) -> Result<(), &'static str> {
    exec_bounded(
        container,
        &arguments,
        COMMAND_STDOUT_LIMIT,
        COMMAND_STDERR_LIMIT,
        "WireGuard or IPv4 capability probe failed",
    )
    .map(|_| ())
}

fn exec_small<const N: usize>(
    container: &OwnedContainer,
    arguments: [&str; N],
) -> Result<String, &'static str> {
    let output = exec_bounded(
        container,
        &arguments,
        NAMESPACE_STDOUT_LIMIT,
        COMMAND_STDERR_LIMIT,
        "Testcontainers namespace probe returned nonzero status",
    )?;
    String::from_utf8(output)
        .map(|value| value.trim().to_owned())
        .map_err(|_| "Testcontainers namespace output was invalid")
}

fn bounded_command<'a>(arguments: &[&'a str]) -> Vec<&'a str> {
    let mut bounded = vec![
        "/bin/busybox",
        "timeout",
        "-s",
        "KILL",
        FIXTURE_COMMAND_TIMEOUT_SECONDS,
    ];
    bounded.extend_from_slice(arguments);
    bounded
}

fn exec_bounded<const N: usize>(
    container: &OwnedContainer,
    arguments: &[&str; N],
    stdout_limit: usize,
    stderr_limit: usize,
    failure_category: &'static str,
) -> Result<Vec<u8>, &'static str> {
    let endpoint = format!("--host={}", container.selection.endpoint);
    let mut command = vec![endpoint.as_str(), "exec", container.container().id()];
    command.extend(bounded_command(arguments));
    match docker_text_with_timeout_and_limits(
        &command,
        FIXTURE_TRANSPORT_TIMEOUT,
        stdout_limit,
        stderr_limit,
    ) {
        Ok(stdout) => Ok(stdout.into_bytes()),
        Err("Docker command timed out") => {
            eprintln!(
                "Docker exec timed out; command completion inside the owned container is uncertain: {}",
                container.name
            );
            Err("Fixture command timed out; completion inside the owned container is uncertain")
        }
        Err("In-container fixture command timed out") => {
            Err("In-container fixture command timed out")
        }
        Err("Docker command output exceeded its limit") => {
            Err("Fixture command output exceeded its limit")
        }
        Err("Docker command returned a nonzero status") => Err(failure_category),
        Err(_) => Err(failure_category),
    }
}

fn find_owned_network(
    selection: &Selection,
    name: &str,
    expected_owner: &str,
) -> Result<Option<String>, &'static str> {
    let endpoint = format!("--host={}", selection.endpoint);
    let format = "--format={{.Id}} {{index .Labels \"wireguard-admin.m8_3_owner\"}}";
    let output = docker_text(&[&endpoint, "network", "inspect", format, name])?;
    let Some((id, owner)) = output.trim().split_once(' ') else {
        return Ok(None);
    };
    if owner != expected_owner
        || id.len() < 12
        || id.len() > 64
        || !id.bytes().all(|byte| byte.is_ascii_hexdigit())
    {
        return Ok(None);
    }
    Ok(Some(id.to_owned()))
}

struct OwnedNetwork {
    selection: Selection,
    name: String,
    id: String,
    removed: bool,
}

struct OwnedContainer {
    name: String,
    selection: Selection,
    container: Option<testcontainers::Container<GenericImage>>,
}

impl OwnedContainer {
    fn container(&self) -> &testcontainers::Container<GenericImage> {
        self.container
            .as_ref()
            .expect("owned fixture container is present before cleanup")
    }

    fn remove(mut self) -> Result<(), &'static str> {
        if let Some(container) = self.container.take()
            && container.rm().is_err()
        {
            eprintln!(
                "Testcontainers cleanup incomplete; owned container may remain: {}",
                self.name
            );
            return Err("Testcontainers could not remove an owned fixture container");
        }
        Ok(())
    }
}

impl Drop for OwnedContainer {
    fn drop(&mut self) {
        if let Some(container) = self.container.take()
            && container.rm().is_err()
        {
            eprintln!(
                "Testcontainers cleanup incomplete; owned container may remain: {}",
                self.name
            );
        }
    }
}

impl OwnedNetwork {
    fn create(selection: &Selection, name: String) -> Result<Self, &'static str> {
        let owner = format!("wireguard-admin.m8_3_owner={}", std::process::id());
        let create_result = docker_text(&[
            "--host",
            &selection.endpoint,
            "network",
            "create",
            "--driver",
            "bridge",
            "--internal",
            "--subnet",
            NETWORK_IPV4,
            "--label",
            &owner,
            &name,
        ]);
        let id = match create_result {
            Ok(id) => id,
            Err("Docker command timed out") => {
                match find_owned_network(selection, &name, &std::process::id().to_string()) {
                    Ok(Some(id)) => id,
                    Ok(None) => {
                        eprintln!(
                            "Docker network creation timed out; exact resource status is unconfirmed: {name}"
                        );
                        return Err("Docker network creation timed out without an owned network");
                    }
                    Err(_) => {
                        eprintln!(
                            "Docker network creation timed out; exact resource status is unknown: {name}"
                        );
                        return Err("Docker network creation timed out; cleanup status is unknown");
                    }
                }
            }
            Err(error) => return Err(error),
        };
        let id = id.trim();
        if id.len() < 12 || id.len() > 64 || !id.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            return Err("Docker returned an invalid ID for the owned test network");
        }
        Ok(Self {
            selection: selection.clone(),
            name,
            id: id.to_owned(),
            removed: false,
        })
    }

    fn remove(mut self) -> Result<(), &'static str> {
        self.remove_inner()?;
        Ok(())
    }

    fn remove_inner(&mut self) -> Result<(), &'static str> {
        if self.removed {
            return Ok(());
        }
        let endpoint = format!("--host={}", self.selection.endpoint);
        docker_text(&[&endpoint, "network", "rm", &self.id])?;
        self.removed = true;
        Ok(())
    }
}

impl Drop for OwnedNetwork {
    fn drop(&mut self) {
        if !self.removed && self.remove_inner().is_err() {
            eprintln!(
                "Testcontainers cleanup incomplete; owned network may remain: {}",
                self.name
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        IMAGE, MAX_PHASE_REQUEST_BYTES, PartialStartStatus, ProbeNames, bounded_command,
        client_connection_material, docker_exchange_report, driver_phase_failure_category,
        driver_ran_exactly_one_test, endpoint_host_route, fixture_container_request,
        ipv4_blackhole_route_present, ipv4_route_absent, missing_driver_phase_category,
        owned_resource_id, parse_ping_summary, partial_start_status, peer_counters,
        peer_has_exact_ipv4_routes, phase_archive_name, require_zero_ping_replies,
        testcontainers_start_failure_category,
    };
    use std::io;
    use std::time::Duration;
    use testcontainers::{
        TestcontainersError,
        core::error::{ClientError, WaitContainerError},
    };

    #[test]
    fn successful_driver_report_requires_restore_and_address_lifecycle_evidence() {
        let report = "test container_driver::alpine_scenario_driver ... phase=fixture_bootstrap status=ok\nphase=create status=ok public_key=public\nphase=commit status=ok\nphase=delivery status=ok activation_warning=true\nphase=delete status=saved current_record=removed backup_record=retained address_reserved=true\nphase=restore status=confirmed precautionary_backup=pre-restore target_client_present=true pending_address_reserved=false\nphase=restore status=confirmed precautionary_backup=pre-restore target_client_present=false pending_address_reserved=true\nphase=fixture_apply status=verified address_released=true\nphase=end status=ok\n";
        assert_eq!(missing_driver_phase_category(report), None);
    }

    #[test]
    fn phase_archive_name_skips_started_commit_and_delete_lines() {
        let report = "phase=commit status=started\nphase=commit status=ok backup_name=commit-123.tar\nphase=delete status=started backup_name=wrong.tar\nphase=delete status=saved backup_name=delete-456.tar";

        assert_eq!(phase_archive_name(report, "commit"), Ok("commit-123.tar"));
        assert_eq!(phase_archive_name(report, "delete"), Ok("delete-456.tar"));
    }

    #[test]
    fn phase_archive_name_rejects_missing_or_malformed_completed_archive() {
        assert_eq!(
            phase_archive_name(
                "phase=commit status=started backup_name=commit-123.tar",
                "commit"
            ),
            Err("Application phase omitted its archive identity")
        );
        assert_eq!(
            phase_archive_name("phase=delete status=saved", "delete"),
            Err("Application phase omitted its archive identity")
        );
        assert_eq!(
            phase_archive_name("phase=commit status=ok backup_name=../escape.tar", "commit"),
            Err("Application phase archive identity was invalid")
        );
        assert_eq!(
            phase_archive_name(
                "phase=delete status=saved backup_name=first.tar backup_name=second.tar",
                "delete"
            ),
            Err("Application phase archive identity was invalid")
        );
    }

    #[test]
    fn driver_report_rejects_zero_test_exact_filter_mismatch() {
        assert!(!driver_ran_exactly_one_test(
            "running 0 tests\ntest result: ok. 0 passed; 0 failed"
        ));
        assert!(driver_ran_exactly_one_test(
            "running 1 test\ntest container_driver::alpine_scenario_driver ... phase=fixture_bootstrap"
        ));
    }

    #[test]
    fn host_phase_transport_rejects_oversized_request_before_starting_docker() {
        let result = docker_exchange_report(
            &["--version"],
            &vec![b'x'; MAX_PHASE_REQUEST_BYTES + 1],
            Duration::from_secs(1),
            16,
            16,
        );
        assert_eq!(result, Err("fixture phase request exceeded its limit"));
    }

    #[test]
    fn delivered_config_material_keeps_wireguard_fields_and_ipv4_only_setup() {
        let bytes = b"[Interface]\nPrivateKey = synthetic\nAddress = 10.20.0.2/32, fd20::2/128\nDNS = 10.20.0.1\n\n[Peer]\nPublicKey = synthetic-server\nAllowedIPs = 0.0.0.0/0, ::/0\nEndpoint = 192.168.253.2:51820\n";
        let (address, endpoint) =
            client_connection_material(bytes).expect("synthetic delivery should parse");
        assert_eq!(address, "10.20.0.2/32");
        assert_eq!(endpoint, "192.168.253.2:51820");
    }

    #[test]
    fn delivered_config_rejects_extra_ipv4_full_tunnel_prefixes() {
        let bytes = b"[Interface]\nPrivateKey = synthetic\nAddress = 10.20.0.2/32\n\n[Peer]\nPublicKey = server\nAllowedIPs = 0.0.0.0/0, 10.20.0.0/24, ::/0\nEndpoint = 192.168.253.2:51820\n";
        assert_eq!(
            client_connection_material(bytes),
            Err("delivered client file did not contain the expected IPv4 full-tunnel route")
        );
    }

    #[test]
    fn observed_peer_route_check_rejects_extra_ipv4_prefixes_but_ignores_ipv6() {
        assert!(peer_has_exact_ipv4_routes(
            "peer=key routes=10.20.0.10/32,fd20::10/128",
            "10.20.0.10/32"
        ));
        assert!(!peer_has_exact_ipv4_routes(
            "peer=key routes=10.20.0.10/32,10.20.0.0/24,fd20::10/128",
            "10.20.0.10/32"
        ));
    }

    #[test]
    fn delivered_config_material_rejects_missing_ipv4_full_tunnel_route() {
        let bytes = b"[Interface]\nPrivateKey = synthetic\nAddress = fd20::2/128\n\n[Peer]\nPublicKey = server\nAllowedIPs = ::/0\nEndpoint = 192.168.253.2:51820\n";
        assert_eq!(
            client_connection_material(bytes),
            Err("delivered client file did not contain the expected IPv4 full-tunnel route")
        );
    }

    #[test]
    fn server_endpoint_keeps_a_host_route_on_the_underlay() {
        assert_eq!(
            endpoint_host_route("192.168.253.4:51820"),
            Ok("192.168.253.4/32".to_owned())
        );
        assert_eq!(
            endpoint_host_route("vpn.example.test:51820"),
            Err("delivered client endpoint was not IPv4")
        );
    }

    #[test]
    fn reader_counter_parser_requires_both_numeric_directions() {
        assert_eq!(
            peer_counters("peer=public received=12 transmitted=34"),
            Some((12, 34))
        );
        assert_eq!(
            peer_counters("peer=public received=unavailable transmitted=34"),
            None
        );
        assert_eq!(
            peer_counters("peer=public received=12 transmitted=unavailable"),
            None
        );
    }

    #[test]
    fn baseline_ping_summary_parses_busybox_packet_counts() {
        assert_eq!(
            parse_ping_summary("3 packets transmitted, 2 packets received, 33% packet loss"),
            Ok((3, 2))
        );
        assert_eq!(
            parse_ping_summary("3 packets transmitted, 0 packets received, 100% packet loss"),
            Ok((3, 0))
        );
        assert_eq!(
            parse_ping_summary("packets transmitted, 0 packets received, 100% packet loss"),
            Err("Baseline ping summary was malformed")
        );
    }

    #[test]
    fn negative_reachability_requires_three_transmits_and_zero_replies() {
        assert_eq!(require_zero_ping_replies(Ok((3, 0))), Ok(()));
        assert_eq!(
            require_zero_ping_replies(Ok((3, 1))),
            Err("client still reached the tunnel during a negative reachability probe")
        );
        assert_eq!(
            require_zero_ping_replies(Ok((0, 0))),
            Err("negative reachability probe did not transmit three packets")
        );
        assert_eq!(
            require_zero_ping_replies(Err("Baseline reachability probe failed")),
            Err("Baseline reachability probe failed")
        );
        assert_eq!(
            require_zero_ping_replies(Err("In-container fixture command timed out")),
            Err("In-container fixture command timed out")
        );
    }

    #[test]
    fn failed_driver_report_maps_only_the_last_typed_phase() {
        assert_eq!(
            driver_phase_failure_category(
                "test container_driver::alpine_scenario_driver ... phase=fixture_bootstrap status=ok\nphase=create status=ok public_key=synthetic\nphase=commit status=started\n"
            ),
            "Application fixture Commit phase failed"
        );
        assert_eq!(
            driver_phase_failure_category("secret-like malformed output"),
            "Application fixture bootstrap failed"
        );
    }

    #[test]
    fn generated_resource_names_are_unique_and_scoped() {
        let first = ProbeNames::unique();
        let second = ProbeNames::unique();
        assert_ne!(first.network, second.network);
        assert!(first.network.starts_with("wga-m83-net-"));
        assert!(first.server.starts_with("wga-m83-server-"));
        assert!(first.baseline.starts_with("wga-m85-baseline-"));
        assert!(first.client.starts_with("wga-m83-client-"));
        assert_ne!(first.baseline, first.client);
    }

    #[test]
    fn fixture_request_keeps_cached_image_and_sets_owned_unique_container_name() {
        let first_name = "wga-m83-server-test-a";
        let second_name = "wga-m83-client-test-b";
        let first = fixture_container_request(first_name, "wga-m83-net-test", "owner-a".into());
        let second = fixture_container_request(second_name, "wga-m83-net-test", "owner-b".into());

        assert_eq!(first.descriptor(), IMAGE);
        assert_eq!(second.descriptor(), IMAGE);
        assert_eq!(first.container_name().as_deref(), Some(first_name));
        assert_eq!(second.container_name().as_deref(), Some(second_name));
        assert_ne!(
            first.container_name().as_deref(),
            second.container_name().as_deref()
        );
        assert_eq!(
            first
                .labels()
                .get("wireguard-admin.m8_3_owner")
                .map(String::as_str),
            Some("owner-a")
        );
        assert_eq!(
            second
                .labels()
                .get("wireguard-admin.m8_3_owner")
                .map(String::as_str),
            Some("owner-b")
        );
    }

    #[test]
    fn cleanup_requires_the_exact_run_owner_label_and_a_valid_id() {
        assert_eq!(
            owned_resource_id("0123456789abcdef owner-7", "owner-7"),
            Some("0123456789abcdef".to_owned())
        );
        assert_eq!(
            owned_resource_id("0123456789abcdef owner-8", "owner-7"),
            None
        );
        assert_eq!(owned_resource_id("not-an-id owner-7", "owner-7"), None);
    }

    #[test]
    fn absent_or_uninspectable_partial_start_remains_uncertain() {
        assert!(matches!(
            partial_start_status(Ok(None)),
            PartialStartStatus::Uncertain
        ));
        assert!(matches!(
            partial_start_status(Err("Docker named resource was not found")),
            PartialStartStatus::Uncertain
        ));
        assert!(matches!(
            partial_start_status(Ok(Some("0123456789abcdef".to_owned()))),
            PartialStartStatus::ConfirmedOwned(_)
        ));
    }

    #[test]
    fn fixture_commands_use_busybox_timeout_and_escalate_to_kill() {
        assert_eq!(
            bounded_command(&["ip", "-4", "route", "show"]),
            [
                "/bin/busybox",
                "timeout",
                "-s",
                "KILL",
                "15",
                "ip",
                "-4",
                "route",
                "show"
            ]
        );
    }

    #[test]
    fn testcontainers_start_failures_map_to_fixed_redacted_categories() {
        let client_init = TestcontainersError::Client(ClientError::Init(
            io::Error::other("secret docker client detail").into(),
        ));
        let create = TestcontainersError::Client(ClientError::CreateContainer(
            io::Error::other("secret create detail").into(),
        ));
        let start = TestcontainersError::Client(ClientError::StartContainer(
            io::Error::other("secret start detail").into(),
        ));
        let timeout = TestcontainersError::WaitContainer(WaitContainerError::StartupTimeout);
        let readiness = TestcontainersError::WaitContainer(WaitContainerError::StateUnavailable);
        let invalid_host = TestcontainersError::Client(ClientError::InvalidDockerHost(
            "tcp://secret.example.invalid".to_owned(),
        ));
        let host_io = TestcontainersError::Io(io::Error::other("private socket path"));
        let other = TestcontainersError::other("secret-bearing daemon response");
        let image_fallback = TestcontainersError::Client(ClientError::PullImage {
            descriptor: "wireguard-admin-alpine:3.22.7".to_owned(),
            err: io::Error::other("synthetic create-not-found fallback detail").into(),
        });

        assert_eq!(
            testcontainers_start_failure_category(&client_init),
            "Testcontainers Docker client initialization failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&create),
            "Testcontainers container creation failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&start),
            "Testcontainers container start failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&timeout),
            "Testcontainers container readiness timed out"
        );
        assert_eq!(
            testcontainers_start_failure_category(&readiness),
            "Testcontainers container readiness failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&invalid_host),
            "Testcontainers Docker host configuration failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&host_io),
            "Testcontainers host I/O failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&other),
            "Testcontainers container startup failed"
        );
        assert_eq!(
            testcontainers_start_failure_category(&image_fallback),
            "Testcontainers container creation not-found image fallback failed"
        );
        let categories = [
            testcontainers_start_failure_category(&client_init),
            testcontainers_start_failure_category(&create),
            testcontainers_start_failure_category(&start),
            testcontainers_start_failure_category(&timeout),
            testcontainers_start_failure_category(&readiness),
            testcontainers_start_failure_category(&invalid_host),
            testcontainers_start_failure_category(&host_io),
            testcontainers_start_failure_category(&other),
            testcontainers_start_failure_category(&image_fallback),
        ];
        assert!(
            categories
                .iter()
                .all(|category| !category.contains("secret"))
        );
        assert!(
            categories
                .iter()
                .all(|category| !category.contains("private"))
        );
    }

    #[test]
    fn route_probe_mutates_then_observes_and_removes_ipv4_route() {
        assert!(ipv4_blackhole_route_present(
            "blackhole 198.51.100.0/24 proto static\n",
            "198.51.100.0/24"
        ));
        assert!(!ipv4_blackhole_route_present(
            "192.0.2.0/24 dev lo\n",
            "198.51.100.0/24"
        ));
        assert!(ipv4_route_absent(
            "192.0.2.0/24 dev lo\n",
            "198.51.100.0/24"
        ));
        assert!(!ipv4_route_absent(
            "blackhole 198.51.100.0/24 proto static\n",
            "198.51.100.0/24"
        ));
    }
}