submilli-engine 0.4.3

//! `submilli:fs` — agent-facing filesystem library.
//!
//! Pure Rust host functions registered directly under the package name. Each
//! gated op runs `check_security` before touching the disk; the set of gated
//! capabilities is cataloged in [`crate::stdlib::capabilities`] — keep it in
//! sync when adding or removing a gate (see AGENTS.md).
//!
//! `Stat` / `Peek` / `DirEntry` / `Info` are host-built backing structs the
//! guest holds opaquely and reads through the registered getters. `FileWriter`
//! and the `lines` / `bytes` / `list` iterators wrap [`handles`] payloads in
//! `externref`s; the iterators are closable (`for…of` releases the OS handle
//! on every loop exit) and GC reclaim is the backstop.

use crate::runtime::effects as runtime_effects;
use crate::runtime::host::{
    abi_arg, fatal_host_error, host_value_async, write_submilli_string_struct,
    write_submilli_uint8array_struct,
};
pub mod declaration;
mod effects;
pub mod handles;
mod install;
mod summary;

pub(crate) use install::HOST_VIEWS;

use crate::runtime::filesystem::FilesystemError as PortError;

use wasmtime::{
    Caller, ExternRef, FieldType, Finality, Mutability, Rooted, StorageType, StructRef,
    StructRefPre, StructType, Val, ValType,
};

use crate::runtime::StoreData;
use crate::runtime::call_log::{Payload, Side, record_payload};
use crate::runtime::decision::CallTicket;
use crate::runtime::fs::ContainError;
use crate::runtime::fuel::{self};
use crate::runtime::gc_singleton::singleton_struct;
use crate::runtime::intrinsic_types::{build_intrinsic_types, intrinsic_types};
use crate::runtime::prelude::iterator::{
    as_struct, build_closable_iterator, iter_done, iter_yield, next_closure_type, void_closure_type,
};
use crate::stdlib::abi::{self, backing_struct, externref_field, f64_field, string_field};
use crate::stdlib::shared::{
    check_security_call, quota_refusal, require_writable, write_target_trap,
};
use handles::{BackendFileWriter, BackendReader, ChargedDirIter, kind_of};

pub const MODULE_NAME: &str = "submilli:fs";

pub use declaration::package_declaration;
pub use install::install;

// `$StatBacking` field indices (0 is the vtable).
const STAT_KIND: usize = 1;
const STAT_SIZE: usize = 2;
const STAT_MODIFIED_AT: usize = 3;

// `$PeekBacking` field indices.
const PEEK_PREVIEW: usize = 1;
const PEEK_ENCODING: usize = 2;
const PEEK_LINE_ENDING: usize = 3;
const PEEK_SIZE: usize = 4;

// `$DirEntryBacking` field indices.
const ENTRY_KIND: usize = 1;
const ENTRY_NAME: usize = 2;
const ENTRY_PATH: usize = 3;
const ENTRY_SIZE: usize = 4;

// `$InfoBacking` field indices.
const INFO_MODE: usize = 1;
const INFO_SIZE_LIMIT: usize = 2;
const INFO_ACCESS: usize = 3;
const INFO_VOLUME: usize = 4;
const INFO_MOUNTS: usize = 5;

// `$MountInfoBacking` field indices.
const MOUNT_PATH: usize = 1;
const MOUNT_MODE: usize = 2;
const MOUNT_VOLUME: usize = 3;
const MOUNT_ACCESS: usize = 4;
const MOUNT_SIZE_LIMIT: usize = 5;

// `$FileWriterBacking`: vtable + the externref handle.
const WRITER_HANDLE: usize = 1;

fn stat_backing_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    let intr = build_intrinsic_types(engine)?;
    backing_struct(
        engine,
        &intr,
        vec![
            string_field(&intr), // kind
            f64_field(),         // size
            f64_field(),         // modifiedAt
        ],
    )
}

fn peek_backing_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    let intr = build_intrinsic_types(engine)?;
    backing_struct(
        engine,
        &intr,
        vec![
            string_field(&intr), // preview
            string_field(&intr), // encoding
            string_field(&intr), // lineEnding
            f64_field(),         // size
        ],
    )
}

fn dir_entry_backing_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    let intr = build_intrinsic_types(engine)?;
    backing_struct(
        engine,
        &intr,
        vec![
            string_field(&intr), // kind
            string_field(&intr), // name
            string_field(&intr), // path
            f64_field(),         // size
            wasmtime::FieldType::new(Mutability::Const, StorageType::ValType(ValType::I64)), // nominal marker
        ],
    )
}

fn info_backing_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    let intr = build_intrinsic_types(engine)?;
    backing_struct(
        engine,
        &intr,
        vec![
            string_field(&intr),     // mode
            f64_field(),             // sizeLimit
            string_field(&intr),     // access
            string_field(&intr),     // volume
            abi::array_field(&intr), // mounts
            wasmtime::FieldType::new(Mutability::Const, StorageType::ValType(ValType::I64)), // nominal marker
        ],
    )
}

fn mount_info_backing_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    let intr = build_intrinsic_types(engine)?;
    backing_struct(
        engine,
        &intr,
        vec![
            string_field(&intr), // path
            string_field(&intr), // mode
            string_field(&intr), // volume
            string_field(&intr), // access
            f64_field(),         // sizeLimit
            wasmtime::FieldType::new(Mutability::Const, StorageType::ValType(ValType::I32)), // nominal marker
        ],
    )
}

/// `$FileWriterBacking` — vtable + the `externref` holding the
/// [`BackendFileWriter`].
fn file_writer_backing_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    let intr = build_intrinsic_types(engine)?;
    backing_struct(engine, &intr, vec![externref_field()])
}

/// The iterator env: a host-private single-field struct carrying the handle
/// `externref` through the closure ABI's `(ref any)` slot.
fn handle_env_struct(engine: &wasmtime::Engine) -> wasmtime::Result<StructType> {
    singleton_struct(
        engine,
        Finality::Final,
        None::<StructType>,
        vec![FieldType::new(
            Mutability::Const,
            StorageType::ValType(ValType::EXTERNREF),
        )],
    )
}

// ---------------------------------------------------------------------------
// Extracted op bodies
// ---------------------------------------------------------------------------

/// `info()`: the active VFS configuration as an `$InfoBacking`.
fn build_info(caller: &mut Caller<'_, StoreData>) -> wasmtime::Result<Val> {
    let info = caller.data().vfs_info.clone();
    let mode = match info.mode {
        crate::runtime::VfsMode::None => "none",
        crate::runtime::VfsMode::Ephemeral => "ephemeral",
        crate::runtime::VfsMode::PerSession => "per_session",
        crate::runtime::VfsMode::Named => "named",
    };
    let vfs = caller.data().filesystem.info();
    let root_limit = vfs.maximum_bytes;
    let access = vfs.access.as_str();
    let volume = vfs.volume.unwrap_or_default();
    // Cloned out of the store, which the string writes below borrow mutably.
    let mounts = vfs.mounts;
    let mount_ty = mount_info_backing_struct(caller.engine())?;
    let mut built = Vec::with_capacity(mounts.len());
    for mount in &mounts {
        built.push(build_mount_info(caller, &mount_ty, mount)?);
    }
    let mounts = abi::new_array(caller, &built)?;
    // A named root's limit is the volume's, which the server shares in rather
    // than writes into the info.
    let size_limit = size_limit_number(info.size_limit.or(root_limit));
    let mode = write_submilli_string_struct(caller, mode)?.to_anyref();
    let access = write_submilli_string_struct(caller, access)?.to_anyref();
    let volume = write_submilli_string_struct(caller, &volume)?.to_anyref();
    let ty = info_backing_struct(caller.engine())?;
    abi::new_backing(
        caller,
        ty,
        &[
            Val::AnyRef(Some(mode)),
            Val::F64(size_limit.to_bits()),
            Val::AnyRef(Some(access)),
            Val::AnyRef(Some(volume)),
            mounts,
            Val::I64(0),
        ],
    )
}

/// One `info().mounts` entry as a `$MountInfoBacking`, fields in `MOUNT_*` order.
fn build_mount_info(
    caller: &mut Caller<'_, StoreData>,
    ty: &StructType,
    mount: &crate::runtime::filesystem::MountInfo,
) -> wasmtime::Result<Val> {
    let path = write_submilli_string_struct(caller, &mount.path)?.to_anyref();
    let mode = write_submilli_string_struct(caller, "named")?.to_anyref();
    let volume = write_submilli_string_struct(caller, &mount.volume)?.to_anyref();
    let access = write_submilli_string_struct(caller, mount.access.as_str())?.to_anyref();
    let size_limit = size_limit_number(mount.maximum_bytes);
    abi::new_backing(
        caller,
        ty.clone(),
        &[
            Val::AnyRef(Some(path)),
            Val::AnyRef(Some(mode)),
            Val::AnyRef(Some(volume)),
            Val::AnyRef(Some(access)),
            Val::F64(size_limit.to_bits()),
            Val::I32(0),
        ],
    )
}

/// `-1` stands for "no limit": the language has no `undefined`, so the field
/// is always present.
fn size_limit_number(limit: Option<u64>) -> f64 {
    limit.map_or(-1.0, |v| v as f64)
}

#[derive(serde::Serialize)]
struct FileWrite {
    path: String,
    bytes: Vec<u8>,
}

async fn write(
    caller: &mut Caller<'_, StoreData>,
    path: String,
    content: Vec<u8>,
) -> wasmtime::Result<()> {
    let bytes = content;
    let ticket = check_security_call(
        &mut *caller,
        "fs.write",
        serde_json::json!({"path": &path, "length": bytes.len()}),
    )?;
    let filesystem = caller.data().filesystem.clone();
    runtime_effects::execute(
        caller,
        "fs.write",
        FileWrite { path, bytes },
        move |caller, effect_id, request| {
            Box::pin(async move {
                charge_path_work(caller, &[&request.path])?;
                let placement = filesystem
                    .placement(&request.path)
                    .map_err(|error| backend_error("fs.write", &request.path, error))?;
                require_writable(&*caller, ticket, &placement, "fs.write", &request.path)?;
                fuel::charge(&mut *caller, fuel::IO, request.bytes.len() as u64)?;
                let work = begin_work(caller)?;
                let result = filesystem
                    .write(&effect_id, Some(work.clone()), &request.path, request.bytes)
                    .await;
                work.settle(&mut *caller)?;
                result.map_err(|error| backend_error("fs.write", &request.path, error))
            })
        },
    )
    .await
}

async fn write_text(
    caller: &mut Caller<'_, StoreData>,
    path: String,
    content: String,
) -> wasmtime::Result<()> {
    let bytes = content.into_bytes();
    let ticket = check_security_call(
        &mut *caller,
        "fs.write",
        serde_json::json!({"path": &path, "length": bytes.len()}),
    )?;
    let filesystem = caller.data().filesystem.clone();
    runtime_effects::execute(
        caller,
        "fs.write",
        FileWrite { path, bytes },
        move |caller, effect_id, request| {
            Box::pin(async move {
                charge_path_work(caller, &[&request.path])?;
                let placement = filesystem
                    .placement(&request.path)
                    .map_err(|error| backend_error("fs.writeText", &request.path, error))?;
                require_writable(&*caller, ticket, &placement, "fs.write", &request.path)?;
                fuel::charge(&mut *caller, fuel::IO, request.bytes.len() as u64)?;
                let work = begin_work(caller)?;
                let result = filesystem
                    .write(&effect_id, Some(work.clone()), &request.path, request.bytes)
                    .await;
                work.settle(&mut *caller)?;
                result.map_err(|error| backend_error("fs.writeText", &request.path, error))
            })
        },
    )
    .await
}

async fn append(
    caller: &mut Caller<'_, StoreData>,
    path: String,
    content: Vec<u8>,
) -> wasmtime::Result<()> {
    let bytes = content;
    let ticket = check_security_call(
        &mut *caller,
        "fs.write",
        serde_json::json!({"path": &path, "length": bytes.len()}),
    )?;
    let filesystem = caller.data().filesystem.clone();
    runtime_effects::execute(
        caller,
        "fs.append",
        FileWrite { path, bytes },
        move |caller, effect_id, request| {
            Box::pin(async move {
                charge_path_work(caller, &[&request.path])?;
                let placement = filesystem
                    .placement(&request.path)
                    .map_err(|error| backend_error("fs.append", &request.path, error))?;
                require_writable(&*caller, ticket, &placement, "fs.write", &request.path)?;
                fuel::charge(&mut *caller, fuel::IO, request.bytes.len() as u64)?;
                let work = begin_work(caller)?;
                let result = filesystem
                    .append(&effect_id, Some(work.clone()), &request.path, request.bytes)
                    .await;
                work.settle(&mut *caller)?;
                result.map_err(|error| backend_error("fs.append", &request.path, error))
            })
        },
    )
    .await
}

async fn append_text(
    caller: &mut Caller<'_, StoreData>,
    path: String,
    content: String,
) -> wasmtime::Result<()> {
    let bytes = content.into_bytes();
    let ticket = check_security_call(
        &mut *caller,
        "fs.write",
        serde_json::json!({"path": &path, "length": bytes.len()}),
    )?;
    let filesystem = caller.data().filesystem.clone();
    runtime_effects::execute(
        caller,
        "fs.append",
        FileWrite { path, bytes },
        move |caller, effect_id, request| {
            Box::pin(async move {
                charge_path_work(caller, &[&request.path])?;
                let placement = filesystem
                    .placement(&request.path)
                    .map_err(|error| backend_error("fs.appendText", &request.path, error))?;
                require_writable(&*caller, ticket, &placement, "fs.write", &request.path)?;
                fuel::charge(&mut *caller, fuel::IO, request.bytes.len() as u64)?;
                let work = begin_work(caller)?;
                let result = filesystem
                    .append(&effect_id, Some(work.clone()), &request.path, request.bytes)
                    .await;
                work.settle(&mut *caller)?;
                result.map_err(|error| backend_error("fs.appendText", &request.path, error))
            })
        },
    )
    .await
}

/// `stat(path)`: a `$StatBacking`, or `undefined` when the path does not exist.
async fn stat_entry(caller: &mut Caller<'_, StoreData>, path: &str) -> wasmtime::Result<Val> {
    let observation = stat_path(caller, path, false).await?;
    let Some(stat) = &observation else {
        return crate::runtime::prelude::undefined::value(caller);
    };
    let kind = write_submilli_string_struct(caller, &stat.kind)?.to_anyref();
    let ty = stat_backing_struct(caller.engine())?;
    abi::new_backing(
        caller,
        ty,
        &[
            Val::AnyRef(Some(kind)),
            Val::F64(stat.size.to_bits()),
            Val::F64(stat.modified_ms.to_bits()),
        ],
    )
}

/// The recorded metadata of `path`, `None` when it does not exist. `follow`
/// resolves a symlink to what it points at, as `list` does; otherwise the link
/// itself is described, as `stat` does.
async fn stat_path(
    caller: &mut Caller<'_, StoreData>,
    path: &str,
    follow: bool,
) -> wasmtime::Result<Option<effects::FileStat>> {
    // `stat` is declared `Stat | undefined` so a program can probe for absence. Resolution now
    // opens the parent, so a missing *ancestor* fails here rather than at the metadata call
    // below — and answering that with a trap would break the contract every caller branches
    // on. Escapes still trap: unreachable-because-outside is not the same as absent.
    // The two reads are recorded under different names, so a replay can never
    // answer one with the other.
    let operation = if follow { "fs.stat.follow" } else { "fs.stat" };
    runtime_effects::execute(
        caller,
        operation,
        path.to_owned(),
        move |caller, effect_id, path| {
            Box::pin(async move {
                charge_path_work(caller, &[&path])?;
                let work = begin_work(caller)?;
                let meta = match caller
                    .data()
                    .filesystem
                    .metadata(&effect_id, Some(work.clone()), &path, follow)
                    .await
                {
                    Ok(meta) => meta,
                    Err(PortError::Containment(ContainError::Io(e)) | PortError::Io(e))
                        if is_absent(&e) =>
                    {
                        return Ok(None);
                    }
                    Err(e) => return Err(backend_error("fs.stat", &path, e)),
                };
                let ft = meta.file_type();
                let size = if ft.is_file() { meta.len() as f64 } else { 0.0 };
                let modified_ms = meta
                    .modified()
                    .ok()
                    .and_then(|t| t.into_std().duration_since(std::time::UNIX_EPOCH).ok())
                    .map_or(0.0, |d| d.as_millis() as f64);
                work.settle(&mut *caller)?;
                Ok(Some(effects::FileStat {
                    kind: kind_of(&ft).to_owned(),
                    size,
                    modified_ms,
                }))
            })
        },
    )
    .await
}

/// `peek(path)`: sniff the first ~256 bytes into a `$PeekBacking`.
async fn peek_file(caller: &mut Caller<'_, StoreData>, path: &str) -> wasmtime::Result<Val> {
    let observation = runtime_effects::execute(
        caller,
        "fs.peek",
        path.to_owned(),
        |caller, effect_id, path| {
            Box::pin(async move {
                charge_path_work(caller, &[&path])?;
                let filesystem = caller.data().filesystem.clone();
                let work = begin_work(caller)?;
                let meta = filesystem
                    .metadata(&effect_id, Some(work.clone()), &path, true)
                    .await
                    .map_err(|e| backend_error("fs.peek", &path, e))?;
                if !meta.is_file() {
                    wasmtime::bail!("fs.peek {}: path is not a regular file", path);
                }
                let buf = filesystem
                    .read(&effect_id, Some(work.clone()), &path, 0, 256)
                    .await
                    .map_err(|e| backend_error("fs.peek", &path, e))?;

                work.settle(&mut *caller)?;
                Ok(effects::FilePreview {
                    bytes: runtime_effects::OutcomeBytes(buf),
                    size: meta.len(),
                })
            })
        },
    )
    .await?;
    let buf = &observation.bytes.0;
    let (encoding, content_start) = detect_encoding(buf);
    let content = &buf[content_start..];
    let line_ending = if content.windows(2).any(|w| w == b"\r\n") {
        "crlf"
    } else {
        "lf"
    };
    let preview = String::from_utf8_lossy(content).into_owned();

    let preview = write_submilli_string_struct(caller, &preview)?.to_anyref();
    let encoding = write_submilli_string_struct(caller, encoding)?.to_anyref();
    let line_ending = write_submilli_string_struct(caller, line_ending)?.to_anyref();
    let ty = peek_backing_struct(caller.engine())?;
    abi::new_backing(
        caller,
        ty,
        &[
            Val::AnyRef(Some(preview)),
            Val::AnyRef(Some(encoding)),
            Val::AnyRef(Some(line_ending)),
            Val::F64((observation.size as f64).to_bits()),
        ],
    )
}

/// BOM-first encoding sniff; falls back to a UTF-8 validity check, then latin1.
/// Returns the detected label and the byte offset where content starts.
fn detect_encoding(buf: &[u8]) -> (&'static str, usize) {
    if buf.starts_with(&[0xEF, 0xBB, 0xBF]) {
        ("utf-8", 3)
    } else if buf.starts_with(&[0xFF, 0xFE]) {
        ("utf-16le", 2)
    } else if buf.starts_with(&[0xFE, 0xFF]) {
        ("utf-16be", 2)
    } else if std::str::from_utf8(buf).is_ok() {
        ("utf-8", 0)
    } else {
        ("latin1", 0)
    }
}

/// The shared `read`/`readText` path: gate, resolve, refuse directories, and
/// load the whole file — or `None` when it exceeds `fs.maxReadSize` (the
/// caller surfaces that as a guest `null`).
async fn read_whole_capped(
    caller: &mut Caller<'_, StoreData>,
    path: &str,
    op: &'static str,
) -> wasmtime::Result<Option<runtime_effects::OutcomeBytes>> {
    let ctx = format!("fs.{op}");
    let ticket = check_security_call(&mut *caller, "fs.read", serde_json::json!({ "path": path }))?;
    let label = if op == "read" {
        "fs.read"
    } else {
        "fs.readText"
    };
    runtime_effects::execute(
        caller,
        label,
        path.to_owned(),
        move |caller, effect_id, path| {
            Box::pin(async move {
                charge_path_work(caller, &[&path])?;
                let filesystem = caller.data().filesystem.clone();
                let work = begin_work(caller)?;
                let meta = filesystem
                    .metadata(&effect_id, Some(work.clone()), &path, true)
                    .await
                    .map_err(|e| backend_error(&ctx, &path, e))?;
                if meta.is_dir() {
                    wasmtime::bail!("{ctx} {}: path is a directory", path);
                }
                if meta.len() > caller.data().fs_max_read_size {
                    record_payload(&*caller, ticket, Side::Response, || {
                        Payload::meta(serde_json::json!({ "over_max_read_size": meta.len() }))
                            .with_size(0)
                    });
                    return Ok(None);
                }
                // The size is known before the read, so the charge comes first.
                fuel::charge(&mut *caller, fuel::IO, meta.len())?;
                let bytes = filesystem
                    .read(
                        &effect_id,
                        Some(work.clone()),
                        &path,
                        0,
                        caller.data().fs_max_read_size,
                    )
                    .await
                    .map_err(|e| backend_error(&ctx, &path, e))?;
                record_payload(&*caller, ticket, Side::Response, || {
                    Payload::meta(serde_json::Value::Null).with_body(&bytes)
                });
                work.settle(&mut *caller)?;
                Ok(Some(runtime_effects::OutcomeBytes(bytes)))
            })
        },
    )
    .await
}

/// `readBytes(path, offset, length)`: validated random-access range read; may
/// come back short at EOF.
async fn read_byte_range(
    caller: &mut Caller<'_, StoreData>,
    path: &str,
    offset: i64,
    length: i64,
) -> wasmtime::Result<runtime_effects::OutcomeBytes> {
    if offset < 0 {
        return Err(crate::runtime::host::range_error(
            "fs.readBytes: negative offset",
        ));
    }
    if length < 0 {
        return Err(crate::runtime::host::range_error(
            "fs.readBytes: negative length",
        ));
    }
    if (length as u64) > caller.data().fs_max_read_size {
        return Err(crate::runtime::host::range_error(format!(
            "fs.readBytes: length {length} exceeds fs.maxReadSize"
        )));
    }
    let request = effects::ReadRange {
        path: path.to_owned(),
        offset: offset as u64,
        length: length as u64,
    };
    runtime_effects::execute(
        caller,
        "fs.readBytes",
        request,
        |caller, effect_id, request| {
            Box::pin(async move {
                charge_path_work(caller, &[&request.path])?;
                fuel::charge(&mut *caller, fuel::IO, request.length)?;
                let work = begin_work(caller)?;
                let result = caller
                    .data()
                    .filesystem
                    .read(
                        &effect_id,
                        Some(work.clone()),
                        &request.path,
                        request.offset,
                        request.length,
                    )
                    .await;
                work.settle(&mut *caller)?;
                result
                    .map(runtime_effects::OutcomeBytes)
                    .map_err(|error| backend_error("fs.readBytes", &request.path, error))
            })
        },
    )
    .await
}

/// `writer(path)`: open the temp sibling and wrap the [`BackendFileWriter`] in
/// a `$FileWriterBacking`.
async fn open_writer(
    caller: &mut Caller<'_, StoreData>,
    ticket: Option<CallTicket>,
    path: &str,
) -> wasmtime::Result<Val> {
    let writer = effects::open_writer(caller, ticket, path.to_owned()).await?;
    let writer = BackendFileWriter::new(writer, &caller.data().tenant_limits)
        .map_err(crate::runtime::host::fatal_host_error)?;
    let handle = ExternRef::new(&mut *caller, writer)?;
    let ty = file_writer_backing_struct(caller.engine())?;
    abi::new_backing(caller, ty, &[Val::ExternRef(Some(handle))])
}

// ---------------------------------------------------------------------------
// Iterators
// ---------------------------------------------------------------------------

type IterCallback =
    for<'a> fn(
        &'a mut Caller<'_, StoreData>,
        &'a [Val],
    ) -> Box<dyn std::future::Future<Output = wasmtime::Result<Val>> + Send + 'a>;

/// Wrap a charged handle in a closable iterator: the handle rides an
/// `externref` inside a host-private env struct; `next_step` / `close_step`
/// are the per-kind callbacks.
fn make_handle_iterator<T: handles::Closable + 'static>(
    caller: &mut Caller<'_, StoreData>,
    payload: T,
    next_step: IterCallback,
) -> wasmtime::Result<Val> {
    let handle = ExternRef::new(&mut *caller, Some(payload))?;
    let env_ty = handle_env_struct(caller.engine())?;
    let pre = StructRefPre::new(&mut *caller, env_ty);
    let env = StructRef::new(&mut *caller, &pre, &[Val::ExternRef(Some(handle))])?;

    let intr = intrinsic_types(&mut *caller)?;
    let (next_ty, _) = next_closure_type(caller.engine(), &intr)?;
    let (close_ty, _) = void_closure_type(caller.engine(), &intr)?;
    let next_fn = host_value_async(&mut *caller, next_ty, next_step);
    let close_fn = host_value_async(&mut *caller, close_ty, |caller, params| {
        Box::new(async move { close_handle_of::<T>(caller, params).await })
    });
    build_closable_iterator(
        caller,
        next_fn,
        close_fn,
        Val::AnyRef(Some(env.to_anyref())),
    )
}

/// Pull the handle `externref` back out of the iterator env struct.
fn env_handle(
    caller: &mut Caller<'_, StoreData>,
    env: &Val,
) -> wasmtime::Result<Option<Rooted<ExternRef>>> {
    let st = as_struct(caller, env, "fs iterator env")?;
    match st.field(&mut *caller, 0)? {
        Val::ExternRef(handle) => Ok(handle),
        other => Err(wasmtime::Error::msg(format!(
            "fs iterator env: expected externref, got {other:?}"
        ))),
    }
}

/// Borrow the typed handle payload out of the env, erroring with `ctx` when
/// the externref was reclaimed or holds an unexpected payload.
fn handle_payload<'a, T: 'static>(
    caller: &'a mut Caller<'_, StoreData>,
    handle: &Rooted<ExternRef>,
    ctx: &str,
) -> wasmtime::Result<&'a mut T> {
    handle
        .data_mut(&mut *caller)?
        .ok_or_else(|| wasmtime::Error::msg(format!("{ctx}: handle externref reclaimed")))?
        .downcast_mut::<T>()
        .ok_or_else(|| wasmtime::Error::msg(format!("{ctx}: unexpected externref payload")))
}

async fn reader_next(caller: &mut Caller<'_, StoreData>, params: &[Val]) -> wasmtime::Result<Val> {
    use crate::runtime::filesystem::ReadItem;
    let Some(handle) = env_handle(caller, abi_arg(params, 0)?)? else {
        return iter_done(caller);
    };
    let descriptor = handle_payload::<Option<BackendReader>>(caller, &handle, "fs.next")?
        .as_ref()
        .map(BackendReader::id);
    let output = runtime_effects::execute(
        caller,
        "fs.reader.next",
        descriptor,
        move |caller, effect_id, _| {
            Box::pin(async move {
                let work = begin_work(caller)?;
                let Some(mut reader) =
                    handle_payload::<Option<BackendReader>>(caller, &handle, "fs.next")?.take()
                else {
                    return Err(fatal_host_error("reader handle is unavailable"));
                };
                let result = caller
                    .data()
                    .filesystem
                    .next_reader(
                        &effect_id,
                        Some(work.clone()),
                        &mut reader.reader,
                        &caller.data().tenant_limits,
                    )
                    .await;
                *handle_payload::<Option<BackendReader>>(caller, &handle, "fs.next")? =
                    Some(reader);
                work.settle(&mut *caller)?;
                result.map_err(|error| backend_error("fs.next", "", error))
            })
        },
    )
    .await?;
    match &output.item {
        Some(ReadItem::Line(text)) => {
            let value = write_submilli_string_struct(caller, text)?;
            iter_yield(caller, Val::AnyRef(Some(value.to_anyref())))
        }
        Some(ReadItem::Bytes(bytes)) => {
            let value = write_submilli_uint8array_struct(caller, bytes)?;
            iter_yield(caller, Val::AnyRef(Some(value.to_anyref())))
        }
        None => iter_done(caller),
    }
}

async fn list_next(caller: &mut Caller<'_, StoreData>, params: &[Val]) -> wasmtime::Result<Val> {
    let Some(handle) = env_handle(caller, abi_arg(params, 0)?)? else {
        return iter_done(caller);
    };
    let outcome = next_entry(caller, handle).await?;
    let Some(entry) = &outcome else {
        return iter_done(caller);
    };
    let size = entry.bytes as f64;
    let kind = write_submilli_string_struct(caller, entry.kind)?.to_anyref();
    let name = write_submilli_string_struct(caller, &entry.name)?.to_anyref();
    let path = write_submilli_string_struct(caller, &entry.path)?.to_anyref();
    let ty = dir_entry_backing_struct(caller.engine())?;
    let entry = abi::new_backing(
        caller,
        ty,
        &[
            Val::AnyRef(Some(kind)),
            Val::AnyRef(Some(name)),
            Val::AnyRef(Some(path)),
            Val::F64(size.to_bits()),
            Val::I64(0),
        ],
    )?;
    iter_yield(caller, entry)
}

/// The next entry of the directory `handle` holds, as one recorded effect.
async fn next_entry(
    caller: &mut Caller<'_, StoreData>,
    handle: Rooted<ExternRef>,
) -> wasmtime::Result<Option<crate::runtime::filesystem::DirectoryEntry>> {
    let descriptor = handle_payload::<Option<ChargedDirIter>>(caller, &handle, "fs.list.next")?
        .as_ref()
        .map(ChargedDirIter::id);
    runtime_effects::execute(
        caller,
        "fs.list.next",
        descriptor,
        move |caller, effect_id, _| {
            Box::pin(async move {
                fuel::charge(&mut *caller, fuel::SYSCALL, 1)?;
                let work = begin_work(caller)?;
                let Some(mut directory) =
                    handle_payload::<Option<ChargedDirIter>>(caller, &handle, "fs.list.next")?
                        .take()
                else {
                    return Err(fatal_host_error("directory handle is unavailable"));
                };
                let result = caller
                    .data()
                    .filesystem
                    .next_directory(
                        &effect_id,
                        Some(work.clone()),
                        &mut directory.walk,
                        &caller.data().tenant_limits,
                    )
                    .await;
                *handle_payload::<Option<ChargedDirIter>>(caller, &handle, "fs.list.next")? =
                    Some(directory);
                work.settle(&mut *caller)?;
                result.map_err(|error| backend_error("fs.list.next", "", error))
            })
        },
    )
    .await
}

/// Eager close: drops the OS handle and refunds bytes now. A reclaimed
/// externref is a no-op — `for…of` fires `close` unconditionally, so a
/// double close must not trap.
async fn close_handle_of<T: handles::Closable + 'static>(
    caller: &mut Caller<'_, StoreData>,
    params: &[Val],
) -> wasmtime::Result<Val> {
    if let Some(handle) = env_handle(caller, abi_arg(params, 0)?)? {
        close_handle::<T>(caller, handle).await?;
    }
    crate::runtime::prelude::undefined::value(caller)
}

/// Close the payload `handle` holds, as one recorded effect, and refund its buffer.
async fn close_handle<T: handles::Closable + 'static>(
    caller: &mut Caller<'_, StoreData>,
    handle: Rooted<ExternRef>,
) -> wasmtime::Result<()> {
    let descriptor = handle_payload::<Option<T>>(caller, &handle, "fs.close")?
        .as_ref()
        .map(handles::Closable::id);
    let live_handle = handle;
    runtime_effects::execute(
        caller,
        "fs.iterator.close",
        descriptor,
        move |caller, effect_id, _| {
            let handle = live_handle;
            Box::pin(async move {
                let work = begin_work(caller)?;
                if let Some(mut payload) =
                    handle_payload::<Option<T>>(caller, &handle, "fs.close")?.take()
                {
                    let result = payload
                        .close(
                            caller.data().filesystem.as_ref(),
                            &effect_id,
                            Some(work.clone()),
                            &caller.data().tenant_limits,
                        )
                        .await;
                    *handle_payload::<Option<T>>(caller, &handle, "fs.close")? = Some(payload);
                    work.settle(&mut *caller)?;
                    result.map_err(|error| backend_error("fs.close", "", error))?;
                }
                Ok(())
            })
        },
    )
    .await?;
    if let Some(payload) = handle_payload::<Option<T>>(caller, &handle, "fs.close")?.as_mut() {
        payload.release_buffer();
    }
    Ok(())
}

/// Borrow the [`BackendFileWriter`] out of a `$FileWriterBacking` receiver.
fn writer_payload<'a>(
    caller: &'a mut Caller<'_, StoreData>,
    receiver: &Val,
    ctx: &str,
) -> wasmtime::Result<&'a mut BackendFileWriter> {
    let st = abi::backing_receiver(caller, receiver)?;
    let handle = match st.field(&mut *caller, WRITER_HANDLE)? {
        Val::ExternRef(Some(handle)) => handle,
        Val::ExternRef(None) => {
            return Err(wasmtime::Error::msg(format!(
                "{ctx}: writer handle is null"
            )));
        }
        other => {
            return Err(wasmtime::Error::msg(format!(
                "{ctx}: expected externref handle, got {other:?}"
            )));
        }
    };
    handle_payload::<BackendFileWriter>(caller, &handle, ctx)
}

// ---------------------------------------------------------------------------
// Shared helpers
// ---------------------------------------------------------------------------

/// Whether an I/O error means "this path is not there", for the two operations declared
/// to answer that without trapping. `NotADirectory` counts: a file used as a path
/// component is a path that cannot exist, not an error to propagate.
fn is_absent(err: &std::io::Error) -> bool {
    matches!(
        err.kind(),
        std::io::ErrorKind::NotFound | std::io::ErrorKind::NotADirectory
    )
}

/// Admit path resolution and the operation's first native filesystem call.
fn charge_path_work(caller: &mut Caller<'_, StoreData>, paths: &[&str]) -> wasmtime::Result<()> {
    let path_cost = paths.iter().fold(0u64, |cost, path| {
        cost.saturating_add(caller.data().filesystem.path_cost(path))
    });
    fuel::charge(&mut *caller, fuel::SYSCALL, 1)?;
    fuel::charge_host_fuel(&mut *caller, path_cost)
}

fn f64_arg(val: &Val, ctx: &str) -> wasmtime::Result<f64> {
    match val {
        Val::F64(bits) => Ok(f64::from_bits(*bits)),
        other => Err(crate::runtime::host::type_error(format!(
            "{ctx}: expected f64, got {other:?}"
        ))),
    }
}

fn i32_flag(val: &Val, ctx: &str) -> wasmtime::Result<bool> {
    match val {
        Val::I32(v) => Ok(*v != 0),
        other => Err(crate::runtime::host::type_error(format!(
            "{ctx}: expected i32 flag, got {other:?}"
        ))),
    }
}

fn backend_error(op: &str, path: &str, error: PortError) -> wasmtime::Error {
    match error {
        PortError::Execution(error) => error,
        PortError::Exhausted(_) => wasmtime::Trap::OutOfFuel.into(),
        PortError::Containment(error) => write_target_trap(op, path, &error),
        PortError::Quota(error) => quota_refusal(op, path, error),
        PortError::Backend(error) => crate::runtime::host::fatal_host_error(error),
        other => wasmtime::Error::msg(format!("{op} {path}: {other}")),
    }
}

struct HostMutationBudget<'a, 'b> {
    caller: &'a mut Caller<'b, StoreData>,
    memory: Option<handles::ByteCharge>,
    changed: bool,
}
impl<'a, 'b> HostMutationBudget<'a, 'b> {
    fn new(caller: &'a mut Caller<'b, StoreData>) -> Self {
        Self {
            caller,
            memory: None,
            changed: false,
        }
    }
}
impl crate::runtime::filesystem::MutationBudget for HostMutationBudget<'_, '_> {
    fn begin(&mut self) -> wasmtime::Result<()> {
        self.memory = Some(handles::ByteCharge::new(
            &self.caller.data().tenant_limits,
            0,
        )?);
        Ok(())
    }
    fn reserve(&mut self, bytes: u64) -> wasmtime::Result<()> {
        self.memory
            .as_mut()
            .ok_or_else(|| {
                crate::runtime::host::fatal_host_error("missing filesystem memory scope")
            })?
            .grow(&self.caller.data().tenant_limits, bytes)
            .map_err(Into::into)
    }
    fn end(&mut self) {
        self.memory = None;
    }
    fn effect(&mut self) {
        self.changed = true;
    }
    fn syscalls(&mut self, count: u64, settled: bool) -> wasmtime::Result<()> {
        if settled {
            fuel::settle(&mut *self.caller, fuel::SYSCALL, count)
        } else {
            fuel::charge(&mut *self.caller, fuel::SYSCALL, count)
        }
    }
    fn copied(&mut self, entries: u64, bytes: u64) -> wasmtime::Result<()> {
        self.changed = true;
        fuel::settle(&mut *self.caller, fuel::SYSCALL, entries)?;
        fuel::settle(&mut *self.caller, fuel::IO, bytes)
    }
}

fn begin_work(
    caller: &mut Caller<'_, StoreData>,
) -> wasmtime::Result<std::sync::Arc<fuel::WorkBudget>> {
    Ok(std::sync::Arc::new(fuel::WorkBudget::new(
        fuel::remaining_fuel(caller)?,
    )))
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use crate::compile_script;
    use crate::runtime::security::{CheckOutcome, SecurityCheck};
    use crate::runtime::{
        RuntimeConfig, StoreData, Vfs, VfsInfo, VfsMode, dispatch_main_async, install_runtime_async,
    };

    #[tokio::test]
    async fn subpath_aliases_move_copy_and_cwd() {
        use crate::runtime::{Access, DiskQuota, MountSpec};
        let volume = tempfile::tempdir().unwrap();
        let quota = Arc::new(DiskQuota::new(8, 0));
        let mut vfs = Vfs::tempdir().unwrap();
        for (guest, sub_path) in [
            ("/all", None),
            ("/personal", Some("users/ada")),
            ("/alias", Some("users/ada")),
        ] {
            vfs = vfs
                .with_mount_subpath(
                    MountSpec {
                        guest_path: guest.into(),
                        host: volume.path().into(),
                        volume: "notes".into(),
                        access: Access::ReadWrite,
                        quota: Some(Arc::clone(&quota)),
                    },
                    sub_path,
                )
                .unwrap();
        }
        let vfs = vfs.with_cwd("/personal").unwrap();
        let source = r#"
            import { cwd, writeText, readText, move, copy, remove, mkdir, list } from "submilli:fs";
            function fails(f: () => void): boolean { try { f(); return false; } catch { return true; } }
            function main(): void {
                assert(cwd() === "/personal", "cwd");
                writeText("a", "12345678");
                move("a", "/alias/a");
                assert(readText("a") === "12345678", "same-file move keeps content");
                assert(fails(() => copy("a", "/alias/a", false)), "self-copy refused");
                move("a", "/all/b");
                assert(readText("/all/b") === "12345678", "same-volume move at quota");
                assert(fails(() => remove("/all/users", true)), "aliased mount ancestor protected");
                mkdir("tree", true);
                assert(fails(() => copy("tree", "/alias/tree/child", true)), "descendant copy refused");
                let found = false;
                for (const entry of list(".", false)) { if (entry.name === "tree") found = entry.path === "/personal/tree"; }
                assert(found, "listing uses cwd");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .unwrap();
        assert_eq!(quota.used(), 8);
    }

    #[tokio::test]
    async fn copy_merges_into_existing_mount_root() {
        let volume = tempfile::tempdir().unwrap();
        let vfs = mount_volume(
            Vfs::tempdir().unwrap(),
            "/notes",
            volume.path(),
            "notes",
            crate::runtime::Access::ReadWrite,
            None,
        );
        run_with(
            r#"
            import { mkdir, writeText, readText, copy } from "submilli:fs";
            function main(): void {
                mkdir("/tree", true);
                writeText("/tree/a", "hello");
                copy("/tree", "/notes", true);
                assert(readText("/notes/a") === "hello", "copy merges into mount root");
            }
        "#,
            StoreData::with_vfs("test-run".into(), vfs),
        )
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn line_limit_is_catchable_and_following_reads_work() {
        let source = r#"
            import { writeText, lines } from "submilli:fs";
            function main(): void {
                writeText("/long", "12345"); writeText("/short", "ok\n");
                let caught = false;
                try { for (const line of lines("/long")) { assert(false, "over-limit line"); } }
                catch (e: RangeError) { caught = e.message.includes("maxReadSize"); }
                assert(caught, "line cap raises a typed error");
                let text = "";
                for (const line of lines("/short")) { text += line; }
                assert(text === "ok", "a later reader still works");
            }
        "#;
        let mut data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().unwrap());
        data.fs_max_read_size = 4;
        run_with(source, data).await.unwrap();
    }

    struct DenyAllFs;
    impl SecurityCheck for DenyAllFs {
        fn check(
            &self,
            _caller: &str,
            capability: &str,
            _context: &serde_json::Value,
        ) -> CheckOutcome {
            if capability.starts_with("fs.") {
                CheckOutcome::Deny {
                    rule: None,
                    reason: format!("denied {capability} in test"),
                }
            } else {
                CheckOutcome::Allow { rule: None }
            }
        }
    }

    #[tokio::test]
    async fn narrowed_host_carriers() {
        let source = r#"
import { info, stat, peek, list, writer, writeText, Info, MountInfo, Stat, Peek, DirEntry, FileWriter } from "submilli:fs";

class Parent { value: unknown = null; reset(value: unknown): void { this.value = value; } }
function rejects(read: () => void): void {
 let caught = false;
 try { read(); } catch (e) { caught = e instanceof TypeError; }
 assert(caught, "unrelated carrier must throw TypeError");
}

class InfoField extends Parent { value: Info = info(); }
class StatField extends Parent { value: Stat | null = null; }
class PeekField extends Parent { value: Peek | null = null; }
class EntryField extends Parent { value: DirEntry | null = null; }
class WriterField extends Parent { value: FileWriter | null = null; }
class MountField extends Parent { value: MountInfo | null = null; }
function main(): void {
 writeText("/input.txt", "text");
 const i = new InfoField(); assert(i.value.mode.length > 0, "Info");
 const s = new StatField(); s.reset(stat("/input.txt")); assert(s.value!.size === 4, "Stat");
 i.reset(s.value); rejects(() => { const v = i.value; });
 s.reset(info()); rejects(() => { const v = s.value; });
 const p = new PeekField(); p.reset(peek("/input.txt")); assert(p.value!.size === 4, "Peek");
 const e = new EntryField();
 for (const entry of list("/", false)) { if (entry.name === "input.txt") e.reset(entry); }
 assert(e.value!.name === "input.txt", "DirEntry");
 p.reset(e.value); rejects(() => { const v = p.value; });
 e.reset(peek("/input.txt")); rejects(() => { const v = e.value; });
 const w = new WriterField(); w.reset(writer("/output.txt")); w.value!.close();
 w.reset(info()); rejects(() => { const v = w.value; });
 const m = new MountField(); m.reset(info().mounts[0]); assert(m.value!.path === "/m", "MountInfo");
 m.reset(info()); rejects(() => { const v = m.value; });
 i.reset(info().mounts[0]); rejects(() => { const v = i.value; });
}
"#;
        let volume = tempfile::tempdir().unwrap();
        let vfs = mount_volume(
            Vfs::tempdir().unwrap(),
            "/m",
            volume.path(),
            "m",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("host guards");
    }

    #[tokio::test]
    async fn lines_iterator_closes_handle_on_loop_exit() {
        let source = r#"
            import { writeText, lines } from "submilli:fs";
            function main(): void {
                writeText("/lines.txt", "a\nb\nc\n");
                let count: number = 0;
                for (const line of lines("/lines.txt")) {
                    count = count + 1;
                }
                assert(count === 3, "iterated all 3 lines");
            }
        "#;
        run_and_assert_handles_empty(source).await;
    }

    #[tokio::test]
    async fn lines_iterator_closes_handle_on_break() {
        let source = r#"
            import { writeText, lines } from "submilli:fs";
            function main(): void {
                writeText("/lines.txt", "a\nb\nc\nd\ne\n");
                let count: number = 0;
                for (const line of lines("/lines.txt")) {
                    count = count + 1;
                    if (count === 2) {
                        break;
                    }
                }
                assert(count === 2, "broke after 2 lines");
            }
        "#;
        run_and_assert_handles_empty(source).await;
    }

    #[tokio::test]
    async fn bytes_iterator_closes_handle_on_break() {
        let source = r#"
            import { write, bytes } from "submilli:fs";
            function main(): void {
                const payload: Uint8Array = new Uint8Array([0, 1, 2, 3, 4, 5, 6, 7]);
                write("/blob.bin", payload);
                let chunks: number = 0;
                for (const chunk of bytes("/blob.bin", 2)) {
                    chunks = chunks + 1;
                    if (chunks === 2) {
                        break;
                    }
                }
                assert(chunks === 2, "broke after 2 chunks");
            }
        "#;
        run_and_assert_handles_empty(source).await;
    }

    #[tokio::test]
    async fn list_iterator_closes_handle_on_break() {
        let source = r#"
            import { mkdir, writeText, list, DirEntry } from "submilli:fs";
            function main(): void {
                mkdir("/d", false);
                writeText("/d/a.txt", "a");
                writeText("/d/b.txt", "b");
                writeText("/d/c.txt", "c");
                let count: number = 0;
                for (const entry of list("/d", false)) {
                    count = count + 1;
                    if (count === 1) {
                        break;
                    }
                }
                assert(count === 1, "broke after first entry");
            }
        "#;
        run_and_assert_handles_empty(source).await;
    }

    #[tokio::test]
    async fn lines_iterator_closes_handle_on_throw() {
        let source = r#"
            import { writeText, lines } from "submilli:fs";
            function main(): void {
                writeText("/lines.txt", "alpha\nbeta\ngamma\n");
                let caught: boolean = false;
                try {
                    for (const line of lines("/lines.txt")) {
                        if (line === "beta") {
                            throw new Error("unwind from loop");
                        }
                    }
                } catch (e: Error) {
                    caught = true;
                }
                assert(caught, "outer catch saw the throw");
            }
        "#;
        run_and_assert_handles_empty(source).await;
    }

    /// The for-of `finally` closes every iterator, and eager `close()` refunds the
    /// reader's charged bytes immediately. So after a clean run the store's
    /// host-attached counter must be back to zero — the externref-era equivalent of
    /// "the handle map is empty".
    async fn run_and_assert_handles_empty(source: &str) {
        let compiled =
            compile_script(source, "test.subm", crate::FileId(0), &[], &[]).expect("compile clean");
        let cfg = RuntimeConfig::default();
        let engine = cfg.engine().expect("engine");
        let data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().expect("tempdir"));
        let mut store = cfg.store(&engine, data).expect("store");
        let module = wasmtime::Module::new(&engine, &compiled.wasm).expect("module");
        let mut linker = wasmtime::Linker::<StoreData>::new(&engine);
        install_runtime_async(&mut linker, &mut store)
            .await
            .expect("install");
        let inst = linker
            .instantiate_async(&mut store, &module)
            .await
            .expect("instantiate");
        dispatch_main_async(&mut store, &inst)
            .await
            .expect("main ran without trap");
        let live = store.data().tenant_limits.host_attached_bytes();
        assert_eq!(
            live, 0,
            "host-attached bytes should be refunded after for-of finally — got {live}",
        );
    }

    struct RecordingCheck {
        seen: std::sync::Mutex<Vec<(String, String)>>,
    }
    impl SecurityCheck for RecordingCheck {
        fn check(
            &self,
            caller: &str,
            capability: &str,
            _context: &serde_json::Value,
        ) -> CheckOutcome {
            self.seen
                .lock()
                .unwrap()
                .push((caller.to_string(), capability.to_string()));
            CheckOutcome::Allow { rule: None }
        }
    }

    struct ReadContextCheck {
        seen: std::sync::Mutex<Vec<serde_json::Value>>,
    }
    impl SecurityCheck for ReadContextCheck {
        fn check(
            &self,
            _caller: &str,
            capability: &str,
            context: &serde_json::Value,
        ) -> CheckOutcome {
            if capability == "fs.read" {
                self.seen.lock().unwrap().push(context.clone());
            }
            CheckOutcome::Allow { rule: None }
        }
    }

    #[tokio::test]
    async fn read_bytes_capability_omits_offset() {
        let source = r#"
            import { writeText, readBytes } from "submilli:fs";
            function main(): void {
                writeText("/x.txt", "abcdef");
                const bytes = readBytes("/x.txt", 2, 3);
                assert(bytes.length === 3, "read requested range");
            }
        "#;
        let recording = Arc::new(ReadContextCheck {
            seen: std::sync::Mutex::new(Vec::new()),
        });
        let mut data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().expect("tempdir"));
        data.security_check = recording.clone();
        run_with(source, data).await.expect("main ran");

        let seen = recording.seen.lock().unwrap();
        let context = seen
            .iter()
            .find(|context| context.get("length").is_some())
            .expect("readBytes capability context");
        assert_eq!(context["path"], "/x.txt");
        assert_eq!(context["length"], 3);
        assert!(
            context.get("offset").is_none(),
            "read offset is transport mechanics, not capability context: {context}"
        );
    }

    #[tokio::test]
    async fn a_script_is_attributed_to_main() {
        let source = r#"
            import { writeText, readText } from "submilli:fs";
            function main(): void {
                writeText("/x.txt", "hi");
                const _back = readText("/x.txt");
            }
        "#;
        let recording = Arc::new(RecordingCheck {
            seen: std::sync::Mutex::new(Vec::new()),
        });
        let compiled =
            compile_script(source, "test.subm", crate::FileId(0), &[], &[]).expect("compile clean");
        let cfg = RuntimeConfig::default();
        let engine = cfg.engine().expect("engine");
        let mut data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().expect("tempdir"));
        data.security_check = recording.clone();
        let mut store = cfg.store(&engine, data).expect("store");
        let module = wasmtime::Module::new(&engine, &compiled.wasm).expect("module");
        let mut linker = wasmtime::Linker::<StoreData>::new(&engine);
        install_runtime_async(&mut linker, &mut store)
            .await
            .expect("install");
        let inst = linker
            .instantiate_async(&mut store, &module)
            .await
            .expect("instantiate");
        dispatch_main_async(&mut store, &inst)
            .await
            .expect("main ran");

        let seen = recording.seen.lock().unwrap().clone();
        assert!(
            !seen.is_empty(),
            "RecordingCheck should have captured at least one fs.* call"
        );
        for (caller, capability) in &seen {
            assert_eq!(
                caller, "main",
                "expected caller=main for capability {capability}; got {caller}"
            );
        }
    }

    /// Simulates a library-above-main call to verify entry-below-top caller semantics.
    #[tokio::test]
    async fn a_package_is_attributed_to_the_package() {
        let source = r#"
            import { writeText, readText } from "submilli:fs";
            function main(): void {
                writeText("/x.txt", "hi");
                const _back = readText("/x.txt");
            }
        "#;
        let recording = Arc::new(RecordingCheck {
            seen: std::sync::Mutex::new(Vec::new()),
        });
        // Compiled under the package name so the identity rides the wasm frame the runtime
        // reads, rather than being declared out-of-band.
        let compiled = crate::compile::compile_script_owned_by(
            "submilli:url",
            source,
            "test.subm",
            crate::FileId(0),
            &[],
            &[],
        )
        .expect("compile clean");
        let cfg = RuntimeConfig::default();
        let engine = cfg.engine().expect("engine");
        let mut data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().expect("tempdir"));
        data.security_check = recording.clone();
        let mut store = cfg.store(&engine, data).expect("store");
        let module = wasmtime::Module::new(&engine, &compiled.wasm).expect("module");
        let mut linker = wasmtime::Linker::<StoreData>::new(&engine);
        install_runtime_async(&mut linker, &mut store)
            .await
            .expect("install");
        let inst = linker
            .instantiate_async(&mut store, &module)
            .await
            .expect("instantiate");
        dispatch_main_async(&mut store, &inst)
            .await
            .expect("main ran");

        let seen = recording.seen.lock().unwrap().clone();
        assert!(
            !seen.is_empty(),
            "RecordingCheck should have captured calls"
        );
        for (caller, capability) in &seen {
            assert_eq!(
                caller, "submilli:url",
                "expected caller=submilli:url for capability {capability}; got {caller}"
            );
        }
    }

    #[tokio::test]
    async fn deny_policy_throws_catchable_permission_denied() {
        let source = r#"
            import { writeText } from "submilli:fs";
            function main(): string {
                try {
                    writeText("/x.txt", "hi");
                } catch (e: PermissionDeniedError) {
                    return e.capability + "|" + e.caller + "|" + e.reason;
                }
                return "not denied";
            }
        "#;
        let mut data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().expect("tempdir"));
        data.security_check = Arc::new(DenyAllFs);
        let value = run_with(source, data)
            .await
            .expect("denial is caught in user code — main must not trap");
        assert_eq!(
            value.as_deref(),
            Some("fs.write|main|denied fs.write in test")
        );
    }

    /// Run `source` with the supplied `StoreData`, returning the dispatch result
    /// so callers can assert success (asserts held) or a trap.
    async fn run_with(source: &str, data: StoreData) -> wasmtime::Result<Option<String>> {
        run_with_fuel(source, data, None).await
    }

    async fn run_with_fuel(
        source: &str,
        data: StoreData,
        fuel: Option<u64>,
    ) -> wasmtime::Result<Option<String>> {
        let compiled =
            compile_script(source, "test.subm", crate::FileId(0), &[], &[]).expect("compile clean");
        let cfg = RuntimeConfig::default();
        let engine = cfg.engine().expect("engine");
        let mut store = cfg.store(&engine, data).expect("store");
        let module = wasmtime::Module::new(&engine, &compiled.wasm).expect("module");
        let mut linker = wasmtime::Linker::<StoreData>::new(&engine);
        install_runtime_async(&mut linker, &mut store)
            .await
            .expect("install");
        let inst = linker
            .instantiate_async(&mut store, &module)
            .await
            .expect("instantiate");
        if let Some(fuel) = fuel {
            store.set_fuel(fuel).unwrap();
        }
        dispatch_main_async(&mut store, &inst).await
    }

    #[tokio::test]
    async fn recursive_remove_finishes_after_fuel_runs_out_following_an_unlink() {
        let root = tempfile::tempdir().unwrap();
        std::fs::create_dir(root.path().join("tree")).unwrap();
        for index in 0..4 {
            std::fs::write(root.path().join(format!("tree/file{index}")), "x").unwrap();
        }
        let vfs = Vfs::external(root.path().to_path_buf()).unwrap();
        let result = run_with_fuel(
            r#"import { remove, exists } from "submilli:fs";
            function main(): void { remove("/tree", true); assert(!exists("/tree")); }"#,
            StoreData::with_vfs("test-run".into(), vfs.clone()),
            Some(9000),
        )
        .await;
        assert!(matches!(
            result.unwrap_err().downcast_ref::<wasmtime::Trap>(),
            Some(wasmtime::Trap::OutOfFuel)
        ));
        assert!(
            !root.path().join("tree").exists(),
            "effects must finish before the next instruction stops"
        );
        run_with(r#"import { writeText, readText } from "submilli:fs";
            function main(): void { writeText("/healthy", "ok"); assert(readText("/healthy") === "ok"); }"#,
            StoreData::with_vfs("test-run".into(), vfs)).await.unwrap();
    }

    #[tokio::test]
    async fn summary_and_is_empty_describe_what_is_there() {
        let root = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(root.path().join("data/nested")).unwrap();
        std::fs::create_dir(root.path().join("empty")).unwrap();
        std::fs::write(root.path().join("data/b.txt"), vec![b'x'; 2048]).unwrap();
        std::fs::write(root.path().join("data/a.txt"), "").unwrap();
        let vfs = Vfs::external(root.path().to_path_buf()).unwrap();
        let output = run_with(
            r#"import { summary, isEmpty } from "submilli:fs";
            function main(): string {
                assert(isEmpty("/empty"), "empty directory");
                assert(!isEmpty("/data"), "directory with entries");
                assert(isEmpty("/data/a.txt"), "zero-byte file");
                assert(!isEmpty("/data/b.txt"), "file with bytes");
                return [summary("/data"), summary("/empty"), summary("/missing"), summary("/data/b.txt")].join("\n");
            }"#,
            StoreData::with_vfs("test-run".into(), vfs),
        )
        .await
        .unwrap()
        .unwrap();
        // Modification times vary per run; every entry must carry one.
        let lines: Vec<String> = output
            .lines()
            .map(|line| match line.find(", modified ") {
                Some(at) => {
                    let end = line[at..].find(')').map_or(line.len(), |end| at + end);
                    format!("{}{}", &line[..at], &line[end..])
                }
                None => line.to_string(),
            })
            .collect();
        assert_eq!(
            lines,
            [
                "/data: directory, 3 entries: 2 files (2.0 KB), 1 directory",
                "  nested/  (directory)",
                "  a.txt  (file, 0 B)",
                "  b.txt  (file, 2.0 KB)",
                "/empty: empty directory",
                "/missing: does not exist",
                "/data/b.txt: file, 2.0 KB",
            ],
            "{output}"
        );
        assert_eq!(output.matches(", modified ").count(), 4, "{output}");
    }

    #[tokio::test]
    async fn summary_and_is_empty_require_only_fs_list() {
        let root = tempfile::tempdir().unwrap();
        std::fs::create_dir(root.path().join("data")).unwrap();
        std::fs::write(root.path().join("data/a.txt"), "abc").unwrap();
        let recording = Arc::new(RecordingCheck {
            seen: std::sync::Mutex::new(Vec::new()),
        });
        let mut data = StoreData::with_vfs(
            "test-run".into(),
            Vfs::external(root.path().to_path_buf()).unwrap(),
        );
        data.security_check = recording.clone();
        run_with(
            r#"import { summary, isEmpty } from "submilli:fs";
            function main(): void {
                summary("/data"); summary("/data/a.txt"); summary("/missing");
                isEmpty("/data"); isEmpty("/data/a.txt");
            }"#,
            data,
        )
        .await
        .unwrap();
        let seen: Vec<String> = recording
            .seen
            .lock()
            .unwrap()
            .iter()
            .map(|(_, capability)| capability.clone())
            .collect();
        assert_eq!(seen, ["fs.list"; 5], "one check per call, and only fs.list");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn summary_and_is_empty_follow_a_symlink_like_list() {
        let root = tempfile::tempdir().unwrap();
        std::fs::create_dir(root.path().join("data")).unwrap();
        std::fs::create_dir(root.path().join("empty")).unwrap();
        std::fs::write(root.path().join("data/a.txt"), "abc").unwrap();
        std::os::unix::fs::symlink("data", root.path().join("link")).unwrap();
        std::os::unix::fs::symlink("empty", root.path().join("empty-link")).unwrap();
        let vfs = Vfs::external(root.path().to_path_buf()).unwrap();
        let output = run_with(
            r#"import { summary, isEmpty } from "submilli:fs";
            function main(): string {
                assert(!isEmpty("/link"), "a link to a directory with entries");
                assert(isEmpty("/empty-link"), "a link to an empty directory");
                return summary("/link") + "\n" + summary("/");
            }"#,
            StoreData::with_vfs("test-run".into(), vfs),
        )
        .await
        .unwrap()
        .unwrap();
        assert!(
            output.starts_with("/link: directory, 1 entry: 1 file (3 B)\n  a.txt  (file, 3 B"),
            "{output}"
        );
        // Inside a listing, a link is described as the link it is, as `list` does.
        assert!(output.contains("\n  link  (symlink"), "{output}");
    }

    #[tokio::test]
    async fn summary_shows_the_first_hundred_entries_and_counts_the_rest() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("we \"quote\"");
        std::fs::create_dir(&dir).unwrap();
        for index in 0..105 {
            std::fs::write(dir.join(format!("f{index:03}")), "").unwrap();
        }
        std::fs::create_dir(dir.join("zz")).unwrap();
        let vfs = Vfs::external(root.path().to_path_buf()).unwrap();
        let output = run_with(
            r#"import { summary } from "submilli:fs";
            function main(): string { return summary("/we \"quote\""); }"#,
            StoreData::with_vfs("test-run".into(), vfs),
        )
        .await
        .unwrap()
        .unwrap();
        let lines: Vec<&str> = output.lines().collect();
        assert_eq!(lines.len(), 102, "{output}");
        assert!(
            lines[1].starts_with("  zz/  (directory"),
            "directories come first: {output}"
        );
        assert!(lines[2].starts_with("  f000  (file"), "{output}");
        assert!(lines[100].starts_with("  f098  (file"), "{output}");
        assert_eq!(
            lines[101],
            "  … 6 more not shown; iterate `list(\"/we \\\"quote\\\"\", false)` for every entry"
        );
    }

    #[tokio::test]
    async fn is_empty_throws_on_a_missing_path() {
        let root = tempfile::tempdir().unwrap();
        let vfs = Vfs::external(root.path().to_path_buf()).unwrap();
        let error = run_with(
            r#"import { isEmpty } from "submilli:fs";
            function main(): boolean { return isEmpty("/missing"); }"#,
            StoreData::with_vfs("test-run".into(), vfs),
        )
        .await
        .unwrap_err();
        assert!(
            format!("{error:#}").contains("fs.isEmpty /missing: no such file or directory"),
            "{error:#}"
        );
    }

    #[tokio::test]
    async fn none_mode_disables_fs_calls() {
        let source = r#"
            import { writeText } from "submilli:fs";
            function main(): void {
                writeText("/x.txt", "hi");
            }
        "#;
        let data = StoreData::with_vfs("test-run".into(), Vfs::none());
        let err = run_with(source, data)
            .await
            .expect_err("none mode must trap fs calls");
        let msg = format!("{err:?}");
        assert!(
            msg.contains("filesystem is disabled"),
            "expected disabled-fs trap; got: {msg}"
        );
    }

    #[tokio::test]
    async fn info_reports_mode_and_limits() {
        let source = r#"
            import { info, Info } from "submilli:fs";
            function main(): void {
                const fs = info();
                assert(fs.mode === "per_session", "mode surfaced");
                assert(fs.sizeLimit === 1048576, "size limit surfaced");
            }
        "#;
        let mut data = StoreData::with_vfs("test-run".into(), Vfs::tempdir().expect("tempdir"));
        data.vfs_info = VfsInfo {
            mode: VfsMode::PerSession,
            size_limit: Some(1024 * 1024),
        };
        run_with(source, data).await.expect("info asserts hold");
    }

    fn limited(limit: u64) -> StoreData {
        let vfs = Vfs::tempdir().expect("tempdir").with_size_limit(limit);
        let mut data = StoreData::with_vfs("test-run".into(), vfs);
        data.vfs_info.size_limit = Some(limit);
        data
    }

    #[tokio::test]
    async fn size_limit_refuses_growth_but_allows_rewrites_within_it() {
        let source = r#"
            import { writeText, readText, append, info } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                assert(info().sizeLimit === 100, "limit surfaced");
                writeText("/a.txt", "x".repeat(60));
                assert(refused(() => writeText("/b.txt", "y".repeat(60))), "a second file past the limit");
                writeText("/a.txt", "z".repeat(90));
                assert(readText("/a.txt") === "z".repeat(90), "a rewrite that ends within the limit");
                assert(refused(() => append("/a.txt", new Uint8Array(20))), "an append past the limit");
                writeText("/a.txt", "");
                writeText("/b.txt", "y".repeat(100));
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("limit asserts hold");
    }

    #[tokio::test]
    async fn size_limit_stops_a_writer_and_frees_its_temp_file() {
        let source = r#"
            import { writer, writeText, exists } from "submilli:fs";
            function main(): void {
                const w = writer("/log.txt");
                w.writeLine("x".repeat(40));
                let refused = false;
                try { w.writeLine("y".repeat(80)); } catch (e) { refused = e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
                assert(refused, "the writer stops at the limit");
                w.close();
                assert(exists("/log.txt"), "what was written before the limit is kept");
                writeText("/other.txt", "z".repeat(59));
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("writer asserts hold");
    }

    #[tokio::test]
    async fn size_limit_stops_a_copy_at_the_file_that_would_pass_it() {
        let source = r#"
            import { writeText, copy, exists } from "submilli:fs";
            function main(): void {
                writeText("/a.txt", "x".repeat(40));
                copy("/a.txt", "/b.txt", false);
                let refused = false;
                try { copy("/a.txt", "/c.txt", false); } catch (e) { refused = e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
                assert(refused, "a third copy passes the limit");
                assert(!exists("/c.txt"), "the refused copy leaves nothing");
                copy("/b.txt", "/a.txt", false);
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("copy asserts hold");
    }

    #[tokio::test]
    async fn size_limit_counts_what_remove_and_move_free() {
        let source = r#"
            import { writeText, remove, move } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                writeText("/a.txt", "x".repeat(60));
                remove("/a.txt", false);
                writeText("/b.txt", "y".repeat(60));
                writeText("/c.txt", "z".repeat(30));
                move("/c.txt", "/b.txt");
                writeText("/d.txt", "w".repeat(70));
                move("/d.txt", "/d.txt");
                assert(refused(() => writeText("/e.txt", "v".repeat(1))), "a move onto itself frees nothing");
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("remove and move free space");
    }

    #[tokio::test]
    async fn size_limit_counts_a_writer_whose_target_shrinks_mid_write() {
        let source = r#"
            import { writer, writeText } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                writeText("/a.txt", "x".repeat(90));
                const w = writer("/a.txt");
                writeText("/a.txt", "");
                w.writeBytes(new Uint8Array(90));
                w.close();
                assert(refused(() => writeText("/b.txt", "y".repeat(90))), "the writer's 90 bytes still count");
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("stale allowance is not honoured");
    }

    #[tokio::test]
    async fn size_limit_counts_every_writer_on_one_path() {
        let source = r#"
            import { writer, writeText } from "submilli:fs";
            function main(): void {
                writeText("/a.txt", "x".repeat(60));
                const first = writer("/a.txt");
                first.writeBytes(new Uint8Array(30));
                const second = writer("/a.txt");
                let refused = false;
                try { second.writeBytes(new Uint8Array(30)); } catch (e) { refused = e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
                assert(refused, "two open writers' bytes both count");
                first.close();
                second.close();
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("each writer is counted");
    }

    #[tokio::test]
    async fn a_directory_over_its_limit_can_be_cleaned_up() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::write(dir.path().join("big.txt"), vec![b'x'; 200]).expect("seed");
        let vfs = Vfs::external_with_mode(dir.path().to_path_buf(), VfsMode::PerSession)
            .expect("external")
            .with_size_limit(100);
        let source = r#"
            import { writeText, remove } from "submilli:fs";
            function main(): void {
                writeText("/big.txt", "");
                writeText("/big.txt", "x".repeat(10));
                remove("/big.txt", false);
                writeText("/after.txt", "y".repeat(100));
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("truncating and removing recover the limit");
    }

    #[tokio::test]
    async fn a_writer_temp_file_moved_away_stays_counted() {
        let source = r#"
            import { writer, list, move, writeText } from "submilli:fs";
            function main(): void {
                const w = writer("/w.bin");
                w.writeBytes(new Uint8Array(60));
                let temp = "";
                for (const e of list("/", false)) { if (e.name.startsWith("w.bin.")) { temp = e.path; } }
                move(temp, "/kept.bin");
                let failed = false;
                try { w.close(); } catch (e) { failed = true; }
                assert(failed, "close notices its temp file was moved");
                let refused = false;
                try { writeText("/x.txt", "y".repeat(50)); } catch (e) { refused = e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
                assert(refused, "the moved bytes still count");
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("moved temp stays counted");
    }

    // Relies on a removed file's link count reaching 0 while the writer holds it,
    // which Windows may report differently; there the count only over-counts.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_writer_leaves_a_file_the_program_put_at_its_temp_name() {
        let source = r#"
            import { writer, list, remove, writeText, exists } from "submilli:fs";
            function main(): void {
                const w = writer("/w.bin");
                w.writeBytes(new Uint8Array(40000));
                let temp = "";
                for (const e of list("/", false)) { if (e.name.startsWith("w.bin.")) { temp = e.path; } }
                remove(temp, false);
                writeText(temp, "");
                let failed = false;
                try { w.close(); } catch (e) { failed = true; }
                assert(failed, "close notices its temp file was replaced");
                assert(exists(temp), "the program's own file is left alone");
                writeText("/x.bin", "y".repeat(100000));
            }
        "#;
        run_with(source, limited(100000))
            .await
            .expect("a replaced temp file is neither committed nor double-freed");
    }

    #[tokio::test]
    async fn an_unmeasured_directory_opens_as_full() {
        let vfs = Vfs::tempdir()
            .expect("tempdir")
            .with_measured_limit(100, None);
        let source = r#"
            import { writeText, remove } from "submilli:fs";
            function main(): string {
                writeText("/empty.txt", "");
                remove("/empty.txt", false);
                try { writeText("/a.txt", "x"); return "allowed"; }
                catch (e) { return (e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError) ? "true " : "false ") + (e as Error).message; }
            }
        "#;
        let result = run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("runs")
            .expect("a result");
        assert!(result.starts_with("true "), "{result}");
        assert!(result.contains("couldn't be measured"), "{result}");
    }

    #[tokio::test]
    async fn size_limit_counts_what_a_recursive_remove_frees() {
        let source = r#"
            import { mkdir, writeText, remove } from "submilli:fs";
            function main(): void {
                mkdir("/d/e", true);
                writeText("/d/e/a.txt", "x".repeat(60));
                writeText("/d/b.txt", "y".repeat(30));
                remove("/d", true);
                writeText("/c.txt", "z".repeat(100));
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("a recursive remove frees its whole tree");
    }

    #[tokio::test]
    async fn a_removed_file_a_reader_holds_counts_until_the_reader_closes() {
        let source = r#"
            import { writeText, bytes, remove } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                writeText("/f.bin", "x".repeat(90));
                for (const chunk of bytes("/f.bin", 1)) {
                    remove("/f.bin", false);
                    assert(refused(() => writeText("/g.txt", "y".repeat(20))), "the open file's bytes still count");
                    break;
                }
                writeText("/g.txt", "y".repeat(100));
            }
        "#;
        run_with(source, limited(100))
            .await
            .expect("a closed reader frees what its removed file held");
    }

    #[tokio::test]
    async fn a_write_over_a_writers_temp_file_is_counted_in_full() {
        let source = r#"
            import { writer, list, writeText } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                const w = writer("/a.bin");
                w.writeBytes(new Uint8Array(40000));
                let temp = "";
                for (const e of list("/", false)) { if (e.name.startsWith("a.bin.")) { temp = e.path; } }
                writeText(temp, "x".repeat(40000));
                let failed = false;
                try { w.close(); } catch (e) { failed = true; }
                assert(failed, "close notices its temp file was replaced");
                assert(refused(() => writeText("/b.txt", "y".repeat(60001))), "the 40000 at the temp name count");
                writeText("/b.txt", "y".repeat(60000));
            }
        "#;
        run_with(source, limited(100000))
            .await
            .expect("the replaced temp file's bytes are counted exactly once");
    }

    #[tokio::test]
    async fn rewriting_a_file_a_reader_holds_counts_both_copies() {
        let source = r#"
            import { writeText, bytes } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                writeText("/f.bin", "x".repeat(40000));
                for (const chunk of bytes("/f.bin", 1)) {
                    writeText("/f.bin", "z".repeat(40000));
                    assert(refused(() => writeText("/g.txt", "y".repeat(20001))), "the old copy is still on disk");
                    break;
                }
                writeText("/g.txt", "y".repeat(60000));
            }
        "#;
        run_with(source, limited(100000))
            .await
            .expect("the old copy is freed when the reader closes");
    }

    #[tokio::test]
    async fn copying_over_a_file_a_reader_holds_counts_only_the_growth() {
        let source = r#"
            import { writeText, copy, bytes } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                writeText("/f.bin", "x".repeat(40000));
                writeText("/src.bin", "z".repeat(40000));
                for (const chunk of bytes("/f.bin", 1)) {
                    copy("/src.bin", "/f.bin", false);
                    writeText("/g.txt", "y".repeat(20000));
                    assert(refused(() => writeText("/h.txt", "w")), "the limit is reached");
                    break;
                }
            }
        "#;
        run_with(source, limited(100000))
            .await
            .expect("a copy rewrites the held file in place");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn copying_onto_a_link_frees_the_file_it_rewrites() {
        let vfs = Vfs::tempdir().expect("tempdir");
        std::fs::write(vfs.root().join("t.bin"), vec![b't'; 50000]).expect("write");
        std::os::unix::fs::symlink("t.bin", vfs.root().join("l")).expect("symlink");
        let vfs = vfs.with_size_limit(100000);
        let mut data = StoreData::with_vfs("test-run".into(), vfs);
        data.vfs_info.size_limit = Some(100000);
        let source = r#"
            import { writeText, copy } from "submilli:fs";
            function main(): void {
                writeText("/s.bin", "0123456789");
                copy("/s.bin", "/l", false);
                writeText("/g.txt", "y".repeat(99980));
            }
        "#;
        run_with(source, data)
            .await
            .expect("the link's target shrank to the copied 10 bytes");
    }

    #[tokio::test]
    async fn copying_over_a_writers_temp_file_leaves_it_for_the_writer_to_settle() {
        let source = r#"
            import { writeText, writer, list, copy, mkdir } from "submilli:fs";
            function refused(write: () => void): boolean {
                try { write(); return false; } catch (e) { return e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
            }
            function main(): void {
                writeText("/empty", "");
                writeText("/keep.bin", "k".repeat(20000));
                const w = writer("/a.bin");
                w.writeBytes(new Uint8Array(40000));
                let temp = "";
                for (const e of list("/", false)) {
                    if (e.name.startsWith("a.bin.")) { temp = e.path; }
                }
                copy("/empty", temp, false);
                mkdir("/a.bin", false);
                writeText("/a.bin/x", "1");
                let closed = true;
                try { w.close(); } catch (e) { closed = false; }
                assert(!closed, "close fails on the directory");
                assert(refused(() => writeText("/big.bin", "y".repeat(80000))), "20001 bytes remain on disk");
                writeText("/big.bin", "y".repeat(79999));
            }
        "#;
        run_with(source, limited(100000))
            .await
            .expect("the writer's temp file is freed once, when the writer discards it");
    }

    fn mount_volume(
        vfs: Vfs,
        guest: &str,
        host: &std::path::Path,
        volume: &str,
        access: crate::runtime::vfs::Access,
        quota: Option<Arc<crate::runtime::DiskQuota>>,
    ) -> Vfs {
        vfs.with_mount(crate::runtime::vfs::MountSpec {
            guest_path: guest.to_string(),
            host: host.to_path_buf(),
            volume: volume.to_string(),
            access,
            quota,
        })
        .expect("mount")
    }

    #[tokio::test]
    async fn mount_routes_its_paths_to_the_volume() {
        let volume = tempfile::tempdir().expect("volume");
        std::fs::write(volume.path().join("kept.txt"), "kept").expect("seed");
        let session = tempfile::tempdir().expect("session");
        let vfs = mount_volume(
            Vfs::external_with_mode(session.path().to_path_buf(), VfsMode::PerSession)
                .expect("root"),
            "/data/memory",
            volume.path(),
            "project-memory",
            crate::runtime::vfs::Access::ReadWrite,
            Some(Arc::new(crate::runtime::DiskQuota::new(1000, 4))),
        );
        let root = vfs.root().to_path_buf();
        let source = r#"
            import { info, readText, writeText, list, stat, exists } from "submilli:fs";
            function main(): void {
                assert(readText("/data/memory/kept.txt") === "kept", "reads the volume");
                writeText("/data/memory/new.txt", "fresh");
                assert(readText("/data/memory/../memory/new.txt") === "fresh", ".. resolves before routing");
                assert(!exists("/data/memoryx/new.txt"), "a sibling prefix is not the mount");
                assert(stat("/data/memory")!.kind === "directory", "the mount point is a directory");
                const names: string[] = [];
                for (const entry of list("/data", false)) { names.push(entry.name); }
                assert(names.length === 1 && names[0] === "memory", "the parent lists the mount point");
                const paths: string[] = [];
                for (const entry of list("/", true)) { paths.push(entry.path); }
                assert(paths.includes("/data/memory/kept.txt"), "a recursive walk descends into the volume");
                assert(paths.includes("/data/memory/new.txt"), "and sees new files");
                const fs = info();
                assert(fs.access === "read_write" && fs.volume === "", "root info");
                assert(fs.mounts.length === 1, "one mount");
                const mount = fs.mounts[0];
                assert(mount.path === "/data/memory" && mount.mode === "named", "mount path and mode");
                assert(mount.volume === "project-memory" && mount.access === "read_write", "mount volume and access");
                assert(mount.sizeLimit === 1000, "mount size limit");
                const json = JSON.stringify(fs)!;
                assert(json.endsWith(`"mounts":[{"path":"/data/memory","mode":"named","volume":"project-memory","access":"read_write","sizeLimit":1000}]}`), json);
            }
        "#;
        let data = StoreData::with_vfs("test-run".into(), vfs);
        let quota = data.vfs.mounts()[0].quota().cloned().expect("quota");
        run_with(source, data).await.expect("mount asserts hold");
        assert_eq!(
            std::fs::read_to_string(volume.path().join("new.txt")).expect("in the volume"),
            "fresh"
        );
        let placeholder = std::fs::read_dir(root.join("data/memory")).expect("placeholder");
        assert_eq!(placeholder.count(), 0, "the root's mount point stays empty");
        assert_eq!(quota.used(), 9, "the write is charged to the volume");
    }

    #[tokio::test]
    async fn mount_read_only_refuses_every_write() {
        let volume = tempfile::tempdir().expect("volume");
        std::fs::create_dir(volume.path().join("dir")).expect("seed dir");
        std::fs::write(volume.path().join("dir/a.txt"), "a").expect("seed");
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir"),
            "/handbook",
            volume.path(),
            "company-handbook",
            crate::runtime::vfs::Access::ReadOnly,
            None,
        );
        let source = r#"
            import { info, readText, writeText, append, writer, mkdir, remove, move, copy, list } from "submilli:fs";
            function denied(op: () => void): string {
                try { op(); } catch (e: PermissionDeniedError) { return e.capability; }
                return "allowed";
            }
            function main(): void {
                writeText("/local.txt", "l");
                assert(readText("/handbook/dir/a.txt") === "a", "reads work");
                let count = 0;
                for (const entry of list("/handbook", true)) { count++; }
                assert(count === 2, "listing works");
                assert(denied(() => writeText("/handbook/b.txt", "b")) === "fs.write", "write");
                assert(denied(() => append("/handbook/dir/a.txt", new Uint8Array(1))) === "fs.write", "append");
                assert(denied(() => writer("/handbook/c.txt")) === "fs.write", "writer");
                assert(denied(() => mkdir("/handbook/new", false)) === "fs.mkdir", "mkdir");
                mkdir("/handbook/dir", true);
                assert(denied(() => remove("/handbook/dir/a.txt", false)) === "fs.remove", "remove");
                assert(denied(() => move("/handbook/dir/a.txt", "/moved.txt")) === "fs.move", "move out");
                assert(denied(() => move("/local.txt", "/handbook/l.txt")) === "fs.move", "move in");
                assert(denied(() => copy("/local.txt", "/handbook/l.txt", false)) === "fs.copy", "copy in");
                copy("/handbook/dir/a.txt", "/copied.txt", false);
                assert(readText("/copied.txt") === "a", "copy out works");
                assert(info().mounts[0].access === "read_only", "info reports read-only");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("read-only asserts hold");
        assert!(!volume.path().join("b.txt").exists());
        assert!(volume.path().join("dir/a.txt").exists());
    }

    #[tokio::test]
    async fn mount_policy_denial_comes_before_the_read_only_refusal() {
        let volume = tempfile::tempdir().expect("volume");
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir"),
            "/ro",
            volume.path(),
            "ro",
            crate::runtime::vfs::Access::ReadOnly,
            None,
        );
        let source = r#"
            import { writeText } from "submilli:fs";
            function main(): string {
                try { writeText("/ro/x.txt", "hi"); } catch (e: PermissionDeniedError) { return e.reason; }
                return "not denied";
            }
        "#;
        let mut data = StoreData::with_vfs("test-run".into(), vfs);
        data.security_check = Arc::new(DenyAllFs);
        let value = run_with(source, data).await.expect("caught");
        assert_eq!(value.as_deref(), Some("denied fs.write in test"));
    }

    #[tokio::test]
    async fn mount_read_only_root_refuses_writes_and_needs_existing_mount_points() {
        let root = tempfile::tempdir().expect("root");
        let volume = tempfile::tempdir().expect("volume");
        let read_only = || {
            Vfs::external_with_mode(root.path().to_path_buf(), VfsMode::Named)
                .expect("root")
                .with_access(crate::runtime::vfs::Access::ReadOnly)
        };
        let missing = read_only().with_mount(crate::runtime::vfs::MountSpec {
            guest_path: "/rw".to_string(),
            host: volume.path().to_path_buf(),
            volume: "rw".to_string(),
            access: crate::runtime::vfs::Access::ReadWrite,
            quota: None,
        });
        assert!(
            matches!(
                missing,
                Err(crate::runtime::vfs::MountError::MountPointUnavailable { .. })
            ),
            "a read-only root is never written, so the mount point must exist"
        );
        assert!(!root.path().join("rw").exists());
        std::fs::create_dir(root.path().join("rw")).expect("mount point");
        let vfs = mount_volume(
            read_only(),
            "/rw",
            volume.path(),
            "rw",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        let source = r#"
            import { writeText, info } from "submilli:fs";
            function main(): void {
                let denied = false;
                try { writeText("/x.txt", "x"); } catch (e: PermissionDeniedError) { denied = true; }
                assert(denied, "the root is read-only");
                writeText("/rw/x.txt", "x");
                assert(info().access === "read_only", "root access surfaced");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("read-only root asserts hold");
        assert!(volume.path().join("x.txt").exists());
    }

    #[tokio::test]
    async fn mount_points_cannot_be_removed_moved_or_copied_through() {
        let volume = tempfile::tempdir().expect("volume");
        std::fs::write(volume.path().join("v.txt"), "v").expect("seed");
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir"),
            "/a/m",
            volume.path(),
            "m",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        let source = r#"
            import { remove, move, copy, writeText, mkdir, exists } from "submilli:fs";
            function fails(op: () => void): boolean {
                try { op(); return false; } catch (e) { return true; }
            }
            function main(): void {
                writeText("/a/sibling.txt", "s");
                mkdir("/a", true);
                assert(fails(() => remove("/a/m", true)), "remove the mount point");
                assert(fails(() => remove("/a", true)), "remove an ancestor");
                assert(fails(() => move("/a/m", "/b")), "move the mount point");
                assert(fails(() => move("/a", "/b")), "move an ancestor");
                assert(fails(() => move("/a/sibling.txt", "/a/m")), "replace the mount point");
                assert(fails(() => copy("/a", "/b", true)), "copy an ancestor");
                assert(fails(() => copy("/", "/a/m/all", true)), "copy the root into a mount");
                assert(exists("/a/m/v.txt"), "the volume is intact");
                remove("/a/sibling.txt", false);
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("mount-point asserts hold");
        assert!(volume.path().join("v.txt").exists());
    }

    #[tokio::test]
    async fn mount_shared_volume_quota_spans_every_vfs_that_mounts_it() {
        let volume = tempfile::tempdir().expect("volume");
        let quota = Arc::new(crate::runtime::DiskQuota::new(100, 0));
        let mount = |vfs: Vfs, guest: &str| {
            mount_volume(
                vfs,
                guest,
                volume.path(),
                "shared",
                crate::runtime::vfs::Access::ReadWrite,
                Some(Arc::clone(&quota)),
            )
        };
        let first = mount(Vfs::tempdir().expect("tempdir"), "/one");
        let second = mount(Vfs::tempdir().expect("tempdir"), "/two");
        let write = r#"
            import { writeText } from "submilli:fs";
            function main(): void { writeText("/one/a.txt", "x".repeat(60)); }
        "#;
        run_with(write, StoreData::with_vfs("test-run".into(), first))
            .await
            .expect("first write fits");
        let refuse = r#"
            import { writeText, readText } from "submilli:fs";
            function main(): void {
                assert(readText("/two/a.txt")!.length === 60, "the other alias sees it");
                let refused = false;
                try { writeText("/two/b.txt", "y".repeat(60)); } catch (e) { refused = e instanceof QuotaExceededError && e instanceof Error && !((e as unknown) instanceof RangeError); }
                assert(refused, "the shared limit refuses the second write");
            }
        "#;
        run_with(refuse, StoreData::with_vfs("test-run".into(), second))
            .await
            .expect("second alias shares the limit");
        assert_eq!(quota.used(), 60);
    }

    #[tokio::test]
    async fn mount_move_between_volumes_copies_then_removes_and_moves_the_charge() {
        let volume = tempfile::tempdir().expect("volume");
        let mount_quota = Arc::new(crate::runtime::DiskQuota::new(1000, 0));
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir").with_size_limit(1000),
            "/memory",
            volume.path(),
            "memory",
            crate::runtime::vfs::Access::ReadWrite,
            Some(Arc::clone(&mount_quota)),
        );
        let root_quota = vfs.quota().cloned().expect("root quota");
        let source = r#"
            import { writeText, readText, move, mkdir, exists } from "submilli:fs";
            function main(): void {
                writeText("/a.txt", "x".repeat(10));
                mkdir("/tree/sub", true);
                writeText("/tree/sub/b.txt", "y".repeat(20));
                move("/a.txt", "/memory/a.txt");
                move("/tree", "/memory/tree");
                assert(!exists("/a.txt") && !exists("/tree"), "sources removed");
                assert(readText("/memory/a.txt")!.length === 10, "file moved");
                assert(readText("/memory/tree/sub/b.txt")!.length === 20, "tree moved");
                move("/memory/a.txt", "/back.txt");
                assert(readText("/back.txt")!.length === 10, "moved back");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("cross-volume move asserts hold");
        assert_eq!(root_quota.used(), 10, "the root holds only back.txt");
        assert_eq!(mount_quota.used(), 20, "the volume holds the tree");
        let leftovers: Vec<_> = std::fs::read_dir(volume.path())
            .expect("volume")
            .map(|entry| entry.expect("entry").file_name())
            .collect();
        assert_eq!(
            leftovers,
            vec![std::ffi::OsString::from("tree")],
            "no staging left"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn mount_link_to_an_ancestor_cannot_reach_a_mount_point() {
        let session = tempfile::tempdir().expect("session");
        let volume = tempfile::tempdir().expect("volume");
        std::fs::write(volume.path().join("v.txt"), "v").expect("seed");
        let vfs = mount_volume(
            Vfs::external_with_mode(session.path().to_path_buf(), VfsMode::PerSession)
                .expect("root"),
            "/a/m",
            volume.path(),
            "m",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        std::os::unix::fs::symlink("a", session.path().join("lnk")).expect("link");
        std::fs::create_dir(session.path().join("r")).expect("dir");
        std::os::unix::fs::symlink("..", session.path().join("r/up")).expect("link");
        std::os::unix::fs::symlink("a/m/x", session.path().join("dangling")).expect("link");
        std::os::unix::fs::symlink(".", session.path().join("here")).expect("link");
        let source = r#"
            import { remove, move, copy, writeText, readText, mkdir, append } from "submilli:fs";
            function refused(op: () => void): boolean {
                try { op(); return false; } catch (e) { return String(e).includes("mount point"); }
            }
            function main(): void {
                assert(refused(() => remove("/lnk/m", false)), "remove the placeholder through a link");
                assert(refused(() => remove("/lnk/m", true)), "recursive remove through a link");
                assert(refused(() => writeText("/lnk/m", "x")), "replace the placeholder through a link");
                assert(refused(() => writeText("/lnk/m/x.txt", "x")), "write inside the placeholder");
                assert(refused(() => move("/r/up/a", "/z")), "move an ancestor through a link");
                assert(refused(() => remove("/r/up/a", true)), "remove an ancestor through a link");
                assert(refused(() => copy("/r/up/a", "/copy", true)), "copy an ancestor through a link");
                copy("/lnk", "/linkcopy", false);
                assert(refused(() => mkdir("/lnk/m/new", false)), "mkdir inside the placeholder");
                assert(refused(() => append("/dangling", new Uint8Array(1))), "append through a dangling link");
                writeText("/plain.txt", "p");
                assert(refused(() => copy("/plain.txt", "/dangling", false)), "copy through a dangling link");
                writeText("/here", "replaced");
                assert(readText("/here") === "replaced", "a write replaces a link to the root rather than refusing");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("link asserts hold");
        assert!(
            session.path().join("a/m").is_dir(),
            "the placeholder survives"
        );
        assert_eq!(
            std::fs::read_dir(session.path().join("a/m"))
                .expect("placeholder")
                .count(),
            0,
            "nothing landed in the placeholder"
        );
        assert!(volume.path().join("v.txt").exists());
        assert!(!session.path().join("copy").exists());
        assert!(
            std::fs::symlink_metadata(session.path().join("linkcopy"))
                .expect("copied")
                .file_type()
                .is_symlink(),
            "copying the link copies the link itself, never the tree behind it"
        );
    }

    #[tokio::test]
    async fn mount_move_too_deep_to_remove_is_refused_before_copying() {
        let volume = tempfile::tempdir().expect("volume");
        let quota = Arc::new(crate::runtime::DiskQuota::new(1 << 20, 0));
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir"),
            "/memory",
            volume.path(),
            "memory",
            crate::runtime::vfs::Access::ReadWrite,
            Some(Arc::clone(&quota)),
        );
        let source = r#"
            import { mkdir, writeText, move, exists } from "submilli:fs";
            function main(): void {
                const deep = "/deep" + "/d".repeat(70);
                mkdir(deep, true);
                writeText(deep + "/f.txt", "x");
                let failed = false;
                try { move("/deep", "/memory/deep"); } catch (e) { failed = true; }
                assert(failed, "the move is refused");
                assert(exists(deep + "/f.txt"), "the source is untouched");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("deep move asserts hold");
        assert_eq!(
            std::fs::read_dir(volume.path()).expect("volume").count(),
            0,
            "no staging copy is left behind"
        );
        assert_eq!(quota.used(), 0, "nothing stays charged");
    }

    #[cfg(any(target_os = "macos", windows))]
    #[test]
    fn mount_case_variant_directory_cannot_stand_in_as_a_mount_point() {
        use crate::runtime::vfs::{Access, MountError, MountSpec};
        let session = tempfile::tempdir().expect("session");
        let volume = tempfile::tempdir().expect("volume");
        std::fs::create_dir(session.path().join("Memory")).expect("variant");
        let result = Vfs::external_with_mode(session.path().to_path_buf(), VfsMode::PerSession)
            .expect("root")
            .with_mount(MountSpec {
                guest_path: "/memory".to_string(),
                host: volume.path().to_path_buf(),
                volume: "memory".to_string(),
                access: Access::ReadWrite,
                quota: None,
            });
        assert!(
            matches!(result, Err(MountError::MountPointUnavailable { .. })),
            "a case variant is refused"
        );
    }

    #[tokio::test]
    async fn mount_move_between_volumes_onto_existing_entries() {
        let volume = tempfile::tempdir().expect("volume");
        let quota = Arc::new(crate::runtime::DiskQuota::new(1000, 0));
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir"),
            "/memory",
            volume.path(),
            "memory",
            crate::runtime::vfs::Access::ReadWrite,
            Some(Arc::clone(&quota)),
        );
        let source = r#"
            import { writeText, readText, move, mkdir, exists } from "submilli:fs";
            function failure(op: () => void): string {
                try { op(); return "ok"; } catch (e) { return String(e); }
            }
            function main(): void {
                writeText("/memory/old.txt", "o".repeat(30));
                writeText("/new.txt", "n".repeat(10));
                move("/new.txt", "/memory/old.txt");
                assert(readText("/memory/old.txt")!.length === 10, "the file is replaced");
                mkdir("/dir", false);
                writeText("/file.txt", "f");
                assert(failure(() => move("/file.txt", "/memory")).includes("mount point"), "onto the mount point");
                mkdir("/memory/full", false);
                writeText("/memory/full/x.txt", "x");
                assert(failure(() => move("/dir", "/memory/full")).includes("not empty"), "onto a non-empty directory");
                assert(failure(() => move("/file.txt", "/memory/full")).includes("onto a directory"), "a file onto a directory");
                assert(exists("/dir") && exists("/file.txt"), "the sources are untouched");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("existing-destination asserts hold");
        assert_eq!(
            quota.used(),
            11,
            "the replaced file is released; only new files count"
        );
        let staged: Vec<_> = std::fs::read_dir(volume.path())
            .expect("volume")
            .map(|entry| entry.expect("entry").file_name())
            .filter(|name| name.to_string_lossy().ends_with(".tmp"))
            .collect();
        assert!(staged.is_empty(), "{staged:?}");
    }

    #[cfg(target_os = "macos")]
    #[tokio::test]
    async fn mount_non_ascii_alias_of_a_mount_point_is_refused() {
        let session = tempfile::tempdir().expect("session");
        let volume = tempfile::tempdir().expect("volume");
        let vfs = mount_volume(
            Vfs::external_with_mode(session.path().to_path_buf(), VfsMode::PerSession)
                .expect("root"),
            "/skills",
            volume.path(),
            "skills",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        // U+017F (long s) folds to `s` on a case-insensitive APFS volume.
        let source = r#"
            import { writeText, remove, move } from "submilli:fs";
            function refused(op: () => void): boolean {
                try { op(); return false; } catch (e) { return String(e).includes("mount point"); }
            }
            function main(): void {
                assert(refused(() => writeText("/ſkills/x.txt", "x")), "write through the alias");
                assert(refused(() => remove("/ſkills", true)), "remove the mount point by its alias");
                assert(refused(() => move("/ſkills", "/elsewhere")), "move the mount point by its alias");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("alias asserts hold");
        assert_eq!(
            std::fs::read_dir(session.path().join("skills"))
                .expect("placeholder")
                .count(),
            0
        );
    }

    #[cfg(target_os = "macos")]
    #[tokio::test]
    async fn mount_nested_non_ascii_aliases_of_a_mount_path_are_refused() {
        let session = tempfile::tempdir().expect("session");
        let volume = tempfile::tempdir().expect("volume");
        let vfs = mount_volume(
            Vfs::external_with_mode(session.path().to_path_buf(), VfsMode::PerSession)
                .expect("root"),
            "/sets/skills",
            volume.path(),
            "skills",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        // U+017F (long s) folds to `s` on a case-insensitive APFS volume, in the
        // parent as well as in the mount point.
        let source = r#"
            import { writeText, remove, move } from "submilli:fs";
            function refused(op: () => void): boolean {
                try { op(); return false; } catch (e) { return String(e).includes("mount point"); }
            }
            function main(): void {
                assert(refused(() => writeText("/\u017Fets/\u017Fkills/x.txt", "x")), "both components aliased");
                assert(refused(() => writeText("/Sets/\u017Fkills/x.txt", "x")), "case and Unicode aliases");
                assert(refused(() => remove("/\u017Fets", true)), "remove an aliased ancestor");
                assert(refused(() => move("/\u017Fets", "/elsewhere")), "move an aliased ancestor");
                writeText("/\u017Fets/beside.txt", "beside");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("nested alias asserts hold");
        assert_eq!(
            std::fs::read_dir(session.path().join("sets/skills"))
                .expect("placeholder")
                .count(),
            0
        );
        assert!(
            session.path().join("sets/beside.txt").exists(),
            "siblings stay writable"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn mount_root_link_loops_and_absolute_links_fail_cleanly() {
        let session = tempfile::tempdir().expect("session");
        let volume = tempfile::tempdir().expect("volume");
        let vfs = mount_volume(
            Vfs::external_with_mode(session.path().to_path_buf(), VfsMode::PerSession)
                .expect("root"),
            "/memory",
            volume.path(),
            "memory",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        std::os::unix::fs::symlink("loop", session.path().join("loop")).expect("link");
        std::os::unix::fs::symlink("/etc", session.path().join("abs")).expect("link");
        let source = r#"
            import { writeText, append } from "submilli:fs";
            function message(op: () => void): string {
                try { op(); return "ok"; } catch (e) { return String(e); }
            }
            function main(): void {
                assert(message(() => append("/loop/x", new Uint8Array(1))) !== "ok", "a loop fails");
                assert(message(() => append("/abs/x", new Uint8Array(1))).includes("escapes"), "an absolute link escapes");
                writeText("/fine.txt", "fine");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("loop asserts hold");
    }

    #[tokio::test]
    async fn mount_move_onto_a_mount_ancestor_is_refused_before_copying() {
        let volume = tempfile::tempdir().expect("volume");
        let other = tempfile::tempdir().expect("other");
        let quota = Arc::new(crate::runtime::DiskQuota::new(1 << 20, 0));
        let root = tempfile::tempdir().expect("root");
        let vfs = mount_volume(
            mount_volume(
                Vfs::external_with_mode(root.path().to_path_buf(), VfsMode::PerSession)
                    .expect("root")
                    .with_size_limit(1 << 20),
                "/memory",
                volume.path(),
                "memory",
                crate::runtime::vfs::Access::ReadWrite,
                None,
            ),
            "/sub/inner",
            other.path(),
            "inner",
            crate::runtime::vfs::Access::ReadWrite,
            Some(Arc::clone(&quota)),
        );
        let source = r#"
            import { writeText, move, exists } from "submilli:fs";
            function main(): void {
                writeText("/memory/tree.txt", "t");
                let failed = false;
                try { move("/memory/tree.txt", "/sub"); } catch (e) { failed = String(e).includes("mount point"); }
                assert(failed, "the move onto a mount's ancestor is refused");
                assert(exists("/memory/tree.txt"), "the source is untouched");
            }
        "#;
        run_with(source, StoreData::with_vfs("test-run".into(), vfs))
            .await
            .expect("ancestor move asserts hold");
        assert_eq!(quota.used(), 0);
        let staged: Vec<_> = std::fs::read_dir(root.path())
            .expect("root")
            .map(|entry| entry.expect("entry").file_name())
            .filter(|name| name.to_string_lossy().ends_with(".tmp"))
            .collect();
        assert!(
            staged.is_empty(),
            "nothing was staged in the root: {staged:?}"
        );
    }

    /// Mounting reads an identity for every directory on the mount path and fails
    /// if it can't, so this passing means the host supplies them.
    #[test]
    fn mount_reads_an_identity_for_each_directory_on_its_path() {
        let volume = tempfile::tempdir().expect("volume");
        let vfs = mount_volume(
            Vfs::tempdir().expect("tempdir"),
            "/data/memory",
            volume.path(),
            "memory",
            crate::runtime::vfs::Access::ReadWrite,
            None,
        );
        let names: Vec<_> = vfs.mounts()[0]
            .placeholders()
            .iter()
            .map(|placeholder| placeholder.name.to_string_lossy().into_owned())
            .collect();
        assert_eq!(
            names,
            ["data", "memory"],
            "every directory on the mount path got a placeholder"
        );
    }

    #[test]
    fn mount_paths_are_plain_ascii() {
        use crate::runtime::vfs::{Access, MountError, MountSpec};
        let volume = tempfile::tempdir().expect("volume");
        for bad in ["/mémoire", "/a b", "/a\\b", "/memory.", "/..."] {
            let result = Vfs::tempdir().expect("tempdir").with_mount(MountSpec {
                guest_path: bad.to_string(),
                host: volume.path().to_path_buf(),
                volume: "v".to_string(),
                access: Access::ReadWrite,
                quota: None,
            });
            assert!(matches!(result, Err(MountError::BadPath(_))), "{bad}");
        }
    }

    #[test]
    fn mounts_reject_ambiguous_tables() {
        use crate::runtime::vfs::{Access, MountError, MountSpec};
        let volume = tempfile::tempdir().expect("volume");
        let spec = |guest: &str, name: &str| MountSpec {
            guest_path: guest.to_string(),
            host: volume.path().to_path_buf(),
            volume: name.to_string(),
            access: Access::ReadWrite,
            quota: None,
        };
        assert!(matches!(
            Vfs::none().with_mount(spec("/a", "a")),
            Err(MountError::RootDisabled)
        ));
        let base = || Vfs::tempdir().expect("tempdir");
        for bad in ["a", "/a/", "/a//b", "/a/./b", "/a/../b", ""] {
            assert!(
                matches!(
                    base().with_mount(spec(bad, "a")),
                    Err(MountError::BadPath(_))
                ),
                "{bad:?} is not a normalized absolute path"
            );
        }
        assert!(matches!(
            base().with_mount(spec("/", "a")),
            Err(MountError::AtRoot)
        ));
        assert!(matches!(
            base().with_mount(spec("/x/.git", "a")),
            Err(MountError::ProtectedPath(_))
        ));
        let one = base().with_mount(spec("/a", "a")).expect("first");
        assert!(matches!(
            one.clone().with_mount(spec("/a/b", "b")),
            Err(MountError::Nested { .. })
        ));
        assert!(one.clone().with_mount(spec("/c", "a")).is_ok());
        let root = one.root().to_path_buf();
        std::fs::write(root.join("file"), "f").expect("file");
        assert!(matches!(
            one.clone().with_mount(spec("/file/x", "c")),
            Err(MountError::MountPointUnavailable { .. })
        ));
        let mut many = base();
        for index in 0..crate::runtime::vfs::MAX_MOUNTS {
            many = many
                .with_mount(spec(&format!("/m{index}"), &format!("v{index}")))
                .expect("within the cap");
        }
        assert!(matches!(
            many.with_mount(spec("/extra", "extra")),
            Err(MountError::TooMany)
        ));
    }
}