//! New IO trait implementations for `NativeSysOps`.
//!
//! These implement the generated `IoClass*` and `IoNamespace*` traits from
//! `sys_types::io`. They coexist with the legacy `SysOp*` trait impls in
//! `lib.rs` during the transition.

use std::{
    any::TypeId,
    sync::{Arc, OnceLock},
};

use bex_heap::{BexExternalValue, BexHeap};
use sys_ops::io::{
    self, CallId, SysOpContext, SysOpOutput, VmBamlError, VmPanic, VmRustFnError, owned,
};
#[cfg(feature = "bundle-http")]
use sys_ops::io::{ObjectType, Type};
use sys_types::{BexRustData, RustDataArc as _, VmInternalError};

const MAX_READ_CHUNK: usize = 64 * 1024;

// Process-level shared BufReader for stdin, preventing data loss when
// BufReader over-reads into its internal buffer across multiple io.input() calls.
static STDIN_READER: OnceLock<tokio::sync::Mutex<tokio::io::BufReader<tokio::io::Stdin>>> =
    OnceLock::new();

fn shared_stdin() -> &'static tokio::sync::Mutex<tokio::io::BufReader<tokio::io::Stdin>> {
    STDIN_READER
        .get_or_init(|| tokio::sync::Mutex::new(tokio::io::BufReader::new(tokio::io::stdin())))
}

use crate::{NativeSysOps, WorkingDir};

// Runtime compilation is intercepted by BexEngine and delegated to the
// compiler trait injected by bex_project. This provider implementation is the
// closed fallback required by the generated IO trait hierarchy.
impl io::IoClassReflectPackage for NativeSysOps {
    fn _compile(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _files: indexmap::IndexMap<String, String>,
        _packages: indexmap::IndexMap<String, io::owned::reflect::Package>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<io::owned::reflect::CompileArtifact> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "runtime-compiler".to_string(),
            message: "runtime compiler is not installed".to_string(),
        })
    }
}

impl io::IoClassReflectSession for NativeSysOps {
    fn _compile(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _session: io::owned::reflect::Session,
        _source: String,
        _type_arg_0: ::sys_types::SapTy,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<io::owned::reflect::CompileArtifact> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "runtime-compiler".to_string(),
            message: "runtime compiler is not installed".to_string(),
        })
    }
}

impl io::IoNamespaceReflect for NativeSysOps {}

// ============================================================================
// Environment
// ============================================================================

impl io::IoNamespaceEnv for NativeSysOps {
    fn get(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        key: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        match baml_env::os_var(&key).map(std::ffi::OsString::into_string) {
            Some(Ok(val)) => SysOpOutput::ok(Some(val)),
            None => SysOpOutput::ok(None),
            Some(Err(_)) => SysOpOutput::err(VmBamlError::ParseError {
                message: format!("Environment variable '{key}' is not valid UTF-8"),
            }),
        }
    }
}

// ============================================================================
// Time
// ============================================================================

impl io::IoClassTimeInstant for NativeSysOps {
    fn now(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::time::Instant> {
        // Wall-clock time as nanoseconds since the UNIX epoch. Per the
        // `Instant.now()` contract, an unavailable/pre-epoch clock panics.
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_else(|_| unreachable!("system clock is set before the UNIX epoch"))
            .as_nanos();
        SysOpOutput::ok(owned::time::Instant {
            _nanoseconds: Arc::new(num_bigint::BigInt::from(nanos)),
        })
    }
}

/// Converts an absolute time (nanoseconds since the Unix epoch, as an
/// arbitrary-precision integer) into a `jiff::Timestamp`, saturating at
/// jiff's representable range. Saturation is fine for offset queries: a
/// timezone's offset is constant beyond the last tzdb transition, so the
/// boundary instant resolves to the same offset as any farther one.
fn saturating_timestamp(ns: &num_bigint::BigInt) -> jiff::Timestamp {
    use num_bigint::Sign;
    match i128::try_from(ns) {
        Ok(ns) => jiff::Timestamp::from_nanosecond(ns).unwrap_or(if ns < 0 {
            jiff::Timestamp::MIN
        } else {
            jiff::Timestamp::MAX
        }),
        Err(_) if ns.sign() == Sign::Minus => jiff::Timestamp::MIN,
        Err(_) => jiff::Timestamp::MAX,
    }
}

impl io::IoNamespaceTime for NativeSysOps {
    fn system_timezone(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        // The host's timezone database / system configuration, per BEP-021
        // (Deno-style: no tzdata is bundled on platforms that provide one).
        match jiff::tz::TimeZone::system().iana_name() {
            Some(name) => SysOpOutput::ok(name.to_string()),
            None => SysOpOutput::err(VmBamlError::Io {
                message: "could not determine the system's IANA timezone".to_string(),
            }),
        }
    }

    fn _tz_offset_at(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        timezone: String,
        at_ns: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<i64>> {
        // Unknown identifier → None (the BAML layer turns this into
        // `UnknownTimezoneError`).
        let Ok(tz) = jiff::tz::TimeZone::get(&timezone) else {
            return SysOpOutput::ok(None);
        };
        let offset = tz.to_offset(saturating_timestamp(&at_ns));
        SysOpOutput::ok(Some(i64::from(offset.seconds()) * 1_000_000_000))
    }

    fn _tz_to_instant(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        timezone: String,
        civil_ns: Arc<num_bigint::BigInt>,
        disambiguation: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<Arc<num_bigint::BigInt>>> {
        let Ok(tz) = jiff::tz::TimeZone::get(&timezone) else {
            return SysOpOutput::ok(None);
        };
        // The civil reading is encoded as nanoseconds since
        // 1970-01-01T00:00:00 as if it were UTC; decode it through UTC.
        let civil = match i128::try_from(&*civil_ns)
            .ok()
            .and_then(|ns| jiff::Timestamp::from_nanosecond(ns).ok())
        {
            Some(ts) => ts.to_zoned(jiff::tz::TimeZone::UTC).datetime(),
            None => {
                return SysOpOutput::err(VmBamlError::Io {
                    message: "civil time is outside the supported year range".to_string(),
                });
            }
        };
        let ambiguous = tz.to_ambiguous_timestamp(civil);
        // TC39 disambiguation semantics, via jiff. "reject" is implemented in
        // the BAML layer (earlier+later agreement), so it never reaches here.
        let resolved = match disambiguation.as_str() {
            "earlier" => ambiguous.earlier(),
            "later" => ambiguous.later(),
            _ => ambiguous.compatible(),
        };
        match resolved {
            Ok(ts) => SysOpOutput::ok(Some(Arc::new(num_bigint::BigInt::from(ts.as_nanosecond())))),
            Err(e) => SysOpOutput::err(VmBamlError::Io {
                message: format!("cannot resolve civil time in timezone '{timezone}': {e}"),
            }),
        }
    }
}

// ============================================================================
// Random (operating-system entropy)
// ============================================================================

impl io::IoClassRandomSystemRandom for NativeSysOps {
    fn random(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        bytes: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Vec<u8>> {
        let Ok(n) = usize::try_from(bytes) else {
            return SysOpOutput::err(VmPanic::UserPanic {
                message: format!("Rng.random: byte count must be non-negative, got {bytes}"),
            });
        };
        // Allocate fallibly so an unsatisfiable request is a catchable
        // `AllocFailure` panic, not a host-process abort.
        let mut buf = Vec::new();
        if buf.try_reserve(n).is_err() {
            return SysOpOutput::err(VmPanic::AllocFailure {
                message: format!("Rng.random: allocation of {n} bytes failed"),
            });
        }
        buf.resize(n, 0u8);
        match getrandom::getrandom(&mut buf) {
            Ok(()) => SysOpOutput::ok(buf),
            Err(e) => SysOpOutput::err(VmPanic::HostUnavailable {
                resource: "randomness".to_string(),
                message: format!("SystemRandom.random: system entropy unavailable: {e}"),
            }),
        }
    }

    fn random_int(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        let mut buf = [0u8; 8];
        match getrandom::getrandom(&mut buf) {
            // Arithmetic shift right by one maps the uniform 64-bit draw onto
            // the BAML i63 range `[INT_MIN, INT_MAX]`.
            Ok(()) => SysOpOutput::ok(i64::from_le_bytes(buf) >> 1),
            Err(e) => SysOpOutput::err(VmPanic::HostUnavailable {
                resource: "randomness".to_string(),
                message: format!("SystemRandom.random_int: system entropy unavailable: {e}"),
            }),
        }
    }
}

impl io::IoNamespaceRandom for NativeSysOps {}

// ============================================================================
// IO (stdin input)
// ============================================================================

/// A write to stdout/stderr failed.
///
/// `baml.io.print`/`println`/`eprint`/`eprintln` declare `throws never`, so a
/// dead output stream is not something they can report as an error value.
/// Following Rust — whose `println!` panics on write failure — a broken pipe
/// or closed console panics with `baml.panics.HostUnavailable` naming the
/// stream. `baml.io.input` is the deliberate exception: reading input is an
/// operation callers want to handle, so it declares `throws root.errors.Io`.
fn output_stream_unavailable(stream: &'static str, context: &str, e: &std::io::Error) -> VmPanic {
    VmPanic::HostUnavailable {
        resource: stream.to_string(),
        message: format!("{context}: {e}"),
    }
}

impl io::IoNamespaceIo for NativeSysOps {
    fn input(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        prompt: Option<String>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        SysOpOutput::async_op(async move {
            use tokio::io::{AsyncBufReadExt, AsyncWriteExt};

            let mut stdin = shared_stdin().lock().await;

            if let Some(p) = prompt {
                let mut stdout = tokio::io::stdout();
                stdout
                    .write_all(p.as_bytes())
                    .await
                    .map_err(|e| VmBamlError::Io {
                        message: format!("Failed to write prompt: {e}"),
                    })?;
                stdout.flush().await.map_err(|e| VmBamlError::Io {
                    message: format!("Failed to flush stdout: {e}"),
                })?;
            }
            let mut line = String::new();
            stdin
                .read_line(&mut line)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to read stdin: {e}"),
                })?;
            // Trim trailing newline (returns "" on EOF when read_line returns 0 bytes)
            if line.ends_with('\n') {
                line.pop();
            }
            if line.ends_with('\r') {
                line.pop();
            }
            Ok(line)
        })
    }

    fn print(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        s: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncWriteExt;
            let mut stdout = tokio::io::stdout();
            stdout
                .write_all(s.as_bytes())
                .await
                .map_err(|e| output_stream_unavailable("stdout", "failed to write stdout", &e))?;
            stdout
                .flush()
                .await
                .map_err(|e| output_stream_unavailable("stdout", "failed to flush stdout", &e))?;
            Ok(())
        })
    }

    fn println(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        s: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncWriteExt;
            let mut stdout = tokio::io::stdout();
            // Single write_all so concurrent calls from spawned threads don't
            // interleave on a line boundary.
            let mut buf = s.into_bytes();
            buf.push(b'\n');
            stdout
                .write_all(&buf)
                .await
                .map_err(|e| output_stream_unavailable("stdout", "failed to write stdout", &e))?;
            stdout
                .flush()
                .await
                .map_err(|e| output_stream_unavailable("stdout", "failed to flush stdout", &e))?;
            Ok(())
        })
    }

    fn eprint(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        s: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncWriteExt;
            let mut stderr = tokio::io::stderr();
            stderr
                .write_all(s.as_bytes())
                .await
                .map_err(|e| output_stream_unavailable("stderr", "failed to write stderr", &e))?;
            stderr
                .flush()
                .await
                .map_err(|e| output_stream_unavailable("stderr", "failed to flush stderr", &e))?;
            Ok(())
        })
    }

    fn eprintln(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        s: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncWriteExt;
            let mut stderr = tokio::io::stderr();
            let mut buf = s.into_bytes();
            buf.push(b'\n');
            stderr
                .write_all(&buf)
                .await
                .map_err(|e| output_stream_unavailable("stderr", "failed to write stderr", &e))?;
            stderr
                .flush()
                .await
                .map_err(|e| output_stream_unavailable("stderr", "failed to flush stderr", &e))?;
            Ok(())
        })
    }
}

// ============================================================================
// File System
// ============================================================================

/// An open file as `$rust_type` data.
struct FsFileHandle(tokio::sync::Mutex<Option<tokio::fs::File>>);

impl BexRustData for FsFileHandle {
    fn measure(&self, _: &mut sys_types::Meter) {}
}

impl std::ops::Deref for FsFileHandle {
    type Target = tokio::sync::Mutex<Option<tokio::fs::File>>;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

fn downcast_handle(file: &owned::fs::File) -> Result<Arc<FsFileHandle>, VmInternalError> {
    file._handle
        .clone()
        .downcast_payload::<FsFileHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<FsFileHandle>(),
            got: file._handle.type_id(),
        })
}

fn closed_err() -> VmBamlError {
    VmBamlError::Io {
        message: "File is closed".into(),
    }
}

impl io::IoClassFsFile for NativeSysOps {
    fn read(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        file: owned::fs::File,
        limit: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<Vec<u8>>> {
        let limit = match limit {
            ..=0 => return SysOpOutput::Ready(Ok(Some(Vec::new()))),
            1.. => usize::try_from(limit)
                .unwrap_or(MAX_READ_CHUNK)
                .min(MAX_READ_CHUNK),
        };
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncReadExt;

            let mut buf = Vec::new();
            buf.try_reserve(limit).map_err(|e| VmPanic::AllocFailure {
                message: e.to_string(),
            })?;
            buf.resize(limit, 0);

            let handle = downcast_handle(&file)?;
            let mut guard = handle.lock().await;
            let f = guard.as_mut().ok_or_else(closed_err)?;
            let len = f.read(&mut buf).await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to read file: {e}"),
            })?;
            match len {
                0 => Ok(None),
                // `buf` was sized to `limit` up front; a short read leaves the
                // tail as the zero fill, so it has to be cut back to what was
                // actually read.
                _ => {
                    buf.truncate(len);
                    Ok(Some(buf))
                }
            }
        })
    }

    fn bytes(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        file: owned::fs::File,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Vec<u8>> {
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncReadExt;

            let handle = downcast_handle(&file)?;
            let mut guard = handle.lock().await;
            let f = guard.as_mut().ok_or_else(closed_err)?;
            let mut buf = Vec::new();
            f.read_to_end(&mut buf).await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to read file: {e}"),
            })?;
            Ok(buf)
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        file: owned::fs::File,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            let handle = downcast_handle(&file)?;
            handle.lock().await.take();
            Ok(())
        })
    }

    fn seek_from(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        file: owned::fs::File,
        whence: BexExternalValue,
        offset: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        use tokio::io::AsyncSeekExt;

        SysOpOutput::async_op(async move {
            let BexExternalValue::String(whence) = whence else {
                return Err(VmRustFnError::from(VmBamlError::InvalidArgument {
                    message: "Invalid whence type".into(),
                }));
            };
            let from = match whence.as_str() {
                "start" => {
                    let off = u64::try_from(offset).map_err(|_| VmBamlError::InvalidArgument {
                        message: format!("Negative offset with whence=\"start\": {offset}"),
                    })?;
                    std::io::SeekFrom::Start(off)
                }
                "current" => std::io::SeekFrom::Current(offset),
                "end" => std::io::SeekFrom::End(offset),
                _ => {
                    return Err(VmRustFnError::from(VmBamlError::InvalidArgument {
                        message: format!(
                            "Unsupported whence '{whence}': expected \"start\", \"current\", or \"end\""
                        ),
                    }));
                }
            };
            let handle = downcast_handle(&file)?;
            let mut guard = handle.lock().await;
            let f = guard.as_mut().ok_or_else(closed_err)?;
            let pos = f.seek(from).await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to seek: {e}"),
            })?;
            i64::try_from(pos)
                .map_err(|_| VmBamlError::Io {
                    message: format!("Seek position out of range: {pos}"),
                })
                .map_err(VmRustFnError::from)
        })
    }

    fn write_some(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        file: owned::fs::File,
        data: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncWriteExt;

            let handle = downcast_handle(&file)?;
            let mut guard = handle.lock().await;
            let f = guard.as_mut().ok_or_else(closed_err)?;
            let len = f.write(&data).await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to write: {e}"),
            })?;
            // `tokio::fs::File` buffers: `write` copies into its buffer and
            // dispatches the real write to the blocking pool, so a rejection
            // (a read-only handle, a full disk) would otherwise surface on a
            // *later* call or be lost entirely. Settling it here keeps
            // `write_some` an unbuffered primitive, matching this interface's
            // other implementors and every other language's raw file handle.
            f.flush().await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to write: {e}"),
            })?;
            let len = i64::try_from(len)
                .unwrap_or_else(|_| unreachable!("There are no systems with this much memory"));
            Ok(len)
        })
    }

    fn flush(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        file: owned::fs::File,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        use tokio::io::AsyncWriteExt;

        SysOpOutput::async_op(async move {
            let handle = downcast_handle(&file)?;
            let mut guard = handle.lock().await;
            let f = guard.as_mut().ok_or_else(closed_err)?;
            f.flush().await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to flush: {e}"),
            })?;
            Ok(())
        })
    }
}

impl io::IoNamespaceFs for NativeSysOps {
    fn metadata(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::fs::Metadata> {
        crate::installation::fs_metadata(self, &path)
    }
    fn rename(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        source: String,
        destination: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        crate::installation::fs_rename(self, &source, &destination)
    }
    fn canonicalize(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        crate::installation::fs_canonicalize(self, &path)
    }
    fn write_atomic(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        content: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        crate::installation::fs_write_atomic(self, &path, content)
    }
    fn lock(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        timeout_ms: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::fs::Lock> {
        crate::installation::fs_lock(self, &path, timeout_ms)
    }

    fn open(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        mode: BexExternalValue,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::fs::File> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            let BexExternalValue::String(mode) = mode else {
                return Err(VmRustFnError::from(VmBamlError::InvalidArgument {
                    message: "Invalid mode type".into(),
                }));
            };
            // Modes that create the file also auto-create missing parent dirs,
            // matching Bun's `Bun.write` behavior.
            let creates = matches!(mode.as_str(), "w" | "w+" | "a" | "a+");
            if creates {
                if let Some(parent) = path.parent() {
                    if !parent.as_os_str().is_empty() {
                        tokio::fs::create_dir_all(parent)
                            .await
                            .map_err(|e| VmBamlError::Io {
                                message: format!(
                                    "Failed to create parent directories for '{}': {e}",
                                    path.display()
                                ),
                            })?;
                    }
                }
            }
            let file = match mode.as_str() {
                "r" => tokio::fs::File::open(&path).await,
                "r+" => {
                    tokio::fs::OpenOptions::new()
                        .read(true)
                        .write(true)
                        .open(&path)
                        .await
                }
                "w" => {
                    tokio::fs::OpenOptions::new()
                        .write(true)
                        .create(true)
                        .truncate(true)
                        .open(&path)
                        .await
                }
                "w+" => {
                    tokio::fs::OpenOptions::new()
                        .read(true)
                        .write(true)
                        .create(true)
                        .truncate(true)
                        .open(&path)
                        .await
                }
                "a" => {
                    tokio::fs::OpenOptions::new()
                        .append(true)
                        .create(true)
                        .open(&path)
                        .await
                }
                "a+" => {
                    tokio::fs::OpenOptions::new()
                        .read(true)
                        .append(true)
                        .create(true)
                        .open(&path)
                        .await
                }
                _ => {
                    return Err(VmRustFnError::from(VmBamlError::InvalidArgument {
                        message: format!(
                            "Unsupported file mode '{mode}': expected \"r\", \"r+\", \"w\", \"w+\", \"a\", or \"a+\""
                        ),
                    }));
                }
            }
            .map_err(|e| VmBamlError::Io {
                message: format!("Failed to open file '{}': {e}", path.display()),
            })?;
            let handle: Arc<dyn BexRustData> =
                Arc::new(FsFileHandle(tokio::sync::Mutex::new(Some(file))));
            Ok(owned::fs::File { _handle: handle })
        })
    }

    fn exists(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<bool> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            tokio::fs::try_exists(&path)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to check existence of '{}': {e}", path.display()),
                })
                .map_err(VmRustFnError::from)
        })
    }

    fn remove(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            match tokio::fs::remove_file(&path).await {
                Ok(()) => Ok(()),
                Err(e) => {
                    // `remove_file` rejects directories, but the raw OS error is
                    // opaque and platform-specific (EISDIR / "Is a directory" on
                    // Linux, EPERM / "Operation not permitted" on macOS). When
                    // the target is actually a directory, surface a message that
                    // points at the directory-removal API instead. `symlink_metadata`
                    // does not follow the final component, so a symlink (which
                    // `remove_file` deletes fine) is never misreported here.
                    if tokio::fs::symlink_metadata(&path)
                        .await
                        .is_ok_and(|m| m.is_dir())
                    {
                        return Err(VmBamlError::Io {
                            message: format!(
                                "Failed to remove '{}': it is a directory; use baml.fs.remove_dir or baml.fs.remove_dir_all to delete directories", path.display()
                            ),
                        }
                        .into());
                    }
                    Err(VmBamlError::Io {
                        message: format!("Failed to remove file '{}': {e}", path.display()),
                    }
                    .into())
                }
            }
        })
    }

    fn remove_dir(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            tokio::fs::remove_dir(&path)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to remove directory '{}': {e}", path.display()),
                })
                .map_err(VmRustFnError::from)
        })
    }

    fn remove_dir_all(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            match tokio::fs::remove_dir_all(&path).await {
                Ok(()) => Ok(()),
                // `force: true` semantics: a missing path is not an error.
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
                Err(e) => Err(VmBamlError::Io {
                    message: format!("Failed to remove directory '{}': {e}", path.display()),
                }
                .into()),
            }
        })
    }

    fn size(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            let metadata = tokio::fs::metadata(&path)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to stat '{}': {e}", path.display()),
                })?;
            i64::try_from(metadata.len())
                .map_err(|_| VmBamlError::Io {
                    message: format!("File '{}' size exceeds i64::MAX", path.display()),
                })
                .map_err(VmRustFnError::from)
        })
    }

    fn read(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            tokio::fs::read_to_string(&path)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to read file '{}': {e}", path.display()),
                })
                .map_err(VmRustFnError::from)
        })
    }

    fn write(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        content: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            write_path(&path, content.as_bytes())
                .await
                .map_err(VmRustFnError::from)
        })
    }

    fn write_bytes(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        content: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            write_path(&path, &content)
                .await
                .map_err(VmRustFnError::from)
        })
    }

    fn read_dir(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Vec<owned::fs::DirEntry>> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            let mut rd = tokio::fs::read_dir(&path)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to read directory '{}': {e}", path.display()),
                })?;
            let mut entries = Vec::new();
            while let Some(entry) = rd.next_entry().await.map_err(|e| VmBamlError::Io {
                message: format!(
                    "Failed to read directory entry in '{}': {e}",
                    path.display()
                ),
            })? {
                let ft = entry.file_type().await.map_err(|e| VmBamlError::Io {
                    message: format!(
                        "Failed to get file type for '{}': {e}",
                        entry.file_name().to_string_lossy()
                    ),
                })?;
                entries.push(owned::fs::DirEntry {
                    name: entry.file_name().to_string_lossy().into_owned(),
                    is_dir: ft.is_dir(),
                    is_file: ft.is_file(),
                    is_symlink: ft.is_symlink(),
                });
            }
            Ok(entries)
        })
    }

    fn mkdir(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        options: owned::fs::MkdirOptions,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            if options.recursive {
                tokio::fs::create_dir_all(&path).await
            } else {
                tokio::fs::create_dir(&path).await
            }
            .map_err(|e| VmBamlError::Io {
                message: format!("Failed to create directory '{}': {e}", path.display()),
            })
            .map_err(VmRustFnError::from)
        })
    }

    fn chmod(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        path: String,
        mode: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        let mode = match permission_bits(mode) {
            Ok(mode) => mode,
            Err(err) => return SysOpOutput::err(err),
        };
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move { chmod_path(&path, mode).await })
    }

    fn symlink(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        target: String,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        let path = self.working_dir.resolve(&path);
        SysOpOutput::async_op(async move {
            symlink_path(&target, &path)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!(
                        "Failed to create symlink '{}' -> '{target}': {e}",
                        path.display()
                    ),
                })
                .map_err(VmRustFnError::from)
        })
    }
}

/// Narrow a BAML `int` mode to the POSIX permission bits.
///
/// The accepted range is the permission triple plus the setuid/setgid/sticky
/// digit. Everything else — negative values, file-type bits from a `stat`
/// result, anything past `u32` — is rejected rather than masked, because a mode
/// outside that range is a caller mistake and the kernel would silently drop the
/// extra bits (`chmod_common` masks with `S_IALLUGO`).
fn permission_bits(mode: i64) -> Result<u32, VmBamlError> {
    u32::try_from(mode)
        .ok()
        .filter(|mode| *mode <= 0o7777)
        .ok_or_else(|| VmBamlError::InvalidArgument {
            // `{:#o}` renders a negative `i64` as its 64-bit two's complement,
            // which reads as a nonsensically large mode; show those in decimal.
            message: match mode {
                0.. => {
                    format!("Invalid file mode {mode:#o}: expected a permission mask in 0..=0o7777")
                }
                _ => format!("Invalid file mode {mode}: expected a permission mask in 0..=0o7777"),
            },
        })
}

#[cfg(unix)]
async fn chmod_path(path: &std::path::Path, mode: u32) -> Result<(), VmRustFnError> {
    use std::os::unix::fs::PermissionsExt as _;

    tokio::fs::set_permissions(path, std::fs::Permissions::from_mode(mode))
        .await
        .map_err(|e| VmBamlError::Io {
            message: format!("Failed to set permissions on '{}': {e}", path.display()),
        })
        .map_err(VmRustFnError::from)
}

/// Windows has no mode bits — `Permissions` carries a single read-only flag —
/// so the owner-write bit decides it and the rest of `mode` is ignored. This
/// matches libuv (and therefore Node's `fs.chmod`), which tests exactly
/// `_S_IWRITE`.
///
/// Reading the current permissions first is what makes a missing `path` fail
/// here as it does on unix, rather than silently succeeding.
#[cfg(windows)]
async fn chmod_path(path: &std::path::Path, mode: u32) -> Result<(), VmRustFnError> {
    let mut permissions = tokio::fs::metadata(path)
        .await
        .map_err(|e| VmBamlError::Io {
            message: format!("Failed to read permissions of '{}': {e}", path.display()),
        })?
        .permissions();
    permissions.set_readonly((mode & 0o200) == 0);
    tokio::fs::set_permissions(path, permissions)
        .await
        .map_err(|e| VmBamlError::Io {
            message: format!("Failed to set permissions on '{}': {e}", path.display()),
        })
        .map_err(VmRustFnError::from)
}

#[cfg(unix)]
async fn symlink_path(target: &str, path: &std::path::Path) -> std::io::Result<()> {
    tokio::fs::symlink(target, path).await
}

/// Windows picks the link flavor at creation time and cannot change it later,
/// so `target` is resolved the way the OS will resolve it — a relative target
/// hangs off the link's own directory — and a directory link is created only
/// when that resolves to a directory today. A dangling link becomes a file
/// link, matching Node's autodetect.
#[cfg(windows)]
async fn symlink_path(target: &str, path: &std::path::Path) -> std::io::Result<()> {
    let target_path = std::path::Path::new(target);
    let resolved = if target_path.is_absolute() {
        target_path.to_path_buf()
    } else {
        path.parent()
            .unwrap_or_else(|| std::path::Path::new(""))
            .join(target_path)
    };
    if tokio::fs::metadata(&resolved)
        .await
        .is_ok_and(|metadata| metadata.is_dir())
    {
        tokio::fs::symlink_dir(target, path).await
    } else {
        tokio::fs::symlink_file(target, path).await
    }
}

// Auto-creates missing parent dirs, matching Bun's `Bun.write` behavior.
async fn write_path(path: &std::path::Path, data: &[u8]) -> Result<i64, VmBamlError> {
    if let Some(parent) = path.parent() {
        if !parent.as_os_str().is_empty() {
            tokio::fs::create_dir_all(parent)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!(
                        "Failed to create parent directories for '{}': {e}",
                        path.display()
                    ),
                })?;
        }
    }
    tokio::fs::write(path, data)
        .await
        .map_err(|e| VmBamlError::Io {
            message: format!("Failed to write file '{}': {e}", path.display()),
        })?;
    i64::try_from(data.len()).map_err(|_| VmBamlError::Io {
        message: format!("Write size {} exceeds i64::MAX", data.len()),
    })
}

// ============================================================================
// Glob
// ============================================================================

use sys_glob::GlobPattern;

/// A compiled pattern as `$rust_type` data.
struct GlobHandle(GlobPattern);

impl BexRustData for GlobHandle {
    fn measure(&self, _: &mut sys_types::Meter) {
        // Bounded by the pattern text; the regex crate does not report its size.
    }
}

impl std::ops::Deref for GlobHandle {
    type Target = GlobPattern;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

fn downcast_glob_handle(glob: &owned::glob::Glob) -> Result<Arc<GlobHandle>, VmInternalError> {
    glob._handle
        .clone()
        .downcast_payload::<GlobHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<GlobHandle>(),
            got: glob._handle.type_id(),
        })
}

impl io::IoNamespaceGlob for NativeSysOps {
    fn new(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        pattern: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::glob::Glob> {
        match GlobPattern::new(&pattern) {
            Ok(gp) => {
                let handle: Arc<dyn BexRustData> = Arc::new(GlobHandle(gp));
                SysOpOutput::ok(owned::glob::Glob { _handle: handle })
            }
            Err(e) => SysOpOutput::err(VmBamlError::ParseError { message: e }),
        }
    }
}

impl io::IoClassGlobGlob for NativeSysOps {
    fn scan(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        glob: owned::glob::Glob,
        root: BexExternalValue,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Vec<String>> {
        let working_dir = self.working_dir.clone();
        SysOpOutput::async_op(async move {
            let handle = downcast_glob_handle(&glob)?;

            let (cwd, dot, absolute, follow_symlinks, throw_on_broken, only_files) = match &root {
                BexExternalValue::String(s) => (s.to_string(), false, false, false, false, true),
                BexExternalValue::Instance { fields, .. } => {
                    let get_string = |key: &str, default: &str| {
                        fields
                            .get(key)
                            .and_then(BexExternalValue::as_string)
                            .map(|s| s.to_string())
                            .unwrap_or_else(|| default.to_string())
                    };
                    let get_bool = |key: &str, default: bool| {
                        fields
                            .get(key)
                            .and_then(BexExternalValue::as_bool)
                            .unwrap_or(default)
                    };
                    let cwd = get_string("cwd", ".");
                    let dot = get_bool("dot", false);
                    let absolute = get_bool("absolute", false);
                    let follow_symlinks = get_bool("follow_symlinks", false);
                    let throw_on_broken = get_bool("throw_error_on_broken_symlink", false);
                    let only_files = get_bool("only_files", true);
                    (
                        cwd,
                        dot,
                        absolute,
                        follow_symlinks,
                        throw_on_broken,
                        only_files,
                    )
                }
                _ => {
                    // `root` is declared `string | ScanOptions`, so a value of
                    // any other shape means the wire value disagrees with the
                    // declared parameter type.
                    return Err(VmRustFnError::from(VmInternalError::BridgeFailure {
                        message: format!(
                            "glob scan root is neither a string nor a ScanOptions: {}",
                            root.type_name()
                        ),
                    }));
                }
            };

            let cwd_path = working_dir.resolve(&cwd);
            let abs_cwd = if cwd_path.is_absolute() {
                cwd_path
            } else {
                std::env::current_dir()
                    .map_err(|e| VmBamlError::Io {
                        message: format!("Failed to get cwd: {e}"),
                    })?
                    .join(cwd_path)
            };

            // Prune dot directories during the walk when `dot=false` so we
            // never descend into trees like `.git/`. The previous post-walk
            // filter still discarded those entries but only after walkdir
            // had paid the I/O to enumerate them. Depth 0 is always kept so
            // the user can scan a dot-prefixed root they explicitly point
            // at (e.g. `Glob.scan(".config")`).
            let walker = walkdir::WalkDir::new(&abs_cwd)
                .follow_links(follow_symlinks)
                .into_iter()
                .filter_entry(move |entry| {
                    dot || entry.depth() == 0
                        || !entry.file_name().to_string_lossy().starts_with('.')
                });

            let mut results = Vec::new();
            for entry in walker {
                let entry = match entry {
                    Ok(e) => e,
                    Err(e) => {
                        // Classify: broken symlink vs. other I/O error. A broken
                        // symlink surfaces as NotFound on a path whose lstat
                        // identifies it as a symlink. The `throw_on_broken`
                        // option is scoped to broken symlinks only — every
                        // other walk error (permission denied, transient I/O,
                        // symlink loop) is a real failure and propagates
                        // regardless. Silently swallowing real errors hides
                        // problems users want to know about.
                        let is_broken_symlink = e
                            .io_error()
                            .is_some_and(|io_err| io_err.kind() == std::io::ErrorKind::NotFound)
                            && e.path()
                                .and_then(|p| std::fs::symlink_metadata(p).ok())
                                .is_some_and(|m| m.file_type().is_symlink());

                        if is_broken_symlink {
                            if throw_on_broken {
                                return Err(VmRustFnError::from(VmBamlError::Io {
                                    message: format!("Broken symlink: {e}"),
                                }));
                            }
                            continue;
                        }
                        return Err(VmRustFnError::from(VmBamlError::Io {
                            message: format!("Walk error: {e}"),
                        }));
                    }
                };

                // Skip the root itself
                if entry.depth() == 0 {
                    continue;
                }

                let ft = entry.file_type();
                if only_files && !ft.is_file() {
                    continue;
                }

                let rel = entry.path().strip_prefix(&abs_cwd).unwrap_or(entry.path());
                let rel_str = rel.to_string_lossy().replace('\\', "/");
                let abs_str = entry.path().to_string_lossy().replace('\\', "/");

                // Dot filtering is handled by `filter_entry` above; no need
                // to re-check here.

                if !handle.is_match_entry(&rel_str, &abs_str) {
                    continue;
                }

                if absolute {
                    results.push(abs_str);
                } else {
                    results.push(rel_str);
                }
            }
            Ok(results)
        })
    }

    fn matches(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        glob: owned::glob::Glob,
        path: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<bool> {
        match downcast_glob_handle(&glob) {
            Ok(handle) => SysOpOutput::ok(handle.is_match(&path)),
            Err(e) => SysOpOutput::err(e),
        }
    }
}

// ============================================================================
// System
// ============================================================================

type NativeProcessResult = Result<owned::sys::ProcessExit, VmBamlError>;
type ProcessKillRequest = tokio::sync::oneshot::Sender<Result<(), VmBamlError>>;

struct LiveProcessHandle {
    kill_tx: std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedSender<ProcessKillRequest>>>,
    exit_rx: tokio::sync::watch::Receiver<Option<NativeProcessResult>>,
    label: String,
}

impl BexRustData for LiveProcessHandle {
    fn measure(&self, meter: &mut sys_types::Meter) {
        meter.bytes(self.label.capacity());
    }
}

struct ReadPipeHandle {
    reader: tokio::sync::Mutex<Option<Box<dyn tokio::io::AsyncRead + Send + Unpin>>>,
    close_tx: tokio::sync::watch::Sender<bool>,
    label: String,
}

impl BexRustData for ReadPipeHandle {
    fn measure(&self, meter: &mut sys_types::Meter) {
        meter.bytes(self.label.capacity());
    }
}

struct WritePipeHandle {
    close_tx: tokio::sync::watch::Sender<bool>,
    writer: tokio::sync::Mutex<Option<Box<dyn tokio::io::AsyncWrite + Send + Unpin>>>,
    label: String,
}

impl BexRustData for WritePipeHandle {
    fn measure(&self, meter: &mut sys_types::Meter) {
        meter.bytes(self.label.capacity());
    }
}

fn process_stdio(mode: &BexExternalValue) -> Result<std::process::Stdio, VmRustFnError> {
    let value = match mode {
        BexExternalValue::String(value) => value.as_str(),
        BexExternalValue::Union { value, .. } => return process_stdio(value),
        _ => {
            return Err(VmBamlError::InvalidArgument {
                message: "Invalid process stream mode".into(),
            }
            .into());
        }
    };
    match value {
        "inherit" => Ok(std::process::Stdio::inherit()),
        "pipe" => Ok(std::process::Stdio::piped()),
        "ignore" => Ok(std::process::Stdio::null()),
        _ => Err(VmBamlError::InvalidArgument {
            message: format!("Invalid process stream mode '{value}'"),
        }
        .into()),
    }
}

#[cfg_attr(
    unix,
    expect(
        unsafe_code,
        reason = "setsid runs in an async-signal-safe pre-exec hook"
    )
)]
fn configure_process(
    cmd: &mut tokio::process::Command,
    options: &owned::sys::ProcessOptions,
    working_dir: &WorkingDir,
) -> Result<(), VmRustFnError> {
    if let Some(dir) = working_dir.for_child(options.cwd.as_deref()) {
        cmd.current_dir(dir);
    }
    if !options.inherit_env {
        cmd.env_clear();
    }
    if let Some(env) = &options.env {
        for (name, value) in env {
            match value {
                Some(value) => {
                    cmd.env(name, value);
                }
                None => {
                    cmd.env_remove(name);
                }
            }
        }
    }
    cmd.stdin(process_stdio(&options.stdin)?)
        .stdout(process_stdio(&options.stdout)?)
        .stderr(process_stdio(&options.stderr)?)
        .kill_on_drop(false);
    if options.input.is_some() {
        cmd.stdin(std::process::Stdio::piped());
    }
    if options.detached {
        #[cfg(unix)]
        {
            use std::os::unix::process::CommandExt as _;
            // SAFETY: this hook only performs the async-signal-safe setsid syscall
            // and reads errno. It does not allocate, acquire locks or touch the VM.
            unsafe {
                cmd.as_std_mut().pre_exec(|| {
                    if libc::setsid() == -1 {
                        Err(std::io::Error::last_os_error())
                    } else {
                        Ok(())
                    }
                });
            }
        }
        #[cfg(windows)]
        cmd.creation_flags(0x0000_0008 | 0x0000_0200);
    }
    Ok(())
}

fn downcast_read_pipe(pipe: &owned::sys::ReadPipe) -> Result<Arc<ReadPipeHandle>, VmInternalError> {
    pipe._pipe
        .clone()
        .downcast_payload::<ReadPipeHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<ReadPipeHandle>(),
            got: pipe._pipe.type_id(),
        })
}

fn downcast_write_pipe(
    pipe: &owned::sys::WritePipe,
) -> Result<Arc<WritePipeHandle>, VmInternalError> {
    pipe._pipe
        .clone()
        .downcast_payload::<WritePipeHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<WritePipeHandle>(),
            got: pipe._pipe.type_id(),
        })
}

fn read_pipe(
    reader: impl tokio::io::AsyncRead + Send + Unpin + 'static,
    label: String,
) -> owned::sys::ReadPipe {
    let (close_tx, _close_rx) = tokio::sync::watch::channel(false);
    owned::sys::ReadPipe {
        _pipe: Arc::new(ReadPipeHandle {
            reader: tokio::sync::Mutex::new(Some(Box::new(reader))),
            close_tx,
            label,
        }),
    }
}

fn read_pipe_closed_error(label: &str) -> VmBamlError {
    VmBamlError::Io {
        message: format!("Read pipe for '{label}' is closed"),
    }
}

async fn close_read_pipe(handle: &ReadPipeHandle) {
    handle.close_tx.send_replace(true);
    handle.reader.lock().await.take();
}

fn write_pipe(
    writer: impl tokio::io::AsyncWrite + Send + Unpin + 'static,
    label: String,
) -> owned::sys::WritePipe {
    let (close_tx, _) = tokio::sync::watch::channel(false);
    owned::sys::WritePipe {
        _pipe: Arc::new(WritePipeHandle {
            close_tx,
            writer: tokio::sync::Mutex::new(Some(Box::new(writer))),
            label,
        }),
    }
}

fn downcast_process_handle(
    process: &owned::sys::Subprocess,
) -> Result<Arc<LiveProcessHandle>, VmInternalError> {
    process
        ._handle
        .clone()
        .downcast_payload::<LiveProcessHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<LiveProcessHandle>(),
            got: process._handle.type_id(),
        })
}

fn process_exit_from_status(status: std::process::ExitStatus) -> owned::sys::ProcessExit {
    #[cfg(unix)]
    let signal = {
        use std::os::unix::process::ExitStatusExt as _;
        status.signal()
    };
    #[cfg(not(unix))]
    let signal: Option<i32> = None;

    owned::sys::ProcessExit {
        exit_code: i64::from(status.code().unwrap_or(128 + signal.unwrap_or(0))),
        signal: signal.map(i64::from),
    }
}

fn process_input(options: &owned::sys::ProcessOptions) -> Result<Option<Vec<u8>>, VmRustFnError> {
    fn bytes(input: &BexExternalValue) -> Result<Vec<u8>, VmRustFnError> {
        match input {
            BexExternalValue::String(text) => Ok(text.as_bytes().to_vec()),
            BexExternalValue::Uint8Array(data) => Ok(data.clone()),
            BexExternalValue::Union { value, .. } => bytes(value),
            _ => Err(VmBamlError::InvalidArgument {
                message: "Process stdin must be text or bytes".into(),
            }
            .into()),
        }
    }
    options.input.as_ref().map(bytes).transpose()
}

fn process_timeout(
    duration: Option<&BexExternalValue>,
) -> Result<Option<std::time::Duration>, VmRustFnError> {
    fn nanos(value: &BexExternalValue) -> Result<u64, VmRustFnError> {
        let invalid = || VmBamlError::InvalidArgument {
            message: "Process timeout must be a nonnegative duration representable in nanoseconds"
                .into(),
        };
        match value {
            BexExternalValue::Instance { fields, .. } => match fields.get("_nanoseconds") {
                Some(BexExternalValue::Bigint(n)) => u64::try_from(n).map_err(|_| invalid().into()),
                _ => Err(invalid().into()),
            },
            BexExternalValue::Union { value, .. } => nanos(value),
            _ => Err(invalid().into()),
        }
    }
    duration
        .map(|duration| nanos(duration).map(std::time::Duration::from_nanos))
        .transpose()
}

fn process_timeout_error(label: &str, timeout: Option<std::time::Duration>) -> VmBamlError {
    VmBamlError::Timeout {
        message: format!(
            "Command '{label}' timed out after {}ms",
            timeout.unwrap_or_default().as_millis()
        ),
        duration: timeout,
        timeout_type: "timeout".to_string(),
    }
}

async fn receive_process_exit(
    mut exit_rx: tokio::sync::watch::Receiver<Option<NativeProcessResult>>,
) -> Result<owned::sys::ProcessExit, VmRustFnError> {
    loop {
        if let Some(result) = exit_rx.borrow().clone() {
            return result.map_err(Into::into);
        }
        exit_rx.changed().await.map_err(|_| VmBamlError::Io {
            message: "Process monitor stopped before reporting an exit status".into(),
        })?;
    }
}

impl io::IoClassSysReadPipe for NativeSysOps {
    fn read(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        readpipe: owned::sys::ReadPipe,
        limit: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<Vec<u8>>> {
        use tokio::io::AsyncReadExt as _;

        let limit = match limit {
            ..=0 => return SysOpOutput::Ready(Ok(Some(Vec::new()))),
            1.. => usize::try_from(limit)
                .unwrap_or(MAX_READ_CHUNK)
                .min(MAX_READ_CHUNK),
        };
        SysOpOutput::async_op(async move {
            let handle = downcast_read_pipe(&readpipe)?;
            let mut close_rx = handle.close_tx.subscribe();
            if *close_rx.borrow() {
                return Err(read_pipe_closed_error(&handle.label).into());
            }
            let mut guard = handle.reader.lock().await;
            if *close_rx.borrow() {
                return Err(read_pipe_closed_error(&handle.label).into());
            }
            let reader = guard
                .as_mut()
                .ok_or_else(|| read_pipe_closed_error(&handle.label))?;
            let mut buffer = vec![0u8; limit];
            let read = tokio::select! {
                biased;
                _ = close_rx.changed() => {
                    guard.take();
                    return Err(read_pipe_closed_error(&handle.label).into());
                }
                result = reader.read(&mut buffer) => result.map_err(|error| VmBamlError::Io {
                    message: format!("Failed to read from '{}': {error}", handle.label),
                })?,
            };
            if read == 0 {
                return Ok(None);
            }
            buffer.truncate(read);
            Ok(Some(buffer))
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        readpipe: owned::sys::ReadPipe,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            let handle = downcast_read_pipe(&readpipe)?;
            close_read_pipe(&handle).await;
            Ok(())
        })
    }
}

impl io::IoClassSysWritePipe for NativeSysOps {
    fn write_some(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        writepipe: owned::sys::WritePipe,
        data: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        use tokio::io::AsyncWriteExt as _;

        SysOpOutput::async_op(async move {
            let handle = downcast_write_pipe(&writepipe)?;
            let mut guard = handle.writer.lock().await;
            let writer = guard.as_mut().ok_or_else(|| VmBamlError::Io {
                message: format!("Write pipe for '{}' is closed", handle.label),
            })?;
            let mut closed = handle.close_tx.subscribe();
            if *closed.borrow() {
                return Err(VmBamlError::Io {
                    message: "Write pipe closed".into(),
                }
                .into());
            }
            let written = tokio::select! {
                _ = closed.changed() => { guard.take(); return Err(VmBamlError::Io { message: "Write pipe closed".into() }.into()); },
                result = writer.write(&data) => result.map_err(|error| VmBamlError::Io {
                    message: format!("Failed to write to '{}': {error}", handle.label),
                })?,
            };
            i64::try_from(written).map_err(|_| {
                VmRustFnError::from(VmBamlError::Io {
                    message: "pipe write count exceeds int range".to_string(),
                })
            })
        })
    }

    fn flush(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        writepipe: owned::sys::WritePipe,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        use tokio::io::AsyncWriteExt as _;

        SysOpOutput::async_op(async move {
            let handle = downcast_write_pipe(&writepipe)?;
            let mut guard = handle.writer.lock().await;
            let writer = guard.as_mut().ok_or_else(|| VmBamlError::Io {
                message: format!("Write pipe for '{}' is closed", handle.label),
            })?;
            let mut closed = handle.close_tx.subscribe();
            if *closed.borrow() {
                return Err(VmBamlError::Io {
                    message: "Write pipe closed".into(),
                }
                .into());
            }
            tokio::select! {
                _ = closed.changed() => { guard.take(); return Err(VmBamlError::Io { message: "Write pipe closed".into() }.into()); },
                result = writer.flush() => result.map_err(|error| VmBamlError::Io {
                    message: format!("Failed to flush '{}': {error}", handle.label),
                })?,
            };
            Ok(())
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        writepipe: owned::sys::WritePipe,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            let handle = downcast_write_pipe(&writepipe)?;
            handle.close_tx.send_replace(true);
            handle.writer.lock().await.take();
            Ok(())
        })
    }
}

impl io::IoClassSysSubprocess for NativeSysOps {
    fn wait(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        process: owned::sys::Subprocess,
        timeout: Option<BexExternalValue>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::sys::ProcessExit> {
        SysOpOutput::async_op(async move {
            let handle = downcast_process_handle(&process)?;
            let timeout = process_timeout(timeout.as_ref())?;
            if let Some(result) = handle.exit_rx.borrow().clone() {
                return result.map_err(Into::into);
            }
            if let Some(timeout) = timeout {
                tokio::time::timeout(timeout, receive_process_exit(handle.exit_rx.clone()))
                    .await
                    .map_err(|_| process_timeout_error(&handle.label, Some(timeout)))?
            } else {
                receive_process_exit(handle.exit_rx.clone()).await
            }
        })
    }

    fn kill(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        process: owned::sys::Subprocess,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            let handle = downcast_process_handle(&process)?;
            if handle.exit_rx.borrow().is_some() {
                return Ok(());
            }
            let tx = handle
                .kill_tx
                .lock()
                .expect("process control lock poisoned")
                .clone();
            if let Some(tx) = tx {
                let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
                if tx.send(ack_tx).is_ok() {
                    // Acknowledge the OS termination request, without waiting for
                    // exit. This also prevents VM shutdown from losing the request.
                    if let Ok(result) = ack_rx.await {
                        result?;
                    }
                }
                Ok(())
            } else {
                Err(VmBamlError::Io {
                    message: "Subprocess handle is closed".into(),
                }
                .into())
            }
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        process: owned::sys::Subprocess,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        if let Ok(handle) = downcast_process_handle(&process) {
            // Closing a control handle must never signal termination. The reaper
            // continues independently until the OS process exits.
            handle
                .kill_tx
                .lock()
                .expect("process control lock poisoned")
                .take();
        }
        // Stream objects own their pipe handles independently. Releasing this
        // process handle must not close a stream another BAML value still owns.
        SysOpOutput::ok(())
    }
}

/// Run a command while writing input and draining both outputs concurrently.
/// The reaper owns the OS child independently of the caller's VM handle.
async fn run_process(
    cmd: &mut tokio::process::Command,
    options: owned::sys::ProcessOptions,
    working_dir: &WorkingDir,
    label: &str,
) -> Result<owned::sys::ProcessOutput, VmRustFnError> {
    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
    let timeout = process_timeout(options.timeout.as_ref())?;
    configure_process(cmd, &options, working_dir)?;
    let input = process_input(&options)?;
    let mut child = cmd.spawn().map_err(|error| VmBamlError::Io {
        message: format!("Failed to spawn '{label}': {error}"),
    })?;
    let stdin = child.stdin.take();
    let stdout = child.stdout.take();
    let stderr = child.stderr.take();
    let (kill_tx, mut kill_rx) = tokio::sync::mpsc::unbounded_channel::<ProcessKillRequest>();
    let (status_tx, status_rx) = tokio::sync::oneshot::channel();
    tokio::spawn(async move {
        let status = loop {
            tokio::select! {
                status = child.wait() => break status,
                request = kill_rx.recv() => {
                    match request {
                        Some(ack) => {
                            let result = child.start_kill().map_err(|error| VmBamlError::Io { message: format!("Failed to terminate process: {error}") });
                            let _ = ack.send(result);
                        }
                        None => break child.wait().await,
                    }
                }
            }
        };
        let _ = status_tx.send(status);
    });
    let (result_tx, result_rx) = tokio::sync::oneshot::channel();
    let label = label.to_owned();
    // Caller cancellation drops only its receiver. This task and the reaper
    // continue without making cancellation an implicit termination request.
    tokio::spawn(async move {
        let operation = async move {
            let write = async move {
                if let (Some(mut stdin), Some(input)) = (stdin, input) {
                    // A command can finish without consuming all supplied input.
                    // Still return its output/status when it closes stdin early.
                    if let Err(error) = stdin.write_all(&input).await
                        && !matches!(
                            error.kind(),
                            std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionReset
                        )
                    {
                        return Err(error);
                    }
                }
                Ok::<_, std::io::Error>(())
            };
            let out = async move {
                let mut bytes = Vec::new();
                if let Some(mut pipe) = stdout {
                    pipe.read_to_end(&mut bytes).await?;
                }
                Ok::<_, std::io::Error>(bytes)
            };
            let err = async move {
                let mut bytes = Vec::new();
                if let Some(mut pipe) = stderr {
                    pipe.read_to_end(&mut bytes).await?;
                }
                Ok::<_, std::io::Error>(bytes)
            };
            let (write, stdout, stderr, status) = tokio::join!(write, out, err, status_rx);
            let io_error = |error: std::io::Error| VmBamlError::Io {
                message: error.to_string(),
            };
            write.map_err(io_error)?;
            let status = status
                .map_err(|_| VmBamlError::Io {
                    message: "Process reaper stopped before reporting completion".into(),
                })?
                .map_err(io_error)?;
            Ok::<_, VmRustFnError>((stdout.map_err(io_error)?, stderr.map_err(io_error)?, status))
        };
        let result = match timeout {
            Some(timeout) => match tokio::time::timeout(timeout, operation).await {
                Ok(result) => result,
                Err(_) => {
                    // An explicitly configured run timeout requests termination.
                    // Reaping continues independently; pipe draining cannot extend
                    // the deadline when descendants still hold their ends open.
                    let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
                    if kill_tx.send(ack_tx).is_ok() {
                        let _ = ack_rx.await;
                    }
                    Err(process_timeout_error(&label, Some(timeout)).into())
                }
            },
            None => operation.await,
        };
        let _ = result_tx.send(result);
    });
    let (stdout, stderr, status) = result_rx.await.map_err(|_| VmBamlError::Io {
        message: "Process recording stopped before reporting completion".into(),
    })??;
    let status = process_exit_from_status(status);
    Ok(owned::sys::ProcessOutput {
        stdout,
        stderr,
        exit_code: status.exit_code,
        signal: status.signal,
    })
}

impl io::IoNamespaceSys for NativeSysOps {
    fn host_info(
        &self,
        _: &Arc<BexHeap>,
        _: CallId,
        _: &SysOpContext,
    ) -> SysOpOutput<owned::sys::HostInfo> {
        use std::io::IsTerminal as _;
        SysOpOutput::ok(owned::sys::HostInfo {
            platform: std::env::consts::OS.to_string(),
            architecture: std::env::consts::ARCH.to_string(),
            cpu_count: std::thread::available_parallelism()
                .ok()
                .and_then(|n| i64::try_from(n.get()).ok())
                .unwrap_or(0),
            stderr_is_terminal: std::io::stderr().is_terminal(),
        })
    }

    fn current_dir(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        crate::installation::sys_current_dir(self)
    }
    fn current_exe(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        crate::installation::sys_current_exe(self)
    }
    fn home_dir(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        crate::installation::sys_home_dir(self)
    }
    fn platform(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        crate::installation::sys_platform(self)
    }
    fn host_target(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        crate::installation::sys_host_target(self)
    }

    fn collect_garbage(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        // BexEngine intercepts this operation before ordinary sys-op dispatch
        // so it can release the calling VM's heap permit and coordinate a
        // stop-the-world collection. This fallback keeps the generated IO
        // contract complete for alternate dispatchers.
        SysOpOutput::ok(())
    }

    fn _run(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        program: String,
        args: Vec<String>,
        options: owned::sys::ProcessOptions,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::sys::ProcessOutput> {
        let working_dir = self.working_dir.clone();
        SysOpOutput::async_op(async move {
            let executable = working_dir
                .resolve_program(&program, options.cwd.as_deref())
                .map_err(|error| VmBamlError::Io {
                    message: format!("Failed to resolve '{program}': {error}"),
                })?;
            let mut cmd = tokio::process::Command::new(executable);
            cmd.args(&args);
            run_process(&mut cmd, options, &working_dir, &program).await
        })
    }

    fn _subprocess(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        program: String,
        args: Vec<String>,
        options: owned::sys::ProcessOptions,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::sys::Subprocess> {
        let working_dir = self.working_dir.clone();
        SysOpOutput::async_op(async move {
            use tokio::io::AsyncWriteExt as _;
            let executable = working_dir
                .resolve_program(&program, options.cwd.as_deref())
                .map_err(|error| VmBamlError::Io {
                    message: format!("Failed to resolve '{program}': {error}"),
                })?;
            let mut cmd = tokio::process::Command::new(executable);
            cmd.args(&args);
            configure_process(&mut cmd, &options, &working_dir)?;
            let input = process_input(&options)?;
            let mut child = cmd.spawn().map_err(|error| VmBamlError::Io {
                message: format!("Failed to spawn '{program}': {error}"),
            })?;
            let pid = i64::from(child.id().ok_or_else(|| VmBamlError::Io {
                message: "Missing child PID".into(),
            })?);
            let stdin = child
                .stdin
                .take()
                .map(|pipe| write_pipe(pipe, format!("{program} stdin")));
            let stdout = child
                .stdout
                .take()
                .map(|pipe| read_pipe(pipe, format!("{program} stdout")));
            let stderr = child
                .stderr
                .take()
                .map(|pipe| read_pipe(pipe, format!("{program} stderr")));
            let (kill_tx, mut kill_rx) =
                tokio::sync::mpsc::unbounded_channel::<ProcessKillRequest>();
            let (exit_tx, exit_rx) = tokio::sync::watch::channel(None);
            // Input delivery is independent of waiting for the OS process. A
            // descendant may inherit stdin after the direct child has exited.
            if let (Some(data), Some(pipe)) = (input, stdin.as_ref()) {
                let pipe = downcast_write_pipe(pipe)?;
                tokio::spawn(async move {
                    let mut closed = pipe.close_tx.subscribe();
                    let mut guard = pipe.writer.lock().await;
                    if *closed.borrow() {
                        guard.take();
                        return;
                    }
                    if let Some(mut writer) = guard.take() {
                        tokio::select! {
                            _ = closed.changed() => {},
                            result = writer.write_all(&data) => {
                                if let Err(error) = result
                                    && !matches!(error.kind(), std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionReset)
                                {
                                    tracing::warn!(%error, "Failed to deliver subprocess input");
                                }
                            },
                        }
                    }
                });
            }
            tokio::spawn(async move {
                let result = {
                    let status = loop {
                        tokio::select! {
                            status = child.wait() => break status,
                            request = kill_rx.recv() => {
                                match request {
                                    Some(ack) => {
                                        let result = child.start_kill().map_err(|error| VmBamlError::Io { message: format!("Failed to terminate process: {error}") });
                                        let _ = ack.send(result);
                                    }
                                    None => {
                                        // Handle release only: keep reaping independently.
                                        break child.wait().await;
                                    }
                                }
                            },
                        }
                    };
                    status
                        .map(process_exit_from_status)
                        .map_err(|error| VmBamlError::Io {
                            message: format!("Failed to wait on process: {error}"),
                        })
                };
                let _ = exit_tx.send(Some(result));
            });

            Ok(owned::sys::Subprocess {
                pid,
                stdin,
                stdout,
                stderr,
                _handle: Arc::new(LiveProcessHandle {
                    kill_tx: std::sync::Mutex::new(Some(kill_tx)),
                    exit_rx,
                    label: program,
                }),
            })
        })
    }

    fn _shell(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        command: String,
        options: owned::sys::ProcessOptions,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::sys::ProcessOutput> {
        let working_dir = self.working_dir.clone();
        SysOpOutput::async_op(async move {
            let resolved = crate::shell::default_shell();
            let mut cmd = tokio::process::Command::new(&resolved.path);
            resolved.apply(&mut cmd, &command);
            run_process(&mut cmd, options, &working_dir, &command).await
        })
    }

    fn _sleep(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        delay_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        // Zero or negative does not wait; past `u64` nanoseconds clamps.
        let delay = timeout_from_nanos(&delay_nanos).unwrap_or_default();
        SysOpOutput::async_op(async move {
            tokio::time::sleep(delay).await;
            Ok(())
        })
    }

    fn pid(&self, _heap: &Arc<BexHeap>, _call_id: CallId, _ctx: &SysOpContext) -> SysOpOutput<i64> {
        // `std::process::id` is a `u32` on every platform this crate builds
        // for, so the widening into BAML's i63 `int` is always exact.
        SysOpOutput::ok(i64::from(std::process::id()))
    }
}

// ============================================================================
// Network
// ============================================================================

// Network handles keep each socket in an optional slot so `close()` can take
// it, after which every other reference observes a closed socket.
//
// `TcpStream` ops need `&mut` access, so they hold the guard across the await
// and select against a close notification. `TcpListener`/`UdpSocket` ops only
// need `&self`, so each socket has an inner `Arc` cloned under a brief lock.
// This keeps `close()` deterministic without serializing concurrent
// `accept`/`recv_from`/`send_to` on the same socket.
struct NetTcpStreamHandle {
    stream: tokio::sync::Mutex<Option<tokio::net::TcpStream>>,
    close_tx: tokio::sync::watch::Sender<bool>,
}

impl BexRustData for NetTcpStreamHandle {
    fn measure(&self, _: &mut sys_types::Meter) {}
}

/// A bound listener as `$rust_type` data.
struct NetTcpListenerHandle(tokio::sync::Mutex<Option<Arc<tokio::net::TcpListener>>>);

impl BexRustData for NetTcpListenerHandle {
    fn measure(&self, _: &mut sys_types::Meter) {}
}

impl std::ops::Deref for NetTcpListenerHandle {
    type Target = tokio::sync::Mutex<Option<Arc<tokio::net::TcpListener>>>;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

/// A bound socket as `$rust_type` data.
struct NetUdpSocketHandle(tokio::sync::Mutex<Option<Arc<tokio::net::UdpSocket>>>);

impl BexRustData for NetUdpSocketHandle {
    fn measure(&self, _: &mut sys_types::Meter) {}
}

impl std::ops::Deref for NetUdpSocketHandle {
    type Target = tokio::sync::Mutex<Option<Arc<tokio::net::UdpSocket>>>;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

/// Convert a `Duration._nanoseconds` value (carried as a bigint across the
/// sys-op boundary) into an operation timeout. Zero or negative disables it
/// (`None` — block indefinitely, matching Rust's `Option<Duration>` socket
/// timeouts); a value too large for `u64` nanoseconds (~584 years) clamps to the
/// maximum. Shared by the net sys-ops and the HTTP server, so it lives here
/// (always compiled) rather than behind the `bundle-http` feature.
pub(crate) fn timeout_from_nanos(nanos: &num_bigint::BigInt) -> Option<std::time::Duration> {
    if nanos.sign() != num_bigint::Sign::Plus {
        return None;
    }
    Some(std::time::Duration::from_nanos(
        u64::try_from(nanos).unwrap_or(u64::MAX),
    ))
}

fn downcast_tcpstream(
    stream: &owned::net::TcpStream,
) -> Result<Arc<NetTcpStreamHandle>, VmInternalError> {
    stream
        ._handle
        .clone()
        .downcast_payload::<NetTcpStreamHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<NetTcpStreamHandle>(),
            got: stream._handle.type_id(),
        })
}

fn net_tcp_stream(stream: tokio::net::TcpStream) -> owned::net::TcpStream {
    let (close_tx, _close_rx) = tokio::sync::watch::channel(false);
    owned::net::TcpStream {
        _handle: Arc::new(NetTcpStreamHandle {
            stream: tokio::sync::Mutex::new(Some(stream)),
            close_tx,
        }),
    }
}

fn tcp_stream_closed_error() -> VmBamlError {
    VmBamlError::Io {
        message: "TcpStream is closed".to_string(),
    }
}

fn downcast_tcplistener(
    listener: &owned::net::TcpListener,
) -> Result<Arc<NetTcpListenerHandle>, VmInternalError> {
    listener
        ._handle
        .clone()
        .downcast_payload::<NetTcpListenerHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<NetTcpListenerHandle>(),
            got: listener._handle.type_id(),
        })
}

fn downcast_udpsocket(
    socket: &owned::net::UdpSocket,
) -> Result<Arc<NetUdpSocketHandle>, VmInternalError> {
    socket
        ._handle
        .clone()
        .downcast_payload::<NetUdpSocketHandle>()
        .map_err(|_| VmInternalError::RustTypeError {
            expected: TypeId::of::<NetUdpSocketHandle>(),
            got: socket._handle.type_id(),
        })
}

impl io::IoClassNetTcpStream for NativeSysOps {
    fn _connect(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        addr: String,
        timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::net::TcpStream> {
        // `timeout = null` in BAML arrives as `0n`; `timeout_from_nanos` maps it
        // to `None`, leaving the connect bounded only by the OS default (matching
        // Rust's plain `TcpStream::connect`).
        let timeout = timeout_from_nanos(&timeout_nanos);
        SysOpOutput::async_op(async move {
            let connect = tokio::net::TcpStream::connect(&addr);
            let stream = match timeout {
                Some(dur) => match tokio::time::timeout(dur, connect).await {
                    Ok(result) => result,
                    Err(_elapsed) => {
                        return Err(VmBamlError::Timeout {
                            message: format!("Connecting to '{addr}' timed out"),
                            duration: Some(dur),
                            timeout_type: "timeout".to_string(),
                        }
                        .into());
                    }
                },
                None => connect.await,
            }
            .map_err(|e| VmBamlError::Io {
                message: format!("Failed to connect to '{addr}': {e}"),
            })?;
            Ok(net_tcp_stream(stream))
        })
    }

    fn read(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        stream: owned::net::TcpStream,
        limit: i64,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<Vec<u8>>> {
        use tokio::io::AsyncReadExt;

        let limit = match limit {
            ..=0 => return SysOpOutput::Ready(Ok(Some(Vec::new()))),
            1.. => usize::try_from(limit)
                .unwrap_or(MAX_READ_CHUNK)
                .min(MAX_READ_CHUNK),
        };
        SysOpOutput::async_op(async move {
            let handle = downcast_tcpstream(&stream)?;
            let mut close_rx = handle.close_tx.subscribe();
            if *close_rx.borrow() {
                return Err(tcp_stream_closed_error().into());
            }
            let mut guard = handle.stream.lock().await;
            if *close_rx.borrow() {
                return Err(tcp_stream_closed_error().into());
            }
            let stream = guard.as_mut().ok_or_else(tcp_stream_closed_error)?;
            let mut buffer = vec![0u8; limit];
            let read = tokio::select! {
                biased;
                _ = close_rx.changed() => {
                    return Err(tcp_stream_closed_error().into());
                }
                result = stream.read(&mut buffer) => result.map_err(|e| VmBamlError::Io {
                    message: format!("Failed to read from socket: {e}"),
                })?,
            };
            if read == 0 {
                return Ok(None);
            }
            buffer.truncate(read);
            Ok(Some(buffer))
        })
    }

    fn write_some(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        stream: owned::net::TcpStream,
        data: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        use tokio::io::AsyncWriteExt;

        SysOpOutput::async_op(async move {
            let handle = downcast_tcpstream(&stream)?;
            let mut close_rx = handle.close_tx.subscribe();
            if *close_rx.borrow() {
                return Err(tcp_stream_closed_error().into());
            }
            let mut guard = handle.stream.lock().await;
            if *close_rx.borrow() {
                return Err(tcp_stream_closed_error().into());
            }
            let stream = guard.as_mut().ok_or_else(tcp_stream_closed_error)?;
            let written = tokio::select! {
                biased;
                _ = close_rx.changed() => {
                    return Err(tcp_stream_closed_error().into());
                }
                result = stream.write(&data) => result.map_err(|e| VmBamlError::Io {
                    message: format!("Failed to write to socket: {e}"),
                })?,
            };
            i64::try_from(written).map_err(|_| {
                VmRustFnError::from(VmBamlError::Io {
                    message: "socket write count exceeds int range".to_string(),
                })
            })
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        stream: owned::net::TcpStream,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        use tokio::io::AsyncWriteExt;

        SysOpOutput::async_op(async move {
            let handle = downcast_tcpstream(&stream)?;
            handle.close_tx.send_replace(true);
            // Take the stream out of the shared handle so any other reference
            // observes a closed socket on its next op. Already-closed is a no-op.
            let Some(mut stream) = handle.stream.lock().await.take() else {
                return Ok(());
            };
            // shutdown() flushes pending writes and closes the write half, so
            // peers waiting on EOF (e.g. curl with "Connection: close") stop
            // blocking. We swallow ENOTCONN / NotConnected since the peer may
            // already have closed. Dropping `stream` afterwards fully closes it.
            match stream.shutdown().await {
                Ok(()) => Ok(()),
                Err(e) if e.kind() == std::io::ErrorKind::NotConnected => Ok(()),
                Err(e) => Err(VmBamlError::Io {
                    message: format!("Failed to close socket: {e}"),
                }),
            }
            .map_err(VmRustFnError::from)
        })
    }
}

impl io::IoClassNetTcpListener for NativeSysOps {
    fn bind(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        addr: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::net::TcpListener> {
        SysOpOutput::async_op(async move {
            let listener =
                tokio::net::TcpListener::bind(&addr)
                    .await
                    .map_err(|e| VmBamlError::Io {
                        message: format!("Failed to bind '{addr}': {e}"),
                    })?;
            let handle: Arc<dyn BexRustData> = Arc::new(NetTcpListenerHandle(
                tokio::sync::Mutex::new(Some(Arc::new(listener))),
            ));
            Ok(owned::net::TcpListener { _handle: handle })
        })
    }

    fn accept(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        listener: owned::net::TcpListener,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::net::TcpStream> {
        SysOpOutput::async_op(async move {
            let handle = downcast_tcplistener(&listener)?;
            // Clone the inner socket out under a brief lock so concurrent
            // accepts aren't serialized and close() stays deterministic.
            let inner = handle
                .lock()
                .await
                .as_ref()
                .cloned()
                .ok_or_else(|| VmBamlError::Io {
                    message: "TcpListener is closed".to_string(),
                })?;
            let (stream, _peer) = inner.accept().await.map_err(|e| VmBamlError::Io {
                message: format!("Failed to accept connection: {e}"),
            })?;
            Ok(net_tcp_stream(stream))
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        listener: owned::net::TcpListener,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            // Drop the listener out of the shared handle; subsequent accepts on
            // any reference return a closed error. Dropping the last Arc closes
            // the OS socket.
            let handle = downcast_tcplistener(&listener)?;
            handle.lock().await.take();
            Ok(())
        })
    }
}

impl io::IoClassNetUdpSocket for NativeSysOps {
    fn bind(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        addr: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::net::UdpSocket> {
        SysOpOutput::async_op(async move {
            let socket = tokio::net::UdpSocket::bind(&addr)
                .await
                .map_err(|e| VmBamlError::Io {
                    message: format!("Failed to bind UDP '{addr}': {e}"),
                })?;
            let handle: Arc<dyn BexRustData> = Arc::new(NetUdpSocketHandle(
                tokio::sync::Mutex::new(Some(Arc::new(socket))),
            ));
            Ok(owned::net::UdpSocket { _handle: handle })
        })
    }

    fn _send_to(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        socket: owned::net::UdpSocket,
        data: Vec<u8>,
        addr: String,
        timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<i64> {
        let timeout = timeout_from_nanos(&timeout_nanos);
        SysOpOutput::async_op(async move {
            let handle = downcast_udpsocket(&socket)?;
            let inner = handle
                .lock()
                .await
                .as_ref()
                .cloned()
                .ok_or_else(|| VmBamlError::Io {
                    message: "UdpSocket is closed".to_string(),
                })?;
            let send = inner.send_to(&data, &addr);
            let n = match timeout {
                Some(dur) => match tokio::time::timeout(dur, send).await {
                    Ok(result) => result,
                    Err(_elapsed) => {
                        return Err(VmBamlError::Timeout {
                            message: format!("Sending to '{addr}' timed out"),
                            duration: Some(dur),
                            timeout_type: "timeout".to_string(),
                        }
                        .into());
                    }
                },
                None => send.await,
            }
            .map_err(|e| VmBamlError::Io {
                message: format!("Failed to send to '{addr}': {e}"),
            })?;
            Ok(i64::try_from(n).unwrap_or(i64::MAX))
        })
    }

    fn _recv_from(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        socket: owned::net::UdpSocket,
        timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::net::Datagram> {
        let timeout = timeout_from_nanos(&timeout_nanos);
        SysOpOutput::async_op(async move {
            let handle = downcast_udpsocket(&socket)?;
            let inner = handle
                .lock()
                .await
                .as_ref()
                .cloned()
                .ok_or_else(|| VmBamlError::Io {
                    message: "UdpSocket is closed".to_string(),
                })?;
            // A single datagram can be up to 65507 bytes for IPv4; size the
            // buffer to the max so we never silently truncate a packet.
            let mut buffer = vec![0u8; 65_536];
            let recv = inner.recv_from(&mut buffer);
            let (n, peer) = match timeout {
                Some(dur) => match tokio::time::timeout(dur, recv).await {
                    Ok(result) => result,
                    Err(_elapsed) => {
                        return Err(VmBamlError::Timeout {
                            message: "Receiving datagram timed out".to_string(),
                            duration: Some(dur),
                            timeout_type: "timeout".to_string(),
                        }
                        .into());
                    }
                },
                None => recv.await,
            }
            .map_err(|e| VmBamlError::Io {
                message: format!("Failed to receive datagram: {e}"),
            })?;
            buffer.truncate(n);
            Ok(owned::net::Datagram {
                data: buffer,
                addr: peer.to_string(),
            })
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        socket: owned::net::UdpSocket,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            // Drop the socket out of the shared handle; subsequent send_to /
            // recv_from on any reference return a closed error.
            let handle = downcast_udpsocket(&socket)?;
            handle.lock().await.take();
            Ok(())
        })
    }
}

impl io::IoNamespaceNet for NativeSysOps {}

// ============================================================================
// HTTP
// ============================================================================

impl io::IoClassHttpResponse for NativeSysOps {
    #[cfg(feature = "bundle-http")]
    fn text(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        response: owned::http::Response,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        SysOpOutput::async_op(async move {
            let network = crate::http_server::network_of(&response._body);
            crate::http_server::downcast_body(&response._body)?
                .read_text(network.as_ref())
                .await
                .map_err(VmRustFnError::from)
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn text(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _response: owned::http::Response,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn bytes(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        response: owned::http::Response,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Vec<u8>> {
        SysOpOutput::async_op(async move {
            let network = crate::http_server::network_of(&response._body);
            crate::http_server::downcast_body(&response._body)?
                .read_bytes(network.as_ref())
                .await
                .map(|b| b.to_vec())
                .map_err(VmRustFnError::from)
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn bytes(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _response: owned::http::Response,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Vec<u8>> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn new(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        status_code: i64,
        headers: indexmap::IndexMap<String, String>,
        body: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        SysOpOutput::ok(crate::http_server::build_response(
            status_code,
            headers,
            body,
        ))
    }

    #[cfg(not(feature = "bundle-http"))]
    fn new(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _status_code: i64,
        _headers: indexmap::IndexMap<String, String>,
        _body: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn new_streaming(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        status_code: i64,
        headers: indexmap::IndexMap<String, String>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        SysOpOutput::ok(crate::http_server::build_streaming_response(
            status_code,
            headers,
        ))
    }

    #[cfg(not(feature = "bundle-http"))]
    fn new_streaming(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _status_code: i64,
        _headers: indexmap::IndexMap<String, String>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn write(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        response: owned::http::Response,
        data: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            crate::http_server::downcast_body(&response._body)?
                .write_chunk(data)
                .await
                .map_err(VmRustFnError::from)
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn write(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _response: owned::http::Response,
        _data: Vec<u8>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn end(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        response: owned::http::Response,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::async_op(async move {
            crate::http_server::downcast_body(&response._body)?
                .end_stream()
                .await
                .map_err(VmRustFnError::from)
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn end(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _response: owned::http::Response,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }
}

/// Map a `reqwest` transport error to a category the HTTP ops declare they can
/// throw (`baml.errors.Io | baml.errors.Timeout`), so BAML `catch` clauses can
/// handle network failures instead of them surfacing as an uncatchable host
/// error. The message names the request URL without its credential-bearing
/// parts (see [`transport_error_text`]).
#[cfg(feature = "bundle-http")]
pub(crate) fn http_transport_error(
    context: &str,
    e: &reqwest::Error,
    timeout_options: HttpTimeoutOptions,
) -> VmBamlError {
    let message = format!("{context}: {}", transport_error_text(e));
    if e.is_timeout() {
        // reqwest doesn't say which of its deadlines elapsed. Only the connect
        // deadline fails inside the connector, so a timeout that is also a
        // connect error is that one; anything else is the total.
        let (timeout_type, duration) = if e.is_connect() && timeout_options.connect.is_some() {
            ("connect_timeout", timeout_options.connect)
        } else {
            ("timeout", timeout_options.total)
        };
        VmBamlError::Timeout {
            message,
            duration,
            timeout_type: timeout_type.to_string(),
        }
    } else {
        VmBamlError::Io { message }
    }
}

/// `e`'s text with its request URL sanitized for logs
/// (`btel_settings::network::Scrub::Log`): reqwest renders the full URL, and
/// providers authenticate through it (Vertex express mode's `?key=`, a
/// presigned URL's signature), so credential-named query values, userinfo and
/// the fragment are redacted. Every transport error a BAML caller sees, `Io`
/// or `Timeout`, is built here.
#[cfg(feature = "bundle-http")]
fn transport_error_text(e: &reqwest::Error) -> String {
    let text = e.to_string();
    match e.url() {
        Some(url) => text.replace(
            url.as_str(),
            &btel_settings::network::sanitize_url(url.as_str(), btel_settings::network::Scrub::Log),
        ),
        None => text,
    }
}

/// Parses `Request.method`. A request built with `Request.from_url` has no
/// method until `set_method` is called, and sending it is an error.
#[cfg(feature = "bundle-http")]
fn http_method(method: &str) -> Result<reqwest::Method, VmBamlError> {
    if method.is_empty() {
        return Err(VmBamlError::InvalidArgument {
            message: "Invalid HTTP method '': the request has no method set".to_string(),
        });
    }
    reqwest::Method::from_bytes(method.as_bytes()).map_err(|e| VmBamlError::InvalidArgument {
        message: format!("Invalid HTTP method '{method}': {e}"),
    })
}

/// The environment variables reqwest reads when it builds a client: the proxy
/// settings (`REQUEST_METHOD` marks a CGI process, where `HTTP_PROXY` is
/// ignored) and, on Linux, where the CA certificates are.
#[cfg(feature = "bundle-http")]
const HTTP_CLIENT_ENV: [&str; 11] = [
    "ALL_PROXY",
    "all_proxy",
    "HTTPS_PROXY",
    "https_proxy",
    "HTTP_PROXY",
    "http_proxy",
    "NO_PROXY",
    "no_proxy",
    "REQUEST_METHOD",
    "SSL_CERT_FILE",
    "SSL_CERT_DIR",
];

/// The deadlines one outgoing HTTP request runs under, as `baml.http` resolved
/// them: `total` is `Request.timeout` (end to end, connecting included) and
/// `connect` is `Request.connect_timeout`. `None` is no limit.
#[cfg(feature = "bundle-http")]
#[derive(Clone, Copy, Debug)]
pub(crate) struct HttpTimeoutOptions {
    pub(crate) total: Option<std::time::Duration>,
    pub(crate) connect: Option<std::time::Duration>,
}

#[cfg(feature = "bundle-http")]
impl HttpTimeoutOptions {
    /// Both deadlines cross the sys-op boundary as bigint nanos; `0n`/negative
    /// means no limit (`timeout_from_nanos` returns `None`).
    fn from_nanos(
        timeout_nanos: &num_bigint::BigInt,
        connect_timeout_nanos: &num_bigint::BigInt,
    ) -> Self {
        Self {
            total: timeout_from_nanos(timeout_nanos),
            connect: timeout_from_nanos(connect_timeout_nanos),
        }
    }

    /// A client that enforces the connect deadline (TCP and TLS).
    ///
    /// A client owns its connection pool, so one is shared per connect
    /// deadline: requests reuse open connections instead of connecting (and
    /// handshaking TLS) every time. reqwest reads `HTTP_CLIENT_ENV` when it
    /// builds a client, so those values are part of the key: setting a proxy
    /// between requests applies to the next one.
    fn client(self) -> Result<reqwest::Client, VmBamlError> {
        use std::{
            collections::HashMap,
            sync::{LazyLock, Mutex, PoisonError},
        };

        type Key = (
            Option<std::time::Duration>,
            [Option<std::ffi::OsString>; HTTP_CLIENT_ENV.len()],
        );
        // Keys rarely vary. A program that derives a new connect deadline for
        // every request starts the cache over rather than growing it forever.
        const MAX_CLIENTS: usize = 32;
        static CLIENTS: LazyLock<Mutex<HashMap<Key, reqwest::Client>>> =
            LazyLock::new(Default::default);

        let key = (self.connect, HTTP_CLIENT_ENV.map(baml_env::os_var));
        let mut clients = CLIENTS.lock().unwrap_or_else(PoisonError::into_inner);
        if let Some(client) = clients.get(&key) {
            return Ok(client.clone());
        }
        let mut builder = reqwest::Client::builder();
        if let Some(connect) = self.connect {
            builder = builder.connect_timeout(connect);
        }
        let client = builder
            .build()
            .map_err(|e| http_transport_error("HTTP client initialization failed", &e, self))?;
        if clients.len() >= MAX_CLIENTS {
            clients.clear();
        }
        clients.insert(key, client.clone());
        Ok(client)
    }

    /// Applies the total deadline to a request. One that elapses surfaces as
    /// `reqwest::Error::is_timeout`, which `http_transport_error` maps to
    /// `baml.errors.Timeout`.
    fn apply(self, builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        match self.total {
            Some(total) => builder.timeout(total),
            None => builder,
        }
    }
}

#[cfg(feature = "bundle-http")]
fn build_io_http_response(
    response: reqwest::Response,
    url: String,
    timeout_options: HttpTimeoutOptions,
    network: Option<sys_types::network::NetworkContext>,
) -> owned::http::Response {
    let status = i64::from(response.status().as_u16());
    let headers: indexmap::IndexMap<String, String> = response
        .headers()
        .iter()
        .map(|(k, v)| (k.as_str().to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();
    let body = crate::http_server::HttpBody::client(response, timeout_options);
    owned::http::Response {
        status_code: status,
        headers,
        url,
        _body: traced(network, body),
    }
}

/// Queue a traced request's status and headers, as they arrived.
#[cfg(feature = "bundle-http")]
fn push_connection(
    network: Option<&sys_types::network::NetworkContext>,
    response: &reqwest::Response,
) {
    if let Some(network) = network {
        network.push(sys_types::network::NetworkEventKind::Connection {
            status: response.status().as_u16(),
            headers: response
                .headers()
                .iter()
                .map(|(name, value)| {
                    (
                        name.as_str().to_owned(),
                        String::from_utf8_lossy(value.as_bytes()).into_owned(),
                    )
                })
                .collect(),
        });
    }
}

/// `data` as `$rust_type` data, behind its request's span when traced.
#[cfg(feature = "bundle-http")]
fn traced(
    network: Option<sys_types::network::NetworkContext>,
    data: Arc<dyn BexRustData>,
) -> Arc<dyn BexRustData> {
    match network {
        Some(network) => sys_types::network::NetworkTraced::wrap(network, data),
        None => data,
    }
}

#[cfg(feature = "bundle-http")]
impl io::IoClassHttpTlsConfig for NativeSysOps {
    fn _new(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        cert_pem: Vec<u8>,
        key_pem: Vec<u8>,
        allow_tls1_2: bool,
        handshake_timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::TlsConfig> {
        crate::http_server::tls_config_new(cert_pem, key_pem, allow_tls1_2, handshake_timeout_nanos)
    }
}

#[cfg(not(feature = "bundle-http"))]
impl io::IoClassHttpTlsConfig for NativeSysOps {
    fn _new(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _cert_pem: Vec<u8>,
        _key_pem: Vec<u8>,
        _allow_tls1_2: bool,
        _handshake_timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::TlsConfig> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }
}

#[cfg(feature = "bundle-http")]
impl io::IoClassHttpServer for NativeSysOps {
    fn bind(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        addr: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Server> {
        crate::http_server::bind(addr)
    }

    fn _serve(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        server: owned::http::Server,
        handler: bex_external_types::Handle,
        websocket: bex_external_types::Handle,
        tls_config: Option<owned::http::TlsConfig>,
        allow_http1: bool,
        allow_http2: bool,
        max_body_size: i64,
        max_connections: i64,
        header_read_timeout_nanos: Arc<num_bigint::BigInt>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        crate::http_server::serve(
            server,
            handler,
            websocket,
            tls_config,
            allow_http1,
            allow_http2,
            max_body_size,
            max_connections,
            header_read_timeout_nanos,
            ctx.spawner.clone(),
            ctx.cancel.clone(),
        )
    }
}

#[cfg(not(feature = "bundle-http"))]
impl io::IoClassHttpServer for NativeSysOps {
    fn bind(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _addr: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Server> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    fn _serve(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _server: owned::http::Server,
        _handler: bex_external_types::Handle,
        _websocket: bex_external_types::Handle,
        _tls_config: Option<owned::http::TlsConfig>,
        _allow_http1: bool,
        _allow_http2: bool,
        _max_body_size: i64,
        _max_connections: i64,
        _header_read_timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }
}

impl io::IoClassHttpSseStream for NativeSysOps {
    #[cfg(feature = "bundle-http")]
    fn next(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        sse_stream: owned::http::SseStream,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        use std::sync::atomic::Ordering;

        SysOpOutput::async_op(async move {
            let handle = sys_types::network::NetworkTraced::inner(sse_stream._handle)
                .downcast_payload::<bex_resource_types::ResourceHandle>()
                .map_err(|handle| VmInternalError::RustTypeError {
                    expected: TypeId::of::<bex_resource_types::ResourceHandle>(),
                    got: handle.payload_type_id(),
                })?;

            let (buffer, notify, closed) = crate::registry::REGISTRY
                .get_sse_stream(handle.key())
                .ok_or(VmInternalError::UnresolvedResourceHandle {
                kind: "SSE stream",
                key: handle.key(),
            })?;

            loop {
                let notified = notify.notified();
                {
                    let mut buf = buffer.lock().await;
                    if closed.load(Ordering::Acquire) {
                        buf.done = true;
                        buf.error = None;
                        return Ok(None);
                    }
                    if !buf.events.is_empty() {
                        let taken = std::mem::take(&mut buf.events);
                        buf.publish_retained();
                        let events: Vec<serde_json::Value> = taken
                            .into_iter()
                            .map(|e| {
                                serde_json::json!({
                                    "event": e.event,
                                    "data": e.data,
                                    "id": e.id,
                                })
                            })
                            .collect();
                        return Ok(Some(serde_json::to_string(&events).map_err(|e| {
                            VmInternalError::BridgeFailure {
                                message: format!("failed to serialize SSE events: {e}"),
                            }
                        })?));
                    }
                    if let Some(err) = buf.error.take() {
                        return Err(VmRustFnError::from(err));
                    }
                    if buf.done {
                        return Ok(None);
                    }
                }
                notified.await;
            }
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn next(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _sse_stream: owned::http::SseStream,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _sse_stream: owned::http::SseStream,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        // Dropping the owned SseStream drops its ResourceHandle, which
        // triggers cleanup via ResourceRegistryRef::remove.
        SysOpOutput::ok(())
    }
}

impl io::IoNamespaceHttp for NativeSysOps {
    #[cfg(feature = "bundle-http")]
    fn _fetch(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        url: String,
        timeout_nanos: Arc<num_bigint::BigInt>,
        connect_timeout_nanos: Arc<num_bigint::BigInt>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        let network = ctx.network.clone();
        SysOpOutput::async_op(async move {
            crate::ensure_rustls_crypto_provider();
            let timeout_options =
                HttpTimeoutOptions::from_nanos(&timeout_nanos, &connect_timeout_nanos);
            let client = timeout_options.client()?;
            let response = timeout_options
                .apply(client.get(&url))
                .send()
                .await
                .map_err(|e| http_transport_error("HTTP fetch failed", &e, timeout_options))?;
            push_connection(network.as_ref(), &response);
            let final_url = response.url().to_string();
            Ok(build_io_http_response(
                response,
                final_url,
                timeout_options,
                network,
            ))
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn _fetch(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _url: String,
        _timeout_nanos: Arc<num_bigint::BigInt>,
        _connect_timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn _send(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        request: owned::http::Request,
        timeout_nanos: Arc<num_bigint::BigInt>,
        connect_timeout_nanos: Arc<num_bigint::BigInt>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        let network = ctx.network.clone();
        SysOpOutput::async_op(async move {
            let method = http_method(&request.method)?;

            crate::ensure_rustls_crypto_provider();
            let timeout_options =
                HttpTimeoutOptions::from_nanos(&timeout_nanos, &connect_timeout_nanos);
            let client = timeout_options.client()?;
            let mut builder = client.request(method, &request.url);

            for (k, v) in &request.headers {
                builder = builder.header(k.as_str(), v.as_str());
            }

            if !request.body.is_empty() {
                builder = builder.body(request.body);
            }

            let response = timeout_options
                .apply(builder)
                .send()
                .await
                .map_err(|e| http_transport_error("HTTP send failed", &e, timeout_options))?;
            push_connection(network.as_ref(), &response);
            let final_url = response.url().to_string();
            Ok(build_io_http_response(
                response,
                final_url,
                timeout_options,
                network,
            ))
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn _send(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _request: owned::http::Request,
        _timeout_nanos: Arc<num_bigint::BigInt>,
        _connect_timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::Response> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn _send_sse(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        request: owned::http::Request,
        timeout_nanos: Arc<num_bigint::BigInt>,
        connect_timeout_nanos: Arc<num_bigint::BigInt>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::SseStream> {
        use std::sync::atomic::{AtomicBool, Ordering};

        use futures::StreamExt;
        use sys_types::{network::NetworkEventKind, sse::StreamDecoder};
        use tokio::sync::{Mutex as TokioMutex, Notify};

        use crate::registry::{REGISTRY, SseBuffer};

        // Each parsed event, queued on the request's span when it is traced.
        fn push_events(
            network: Option<&sys_types::network::NetworkContext>,
            events: &[sys_types::sse::SseEvent],
        ) {
            if let Some(network) = network {
                for event in events {
                    network.push(NetworkEventKind::SseEvent {
                        event: (!event.event.is_empty()).then(|| event.event.clone()),
                        data: event.data.clone(),
                        id: event.id.clone(),
                    });
                }
            }
        }

        let network = ctx.network.clone();
        SysOpOutput::async_op(async move {
            let method = http_method(&request.method)?;

            crate::ensure_rustls_crypto_provider();
            let timeout_options =
                HttpTimeoutOptions::from_nanos(&timeout_nanos, &connect_timeout_nanos);
            let client = timeout_options.client()?;
            let mut builder = client.request(method, &request.url);

            for (key, value) in &request.headers {
                builder = builder.header(key.as_str(), value.as_str());
            }

            if !request.body.is_empty() {
                builder = builder.body(request.body.clone());
            }

            let response =
                timeout_options.apply(builder).send().await.map_err(|e| {
                    http_transport_error("SSE connection failed", &e, timeout_options)
                })?;
            push_connection(network.as_ref(), &response);

            if !response.status().is_success() {
                let status = response.status().as_u16();
                // Lossy UTF-8, as `reqwest`'s `text()` without `charset`.
                let body = match response.bytes().await {
                    Ok(body) => {
                        if let Some(network) = &network {
                            network.push(NetworkEventKind::Body(body.clone()));
                        }
                        String::from_utf8_lossy(&body).into_owned()
                    }
                    Err(error) if error.is_timeout() => {
                        return Err(VmRustFnError::from(http_transport_error(
                            "SSE error response body failed",
                            &error,
                            timeout_options,
                        )));
                    }
                    Err(_) => "<could not read body>".to_string(),
                };
                return Err(VmRustFnError::from(VmBamlError::Io {
                    message: format!("SSE request failed with status {status}: {body}"),
                }));
            }

            let url = response.url().to_string();
            let status_code = i64::from(response.status().as_u16());
            let headers: indexmap::IndexMap<String, String> = response
                .headers()
                .iter()
                .map(|(k, v)| (k.as_str().to_string(), v.to_str().unwrap_or("").to_string()))
                .collect();
            // Text SSE, or a binary AWS event stream (Bedrock ConverseStream)
            // surfaced as the same events.
            let content_type = response
                .headers()
                .get(reqwest::header::CONTENT_TYPE)
                .and_then(|v| v.to_str().ok())
                .map(str::to_owned);
            let retained = Arc::new(sys_types::RetainedBytes::new(0));
            let buffer = Arc::new(TokioMutex::new(SseBuffer {
                events: Vec::new(),
                done: false,
                error: None,
                retained: Arc::clone(&retained),
            }));
            let closed = Arc::new(AtomicBool::new(false));
            let notify = Arc::new(Notify::new());

            let buf_clone = buffer.clone();
            let closed_clone = closed.clone();
            let notify_clone = notify.clone();
            let consumer_network = network.clone();
            let consumer = tokio::spawn(async move {
                struct SseDropGuard {
                    buffer: Arc<TokioMutex<SseBuffer>>,
                    closed: Arc<AtomicBool>,
                    notify: Arc<Notify>,
                    completed: bool,
                }

                impl Drop for SseDropGuard {
                    fn drop(&mut self) {
                        if !self.completed {
                            if let Ok(mut buf) = self.buffer.try_lock() {
                                if !buf.done {
                                    if !self.closed.load(Ordering::Acquire) {
                                        buf.error = Some(VmBamlError::Io {
                                            message: "SSE stream task was cancelled".into(),
                                        });
                                    }
                                    buf.done = true;
                                }
                            }
                            self.notify.notify_waiters();
                        }
                    }
                }

                let network = consumer_network.as_ref();
                let mut guard = SseDropGuard {
                    buffer: buf_clone.clone(),
                    closed: closed_clone.clone(),
                    notify: notify_clone.clone(),
                    completed: false,
                };

                let mut parser = StreamDecoder::for_content_type(content_type.as_deref());
                let mut byte_stream = response.bytes_stream();

                loop {
                    let next_chunk = byte_stream.next().await;
                    let Some(chunk_result) = next_chunk else {
                        break;
                    };
                    match chunk_result {
                        Ok(bytes) => {
                            let events = match parser.feed(&bytes) {
                                Ok(events) => events,
                                Err(e) => {
                                    let mut buf = buf_clone.lock().await;
                                    buf.error = Some(VmBamlError::Io {
                                        message: e.to_string(),
                                    });
                                    buf.done = true;
                                    notify_clone.notify_waiters();
                                    guard.completed = true;
                                    return;
                                }
                            };
                            if !events.is_empty() {
                                push_events(network, &events);
                                let mut buf = buf_clone.lock().await;
                                buf.events.extend(events);
                                buf.publish_retained();
                                notify_clone.notify_waiters();
                            }
                        }
                        Err(e) => {
                            let mut buf = buf_clone.lock().await;
                            buf.error = Some(http_transport_error(
                                "SSE stream failed",
                                &e,
                                timeout_options,
                            ));
                            buf.done = true;
                            notify_clone.notify_waiters();
                            guard.completed = true;
                            return;
                        }
                    }
                }

                // Stream ended cleanly — flush any event buffered without a
                // trailing blank line (some servers omit the final delimiter).
                let final_events = match parser.finish() {
                    Ok(events) => events,
                    Err(e) => {
                        let mut buf = buf_clone.lock().await;
                        buf.error = Some(VmBamlError::Io {
                            message: e.to_string(),
                        });
                        buf.done = true;
                        notify_clone.notify_waiters();
                        guard.completed = true;
                        return;
                    }
                };
                push_events(network, &final_events);
                if let Some(network) = network {
                    network.push(NetworkEventKind::StreamEnd);
                }
                let mut buf = buf_clone.lock().await;
                if !final_events.is_empty() {
                    buf.events.extend(final_events);
                    buf.publish_retained();
                }
                buf.done = true;
                notify_clone.notify_waiters();
                guard.completed = true;
            });

            let handle = REGISTRY.register_sse_stream(
                buffer,
                closed,
                notify,
                consumer.abort_handle(),
                url.clone(),
                retained,
            );
            Ok(owned::http::SseStream {
                url,
                status_code,
                headers,
                _handle: traced(network, Arc::new(handle)),
            })
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn _send_sse(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _request: owned::http::Request,
        _timeout_nanos: Arc<num_bigint::BigInt>,
        _connect_timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::http::SseStream> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "http".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }
}

/// The VM [`Type`] an owned sys-op argument denotes.
///
/// Arguments reach a sys-op through `as_owned_but_very_slow`, so these are the
/// shapes it can hand back. The reference-like variants have no VM object type
/// to name here — a `Handle` needs the heap to resolve, a host value has no VM
/// object at all — so they report [`ObjectType::Any`], the documented top of
/// the lattice, rather than a fabricated tag.
#[cfg(feature = "bundle-http")]
fn arg_value_type(value: &BexExternalValue) -> Type {
    match value {
        // `Type::of` likewise reports the lattice top for null.
        BexExternalValue::Null => Type::Object(ObjectType::Any),
        BexExternalValue::Int(_) => Type::Int,
        BexExternalValue::Float(_) => Type::Float,
        BexExternalValue::Bool(_) => Type::Bool,
        BexExternalValue::Bigint(_) => Type::Object(ObjectType::Bigint),
        BexExternalValue::String(_) => Type::Object(ObjectType::String),
        BexExternalValue::Uint8Array(_) => Type::Object(ObjectType::Uint8Array),
        BexExternalValue::Array { .. } => Type::Object(ObjectType::Array),
        BexExternalValue::Map { .. } => Type::Object(ObjectType::Map),
        BexExternalValue::Instance { .. } => Type::Object(ObjectType::Instance),
        // `ObjectType::of` folds both enum objects into `Enum`.
        BexExternalValue::Variant { .. } => Type::Object(ObjectType::Enum),
        BexExternalValue::RustData(_) => Type::Object(ObjectType::RustData),
        // A union tag is not itself a runtime type; the payload carries one.
        BexExternalValue::Union { value, .. } => arg_value_type(value),
        BexExternalValue::Handle(_)
        | BexExternalValue::HostValue(_)
        | BexExternalValue::HostCallable { .. }
        | BexExternalValue::Adt(_) => Type::Object(ObjectType::Any),
    }
}

/// Resolve the registry resource behind a `baml.ws.WebSocket`'s opaque
/// `_handle`. Both failure modes are engine bugs — the handle is minted by
/// `baml.ws._connect` and only ever read back here — so neither is something a
/// BAML program is expected to catch.
#[cfg(feature = "bundle-http")]
fn ws_resource(
    websocket: &owned::ws::WebSocket,
) -> Result<Arc<crate::registry::WsStreamResource>, VmInternalError> {
    let handle = Arc::clone(&websocket._handle)
        .downcast_payload::<bex_resource_types::ResourceHandle>()
        .map_err(|handle| VmInternalError::RustTypeError {
            expected: TypeId::of::<bex_resource_types::ResourceHandle>(),
            got: handle.payload_type_id(),
        })?;
    crate::registry::REGISTRY.get_ws_stream(handle.key()).ok_or(
        VmInternalError::UnresolvedResourceHandle {
            kind: "WebSocket",
            key: handle.key(),
        },
    )
}

/// Build the `baml.ws.CloseEvent` that `next` hands back once the connection
/// has ended.
#[cfg(feature = "bundle-http")]
fn ws_close_event(close: &crate::registry::WsClose) -> BexExternalValue {
    use sys_ops::io::AsBexExternalValue;

    owned::ws::CloseEvent {
        code: i64::from(close.code),
        reason: close.reason.clone(),
    }
    .into_bex_external_value()
}

impl io::IoClassWsWebSocket for NativeSysOps {
    #[cfg(feature = "bundle-http")]
    fn send(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        websocket: owned::ws::WebSocket,
        data: BexExternalValue,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        use futures::SinkExt;
        use tokio_tungstenite::tungstenite::Message;

        // `data: string | uint8array` is enforced by the compiler, so any other
        // shape here is a VM/codegen fault rather than a caller error. `Type`
        // has no union form, so `expected` names the string arm.
        let frame = match data {
            BexExternalValue::String(text) => Message::text(text.to_string()),
            BexExternalValue::Uint8Array(bytes) => Message::binary(bytes),
            other => {
                return SysOpOutput::err(VmInternalError::TypeError {
                    expected: Type::Object(ObjectType::String),
                    got: arg_value_type(&other),
                });
            }
        };

        SysOpOutput::async_op(async move {
            let ws = ws_resource(&websocket)?;
            let mut sink = ws.sink.lock().await;
            let sink = sink.as_mut().ok_or_else(|| VmBamlError::Io {
                message: "WebSocket send failed: the connection is closed".to_string(),
            })?;
            sink.send(frame).await.map_err(|error| VmBamlError::Io {
                message: format!("WebSocket send failed: {error}"),
            })?;
            Ok(())
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn send(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _websocket: owned::ws::WebSocket,
        _data: BexExternalValue,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "websocket".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn next(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        websocket: owned::ws::WebSocket,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<BexExternalValue> {
        use futures::StreamExt;
        use tokio_tungstenite::tungstenite::Message;

        use crate::registry::WsClose;

        SysOpOutput::async_op(async move {
            let ws = ws_resource(&websocket)?;
            if let Some(close) = ws.close.get() {
                return Ok(ws_close_event(close));
            }
            let mut source = ws.source.lock().await;
            loop {
                // Re-checked under the lock: a concurrent `next` may have ended
                // the connection, which publishes the close event and drops the
                // transport in one step.
                let Some(stream) = source.as_mut() else {
                    break Ok(ws_close_event(ws.close.get().unwrap_or_else(|| {
                        unreachable!("WebSocket transport released without a close event")
                    })));
                };
                let frame = match stream.next().await {
                    Some(Ok(frame)) => frame,
                    // End of stream with no closing handshake.
                    None => {
                        let close = WsClose {
                            code: 1006,
                            reason: String::new(),
                        };
                        break Ok(ws_close_event(ws.finish(&mut source, close).await));
                    }
                    // tokio-tungstenite fuses the stream on any read error —
                    // every later poll is `None` — so the connection is over
                    // as of this frame. Publish that now, so `send` and
                    // `hangup` answer from `close`, then report what ended it.
                    Some(Err(error)) => {
                        let close = WsClose {
                            code: 1006,
                            reason: String::new(),
                        };
                        ws.finish(&mut source, close).await;
                        return Err(VmBamlError::Io {
                            message: format!("WebSocket receive failed: {error}"),
                        }
                        .into());
                    }
                };
                match frame {
                    Message::Text(text) => {
                        break Ok(BexExternalValue::String(text.as_str().into()));
                    }
                    Message::Binary(bytes) => {
                        break Ok(BexExternalValue::Uint8Array(bytes.to_vec()));
                    }
                    Message::Close(frame) => {
                        let close = frame.map_or(
                            WsClose {
                                code: 1005,
                                reason: String::new(),
                            },
                            |frame| WsClose {
                                code: u16::from(frame.code),
                                reason: frame.reason.as_str().to_string(),
                            },
                        );
                        break Ok(ws_close_event(ws.finish(&mut source, close).await));
                    }
                    // Tungstenite answers pings itself while reading; neither
                    // control frame is part of the BAML surface.
                    Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {}
                }
            }
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn next(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _websocket: owned::ws::WebSocket,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<BexExternalValue> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "websocket".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }

    #[cfg(feature = "bundle-http")]
    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        websocket: owned::ws::WebSocket,
        code: i64,
        reason: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        use futures::SinkExt;
        use tokio_tungstenite::tungstenite::{
            Message,
            protocol::{CloseFrame, frame::coding::CloseCode},
        };

        // RFC 6455 §7.4: an endpoint may send 1000-1003, 1007-1013, and the
        // registered (3000-3999) and private (4000-4999) ranges. The undefined
        // codes and the four a peer may only ever infer (1004/1005/1006/1015)
        // are caller errors — `is_allowed` draws exactly that line.
        let close_code = match u16::try_from(code).map(CloseCode::from) {
            Ok(close_code) if close_code.is_allowed() => close_code,
            Ok(_) | Err(_) => {
                return SysOpOutput::err(VmBamlError::InvalidArgument {
                    message: format!("{code} is not a valid WebSocket close code"),
                });
            }
        };

        // RFC 6455 §5.5: a control frame carries at most 125 payload bytes,
        // and a close frame spends two on the status code. Tungstenite checks
        // this only when reading — `Frame::close` neither checks nor
        // truncates — so an oversized reason would reach the wire as a
        // protocol violation for the peer to fail the connection over.
        if reason.len() > 123 {
            return SysOpOutput::err(VmBamlError::InvalidArgument {
                message: format!(
                    "WebSocket close reason is {} bytes; a close frame carries at most 123",
                    reason.len()
                ),
            });
        }

        SysOpOutput::async_op(async move {
            let ws = ws_resource(&websocket)?;
            // Best effort from here on: `close` declares only
            // `InvalidArgument`, and a peer that has already gone away is not a
            // caller error. The socket itself is released by `next` when the
            // peer's echo (or the end of the stream) arrives.
            if let Some(sink) = ws.sink.lock().await.as_mut() {
                let _ = sink
                    .send(Message::Close(Some(CloseFrame {
                        code: close_code,
                        reason: reason.into(),
                    })))
                    .await;
            }
            Ok(())
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn close(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _websocket: owned::ws::WebSocket,
        _code: i64,
        _reason: String,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<()> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "websocket".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }
}

impl io::IoNamespaceWs for NativeSysOps {
    #[cfg(feature = "bundle-http")]
    fn _connect(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        url: String,
        headers: indexmap::IndexMap<String, String>,
        timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::ws::WebSocket> {
        use futures::StreamExt;
        use tokio_tungstenite::tungstenite::{
            client::IntoClientRequest,
            http::{HeaderName, HeaderValue},
        };

        let timeout = timeout_from_nanos(&timeout_nanos);
        SysOpOutput::async_op(async move {
            crate::ensure_rustls_crypto_provider();

            let mut request =
                url.as_str()
                    .into_client_request()
                    .map_err(|error| VmBamlError::Io {
                        message: format!("invalid WebSocket URL '{url}': {error}"),
                    })?;
            for (name, value) in &headers {
                let name =
                    HeaderName::from_bytes(name.as_bytes()).map_err(|error| VmBamlError::Io {
                        message: format!("invalid WebSocket header name '{name}': {error}"),
                    })?;
                let value = HeaderValue::from_str(value).map_err(|error| VmBamlError::Io {
                    message: format!("invalid WebSocket header value for '{name}': {error}"),
                })?;
                request.headers_mut().insert(name, value);
            }

            let connect = tokio_tungstenite::connect_async(request);
            let connected = match timeout {
                Some(duration) => match tokio::time::timeout(duration, connect).await {
                    Ok(result) => result,
                    Err(_) => {
                        return Err(VmBamlError::Timeout {
                            message: format!("Connecting WebSocket to '{url}' timed out"),
                            duration: Some(duration),
                            timeout_type: "timeout".to_string(),
                        }
                        .into());
                    }
                },
                None => connect.await,
            };
            let (transport, _) = connected.map_err(|error| VmBamlError::Io {
                message: format!("WebSocket connect failed: {error}"),
            })?;
            let (sink, source) = transport.split();
            let handle =
                crate::registry::REGISTRY.register_ws_stream(Box::new(sink), Box::new(source), url);
            Ok(owned::ws::WebSocket {
                _handle: Arc::new(handle),
            })
        })
    }

    #[cfg(not(feature = "bundle-http"))]
    fn _connect(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        _url: String,
        _headers: indexmap::IndexMap<String, String>,
        _timeout_nanos: Arc<num_bigint::BigInt>,
        _ctx: &SysOpContext,
    ) -> SysOpOutput<owned::ws::WebSocket> {
        SysOpOutput::err(VmPanic::HostUnavailable {
            resource: "websocket".to_string(),
            message: "Operation not supported on this platform".to_string(),
        })
    }
}

// ============================================================================
// Provider auth (GCP OAuth2 tokens, AWS SigV4)
// ============================================================================

/// Surface a `sys_auth` failure as the matching `baml.errors.*` class: a bad or
/// missing credential is an `AccessError` (retrying will not help), a transport
/// failure while resolving one is `Io`.
fn auth_error(err: sys_auth::AuthError) -> VmBamlError {
    match err {
        sys_auth::AuthError::Access(message) => VmBamlError::AccessError { message },
        sys_auth::AuthError::Io(message) => VmBamlError::Io { message },
    }
}

impl io::IoNamespaceAiInternal for NativeSysOps {
    fn render_output_format(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        return_type: ::sys_types::SapTy,
        ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        sys_ops::render_output_format_op(&return_type, ctx)
    }

    fn build_output_format(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        return_type: ::sys_types::SapTy,
        ctx: &SysOpContext,
    ) -> SysOpOutput<owned::ai::OutputFormat> {
        sys_ops::build_output_format_op(&return_type, ctx)
    }

    fn reaches_media(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        t: ::sys_types::SapTy,
        ctx: &SysOpContext,
    ) -> SysOpOutput<bool> {
        sys_ops::reaches_media_op(&t, ctx)
    }

    fn get_return_type(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        function_name: String,
        ctx: &SysOpContext,
    ) -> SysOpOutput<::sys_types::SapTy> {
        sys_ops::get_return_type_op(&function_name, ctx)
    }

    fn _gcp_access_token(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        credentials_json: Option<String>,
        scope: String,
        ctx: &SysOpContext,
    ) -> SysOpOutput<String> {
        let runtime_io = ctx.runtime_io.clone();
        SysOpOutput::async_op(async move {
            sys_auth::access_token(runtime_io, credentials_json, &scope)
                .await
                .map_err(|e| auth_error(e).into())
        })
    }

    fn _gcp_project_id(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        credentials_json: Option<String>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        let runtime_io = ctx.runtime_io.clone();
        SysOpOutput::async_op(async move {
            sys_auth::project_id(runtime_io, credentials_json)
                .await
                .map_err(|e| auth_error(e).into())
        })
    }

    fn _gcp_quota_project_id(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        credentials_json: Option<String>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        let runtime_io = ctx.runtime_io.clone();
        SysOpOutput::async_op(async move {
            Ok(sys_auth::quota_project_id(runtime_io, credentials_json).await)
        })
    }

    fn _aws_sign_request(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        request: BexExternalValue,
        service: String,
        region: Option<String>,
        profile: Option<String>,
        access_key_id: Option<String>,
        secret_access_key: Option<String>,
        session_token: Option<String>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<BexExternalValue> {
        // `baml.http.Request` crosses the sys-op boundary untyped (the
        // generated marshaller only maps classes from the sysop's own
        // namespace), so decode it here. The type checker guarantees the shape,
        // so a failure here is a marshalling bug; it is reported as an
        // `AccessError` because that is what the op's declared contract allows.
        let mut request = match owned::http::Request::from_external(request) {
            Ok(request) => request,
            Err(e) => {
                return SysOpOutput::err(VmBamlError::AccessError {
                    message: format!(
                        "ai.internal._aws_sign_request: argument is not a baml.http.Request: {e:?}"
                    ),
                });
            }
        };
        let runtime_io = ctx.runtime_io.clone();
        SysOpOutput::async_op(async move {
            let opts = sys_auth::AwsSignOptions {
                region,
                profile,
                access_key_id,
                secret_access_key,
                session_token,
                service,
            };
            let headers: Vec<(String, String)> = request
                .headers
                .iter()
                .map(|(k, v)| (k.clone(), v.clone()))
                .collect();
            let signed = sys_auth::sign_request(
                runtime_io,
                &request.method,
                &request.url,
                &headers,
                request.body.as_bytes(),
                &opts,
            )
            .await
            .map_err(auth_error)?;
            for (name, value) in signed {
                request.headers.insert(name, value);
            }
            Ok(sys_ops::io::AsBexExternalValue::into_bex_external_value(
                request,
            ))
        })
    }

    fn _aws_resolve_region(
        &self,
        _heap: &Arc<BexHeap>,
        _call_id: CallId,
        region: Option<String>,
        profile: Option<String>,
        ctx: &SysOpContext,
    ) -> SysOpOutput<Option<String>> {
        let runtime_io = ctx.runtime_io.clone();
        SysOpOutput::async_op(async move {
            Ok(sys_auth::resolve_region(runtime_io, region, profile).await)
        })
    }
}

// BEP-034 Future methods live on the heap `Object::Future` itself (atomic
// state + SetOnce + cancel token) and are dispatched via the native-call
// path (`$rust_function` in `ns_future/future.baml`), not through sys-ops.
// See `bex_vm::package_baml` for the trait impl.

#[cfg(all(test, feature = "bundle-http"))]
mod tests {
    use super::*;

    /// A transport error names its request URL with credential-named query
    /// values, userinfo and the fragment redacted, where providers put credentials (Vertex
    /// express mode's `?key=`, a presigned URL's signature). Port 1 on
    /// loopback refuses the connection, so the failure is immediate and
    /// deterministic.
    #[tokio::test]
    async fn transport_errors_drop_url_credentials() {
        let e = reqwest::Client::new()
            .get("http://user:hunter2@127.0.0.1:1/v1/m:generateContent?key=SEKRIT&alt=sse#frag")
            .send()
            .await
            .unwrap_err();
        let text = transport_error_text(&e);
        for secret in ["SEKRIT", "hunter2", "frag"] {
            assert!(
                !text.contains(secret),
                "transport error leaks {secret:?}: {text}"
            );
        }
        assert!(
            text.contains(
                "(http://127.0.0.1:1/v1/m:generateContent?key=REDACTED&alt=sse#REDACTED)"
            ),
            "{text}"
        );
    }
}