submilli-engine 0.3.0

//! Capability-controlled Git operations. No subprocesses or guest-visible credentials.
use crate::runtime::host::{abi_arg, abi_result};
pub(crate) mod class;
pub mod declaration;
mod diff;
mod history;
mod index_limits;
mod location;
mod lock;
pub(crate) mod log_cache;
mod metadata_scan;
mod meter;
mod object;
mod operations;
mod pack_index_check;
mod pack_limits;
mod stage;
mod storage;
mod transport;
mod work;
mod worker;

use crate::runtime::decision::{CallSite, EntryPath, SourceLine};
use crate::runtime::fuel;
use crate::runtime::host::{
    permission_denied, read_string_arg, register_host_fn_async, write_submilli_string_struct,
    write_submilli_uint8array_struct,
};
use crate::runtime::intrinsic_types::{IntrinsicTypes, build_intrinsic_types};
use crate::runtime::{HttpClient, SecretProvider, SecurityCheck, StoreData};
use crate::stdlib::{session::value, shared::running_package};
use serde_json::Value;
use std::sync::{
    Arc, Mutex, OnceLock,
    atomic::{AtomicBool, AtomicU64, Ordering},
};
use wasmtime::{Caller, FuncType, HeapType, Linker, RefType, Result, Val, ValType, bail};

static WORKERS: OnceLock<Arc<tokio::sync::Semaphore>> = OnceLock::new();

pub const MODULE_NAME: &str = "submilli:git";
pub use declaration::package_declaration;
pub(crate) use transport::canonical_url;

#[derive(Clone, Debug)]
pub struct GitConfig {
    pub name: String,
    pub email: String,
    pub username: Option<String>,
}

#[derive(Clone)]
struct Job {
    op: String,
    path: String,
    caller: String,
    config: GitConfig,
    security: Arc<dyn SecurityCheck>,
    secrets: Arc<dyn SecretProvider>,
    http: Arc<dyn HttpClient>,
    runtime: tokio::runtime::Handle,
    /// When the operation times out, waits included.
    deadline: tokio::time::Instant,
    cancelled: Arc<AtomicBool>,
    max_bytes: u64,
    transferred: Arc<AtomicU64>,
    /// The worker's own file and object work, charged when it returns.
    meter: Arc<meter::Meter>,
    algorithm_fuel: Arc<work::AlgorithmWork>,
    denial: Arc<Mutex<Option<(String, String)>>>,
    history_cache: Option<Arc<log_cache::Cache>>,
    /// The program line that started the operation, captured only while recording.
    line: Option<SourceLine>,
}
impl Job {
    fn remote_capability(&self) -> &'static str {
        if self.op == "clone" {
            "git.clone"
        } else {
            "git.fetch"
        }
    }

    fn check(&self, capability: &str, context: Value) -> Result<()> {
        self.check_cancelled()?;
        // Git's decisions are made on a worker thread, after the program's stack is gone;
        // they reuse the line the host call captured on entry.
        let site = CallSite::new(
            self.security
                .recorder()
                .map(|recorder| recorder.begin_call(&self.caller, capability, self.line)),
            EntryPath::Git,
        );
        crate::stdlib::shared::check_and_audit(
            self.security.as_ref(),
            &self.caller,
            capability,
            &context,
            "/",
            site,
        )
        .map_err(|reason| {
            // Poison may leave the denied capability and reason partly updated.
            // AGENTS.md permits poisoned-lock panics rather than recovering this
            // attribution; it does not permit the panic that caused poisoning.
            let mut denial = self.denial.lock().expect("git worker denial lock poisoned");
            *denial = Some((capability.to_owned(), reason.clone()));
            permission_denied(&self.caller, capability, reason)
        })
    }

    /// One of the workers Git shares across the process, waited for until
    /// the operation's deadline or its cancellation.
    fn worker_permit(&self) -> Result<tokio::sync::OwnedSemaphorePermit> {
        let workers = WORKERS
            .get_or_init(|| Arc::new(tokio::sync::Semaphore::new(4)))
            .clone();
        // Polled from the worker's own thread, which is not the runtime's.
        loop {
            match workers.clone().try_acquire_owned() {
                Ok(permit) => return Ok(permit),
                Err(tokio::sync::TryAcquireError::Closed) => {
                    return Err(crate::runtime::host::fatal_host_error(
                        "git: workers closed",
                    ));
                }
                Err(tokio::sync::TryAcquireError::NoPermits) => {}
            }
            self.check_cancelled()?;
            if tokio::time::Instant::now() >= self.deadline {
                bail!("git: operation timed out");
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
    }

    fn check_cancelled(&self) -> Result<()> {
        if self.cancelled.load(Ordering::Relaxed) {
            bail!("git: operation cancelled");
        }
        Ok(())
    }
}
struct CancelOnDrop(Arc<AtomicBool>);
impl Drop for CancelOnDrop {
    fn drop(&mut self) {
        self.0.store(true, Ordering::Relaxed);
    }
}

pub fn install(linker: &mut Linker<StoreData>) -> Result<()> {
    let engine = linker.engine().clone();
    let intr = build_intrinsic_types(&engine)?;
    install_repository_reads(linker, &engine, &intr)?;
    install_repository_writes(linker, &engine, &intr)
}

fn install_repository_reads(
    linker: &mut Linker<StoreData>,
    engine: &wasmtime::Engine,
    intr: &IntrinsicTypes,
) -> Result<()> {
    let argument = ValType::Ref(RefType::new(
        true,
        HeapType::ConcreteStruct(intr.object.clone()),
    ));
    let receiver = ValType::Ref(RefType::new(
        false,
        HeapType::ConcreteStruct(intr.object.clone()),
    ));
    let object = ValType::Ref(RefType::new(
        false,
        HeapType::ConcreteStruct(intr.object_shape.clone()),
    ));
    let bytes = ValType::Ref(RefType::new(
        false,
        HeapType::ConcreteStruct(intr.uint8_array.clone()),
    ));
    let array = ValType::Ref(RefType::new(
        false,
        HeapType::ConcreteStruct(intr.array.clone()),
    ));
    let repository = crate::mangle::package_symbol(MODULE_NAME, "Repository");
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "status"),
        FuncType::new(engine, [receiver.clone()], [object.clone()]),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "status", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "log"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone()],
            [object.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "log", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "diff"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone()],
            [object.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "diff", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "show"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone(), argument.clone()],
            [bytes.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "show", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "branches"),
        FuncType::new(engine, [receiver.clone()], [array.clone()]),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "branches", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "remotes"),
        FuncType::new(engine, [receiver.clone()], [array.clone()]),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "remotes", true, params).await?;
                Ok(())
            })
        },
    )?;
    Ok(())
}

fn install_repository_writes(
    linker: &mut Linker<StoreData>,
    engine: &wasmtime::Engine,
    intr: &IntrinsicTypes,
) -> Result<()> {
    let argument = ValType::Ref(RefType::new(
        true,
        HeapType::ConcreteStruct(intr.object.clone()),
    ));
    let receiver = ValType::Ref(RefType::new(
        false,
        HeapType::ConcreteStruct(intr.object.clone()),
    ));
    let object = ValType::Ref(RefType::new(
        false,
        HeapType::ConcreteStruct(intr.object_shape.clone()),
    ));
    // Class method slots carry a value even when their source return type is void.
    let completion = argument.clone();
    let repository = crate::mangle::package_symbol(MODULE_NAME, "Repository");
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "add"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone()],
            [completion.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                invoke(caller, "add", true, params).await?;
                *abi_result(results, 0)? = crate::runtime::prelude::undefined::value(caller)?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "commit"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone()],
            [ValType::Ref(RefType::new(
                false,
                intr.string.clone().into(),
            ))],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "commit", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "createBranch"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone(), argument.clone()],
            [completion.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                invoke(caller, "createBranch", true, params).await?;
                *abi_result(results, 0)? = crate::runtime::prelude::undefined::value(caller)?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "switchBranch"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone()],
            [completion.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                invoke(caller, "switchBranch", true, params).await?;
                *abi_result(results, 0)? = crate::runtime::prelude::undefined::value(caller)?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "addRemote"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone(), argument.clone()],
            [completion.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                invoke(caller, "addRemote", true, params).await?;
                *abi_result(results, 0)? = crate::runtime::prelude::undefined::value(caller)?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "setRemoteUrl"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone(), argument.clone()],
            [completion.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                invoke(caller, "setRemoteUrl", true, params).await?;
                *abi_result(results, 0)? = crate::runtime::prelude::undefined::value(caller)?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "fetch"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone(), argument.clone()],
            [object.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "fetch", true, params).await?;
                Ok(())
            })
        },
    )?;
    register_host_fn_async(
        linker,
        MODULE_NAME,
        crate::mangle::extend(&repository, "pull"),
        FuncType::new(
            engine,
            [receiver.clone(), argument.clone(), argument.clone()],
            [object.clone()],
        ),
        false,
        |caller, params, results| {
            Box::pin(async move {
                *abi_result(results, 0)? = invoke(caller, "pull", true, params).await?;
                Ok(())
            })
        },
    )?;
    Ok(())
}

async fn invoke(
    caller: &mut Caller<'_, StoreData>,
    op: &str,
    method: bool,
    params: &[Val],
) -> Result<Val> {
    let deadline = tokio::time::Instant::now()
        .checked_add(std::time::Duration::from_secs(60))
        .ok_or_else(|| crate::runtime::host::fatal_host_error("git deadline is out of range"))?;
    let config = caller.data().git.clone().ok_or_else(|| {
        wasmtime::Error::msg("submilli:git is disabled; configure the blueprint git block")
    })?;
    if op == "log" && caller.data().git_history.is_none() {
        let cache = log_cache::Cache::new(&caller.data().tenant_limits)?;
        caller.data_mut().git_history = Some(Arc::new(cache));
    }
    if matches!(
        op,
        "init" | "clone" | "add" | "commit" | "createBranch" | "switchBranch" | "fetch" | "pull"
    ) && let Some(cache) = &caller.data().git_history
    {
        cache.invalidate()?;
    }
    let budget = WorkingBudget::reserve(&caller.data().tenant_limits, op)?;
    let max_bytes = budget.max_bytes(op);
    let (args, path) = before_deadline(
        deadline,
        decode_arguments(caller, op, method, params, max_bytes),
    )
    .await??;
    let principal = running_package(&*caller)
        .map_err(|_| crate::runtime::host::fatal_host_error("git: cannot identify caller"))?;
    let cancelled = Arc::new(AtomicBool::new(false));
    let cancel_guard = CancelOnDrop(cancelled.clone());
    // The worker stops itself once its work passes what the run has left.
    let meter = Arc::new(meter::Meter::new(
        meter::remaining_fuel(&mut *caller)?,
        Some(cancelled.clone()),
    ));
    let job = Job {
        op: op.to_owned(),
        path: path.clone(),
        caller: principal,
        config,
        security: caller.data().security_check.clone(),
        secrets: caller.data().secret_provider.clone(),
        http: caller.data().http_client.clone(),
        runtime: tokio::runtime::Handle::try_current().map_err(|_| {
            crate::runtime::host::fatal_host_error("git worker requires a Tokio runtime")
        })?,
        deadline,
        cancelled,
        max_bytes,
        transferred: Arc::new(AtomicU64::new(0)),
        meter: Arc::clone(&meter),
        algorithm_fuel: Arc::new(work::AlgorithmWork::with_meter(Arc::clone(&meter))),
        denial: Arc::new(Mutex::new(None)),
        history_cache: caller.data().git_history.clone(),
        line: crate::stdlib::shared::source_line(&*caller),
    };
    let vfs = caller.data().vfs.clone();
    let op = op.to_owned();
    let is_constructor = !method;
    let returns_string = op == "commit";
    let denial = job.denial.clone();
    let principal = job.caller.clone();
    let worker = async {
        caller
            .data_mut()
            .blocking_work
            .spawn(move || {
                job.check_cancelled()?;
                worker::run(&vfs, &job, &op, &args).map(|result| (result, budget))
            })
            .await
            .map_err(blocking_worker_failure)?
    };
    let outcome = finish_worker(worker, &cancel_guard.0, deadline).await;
    // The work is done, and may be published, whether or not it succeeded:
    // it is settled, not refused.
    meter.settle(&mut *caller)?;
    if let Err(error) = &outcome
        && meter.is_exhausted()
        && !crate::runtime::host::ends_the_run(error)
        && !error.is::<stage::NeedsHostRecovery>()
    {
        // Stopped for fuel before publishing: the run ends as a Wasm loop
        // out of fuel would, and the program cannot catch it. A host failure,
        // or a publication that needs recovery, says more and goes first.
        return Err(wasmtime::Trap::OutOfFuel.into());
    }

    let outcome = match outcome {
        Err(error)
            if crate::runtime::host::ends_the_run(&error)
                || error.is::<stage::NeedsHostRecovery>() =>
        {
            return Err(error);
        }
        other => other,
    };
    // Poison may leave denial attribution partly updated by the worker. AGENTS.md
    // permits this poisoned-lock panic instead of recovery, including after worker
    // cleanup; it does not permit the panic that caused poisoning.
    let denial = denial.lock().expect("git: denial lock poisoned").take();
    drop(cancel_guard);
    fuel::settle_result(caller, |caller| {
        if let Some((capability, reason)) = denial {
            return Err(crate::runtime::host::throw_host_error(
                caller,
                permission_denied(&principal, &capability, reason),
            ));
        }
        let (result, _budget) =
            outcome.map_err(|error| crate::runtime::host::throw_host_error(caller, error))?;
        encode_result(
            caller,
            result,
            &path,
            is_constructor,
            returns_string,
            max_bytes,
        )
        .map_err(|error| crate::runtime::host::throw_host_error(caller, error))
    })
}

async fn before_deadline<T>(
    deadline: tokio::time::Instant,
    future: impl std::future::Future<Output = T>,
) -> Result<T> {
    if tokio::time::Instant::now() >= deadline {
        bail!("git: operation timed out");
    }
    tokio::time::timeout_at(deadline, future)
        .await
        .map_err(|_| wasmtime::Error::msg("git: operation timed out"))
}

async fn finish_worker<T>(
    worker: impl std::future::Future<Output = Result<T>>,
    cancelled: &AtomicBool,
    deadline: tokio::time::Instant,
) -> Result<T> {
    tokio::pin!(worker);
    match before_deadline(deadline, &mut worker).await {
        Ok(result) => result,
        Err(error) => {
            cancelled.store(true, Ordering::Relaxed);
            // A catchable timeout must not let guest code resume while the
            // worker is still publishing or rolling back repository changes.
            match worker.await {
                Err(worker_error)
                    if crate::runtime::host::ends_the_run(&worker_error)
                        || worker_error.is::<stage::NeedsHostRecovery>() =>
                {
                    Err(worker_error)
                }
                _ => Err(error),
            }
        }
    }
}

fn blocking_worker_failure(error: crate::runtime::blocking::BlockingWorkError) -> wasmtime::Error {
    crate::runtime::host::fatal_host_error("git blocking worker failed").context(error)
}

async fn decode_arguments(
    caller: &mut Caller<'_, StoreData>,
    op: &str,
    method: bool,
    params: &[Val],
    max_bytes: u64,
) -> Result<(Vec<Value>, String)> {
    let mut args = Vec::new();
    let offset = usize::from(method);
    for (index, param) in params
        .get(offset..)
        .ok_or_else(|| crate::runtime::host::invariant_trap("git ABI: missing arguments"))?
        .iter()
        .enumerate()
    {
        if crate::runtime::prelude::undefined::is_undefined(caller, param)? {
            args.push(undefined_argument(op, index)?);
            continue;
        }
        let units = value::serialize(caller, param).await?;
        if units.len() as u64 * 2 > max_bytes {
            bail!("git: argument size limit exceeded");
        }
        fuel::charge(&mut *caller, fuel::PARSE, units.len() as u64)?;
        let text = String::from_utf16(&units)
            .map_err(|_| wasmtime::Error::msg("git: arguments must be valid Unicode"))?;
        args.push(serde_json::from_str::<Value>(&text)?);
    }
    let path = if method {
        let field = class::path(caller, abi_arg(params, 0)?)?;
        read_string_arg(&mut *caller, &field, "git repository")?
    } else {
        operations::text_arg(&args, usize::from(op == "clone"))?.to_owned()
    };
    let path = crate::runtime::fs::guest_normalize(caller.data().vfs.cwd(), &path)?;
    if path
        .split('/')
        .any(|part| part.eq_ignore_ascii_case(".git") || stage::is_reserved_stage_name(part))
    {
        bail!("git: repository path targets protected metadata");
    }
    Ok((args, path))
}

// Virtual calls apply defaults in the selected method, just as guest classes do.
fn undefined_argument(op: &str, index: usize) -> Result<Value> {
    match (op, index) {
        ("init", 1) | ("clone", 2) | ("log" | "diff", 0) => {
            Ok(Value::Object(serde_json::Map::new()))
        }
        ("createBranch", 1) => Ok(Value::String("HEAD".into())),
        ("fetch" | "pull", 0) => Ok(Value::String("origin".into())),
        ("fetch" | "pull", 1) => Ok(Value::String(String::new())),
        _ => bail!("git: undefined is not accepted for this argument to {op}"),
    }
}

fn encode_result(
    caller: &mut Caller<'_, StoreData>,
    result: worker::Output,
    path: &str,
    is_constructor: bool,
    returns_string: bool,
    max_bytes: u64,
) -> Result<Val> {
    if is_constructor {
        let path = write_submilli_string_struct(caller, path)?;
        return Ok(Val::AnyRef(Some(path.to_anyref())));
    }
    let result = match result {
        worker::Output::Bytes(bytes) => {
            let value = write_submilli_uint8array_struct(caller, &bytes)?;
            return Ok(Val::AnyRef(Some(value.to_anyref())));
        }
        worker::Output::Json(value) => value,
    };
    if returns_string {
        let text = result
            .as_str()
            .ok_or_else(|| wasmtime::Error::msg("git: invalid string result"))?;
        let value = write_submilli_string_struct(caller, text)?;
        return Ok(Val::AnyRef(Some(value.to_anyref())));
    }
    let text = serde_json::to_string(&result)?;
    if text.len() > max_bytes as usize {
        bail!("git: result limit exceeded");
    }
    // Serialized and parsed back; the operation is already published.
    fuel::settle(&mut *caller, fuel::PARSE, 2 * text.len() as u64)?;
    value::deserialize(caller, &text.encode_utf16().collect::<Vec<_>>())
}

/// The least Git works with: what opening a repository and a fetch's index
/// buffers take, measured at under 1 MB, with room to spare.
const MIN_WORKING_BYTES: u64 = 4 * 1024 * 1024;

/// Working memory reserved for one Git operation against the same cap as
/// Wasm, while the program waits for it.
///
/// gix runs natively, outside the metered heap, so the operation reserves
/// three quarters of what the run has free, and bounds what it holds by
/// counting: trees, the index, worktree paths, a blob, a diff, a fetched
/// pack's delta chains. Each count is held to the reservation divided by
/// what the operation was measured to need per byte counted, beyond its
/// fixed working set (`tests/git_memory.rs`).
struct WorkingBudget {
    bytes: u64,
    counter: Arc<std::sync::atomic::AtomicU64>,
}
impl WorkingBudget {
    fn reserve(limits: &crate::runtime::limits::TenantLimits, op: &str) -> Result<Self> {
        let available = limits
            .max_total_bytes
            .saturating_sub(limits.observed_bytes())
            .saturating_sub(limits.host_attached_bytes());
        let bytes = available / 4 * 3;
        if bytes < MIN_WORKING_BYTES {
            const MB: u64 = 1024 * 1024;
            bail!(
                "git.{op} needs at least {} MB of working memory; {} MB is free under \
                 max_execution_memory ({} MB). Raise max_execution_memory",
                MIN_WORKING_BYTES * 4 / 3 / MB,
                available / MB,
                limits.max_total_bytes / MB
            );
        }
        limits.charge_host_bytes(bytes)?;
        Ok(Self {
            bytes,
            counter: limits.host_attached_counter(),
        })
    }

    /// What `op` may count while it works: the reservation divided by what
    /// it was measured to hold per byte counted. A diff holds its patch two
    /// or three times over as it is built and encoded; everything else
    /// holds at most about three times what it counts.
    fn max_bytes(&self, op: &str) -> u64 {
        let ratio = if op == "diff" { 8 } else { 4 };
        (self.bytes.saturating_sub(MIN_WORKING_BYTES / 4) / ratio).min(storage::MAX_WORKING_BYTES)
    }
}
impl Drop for WorkingBudget {
    fn drop(&mut self) {
        self.counter.fetch_sub(self.bytes, Ordering::Relaxed);
    }
}

#[cfg(test)]
mod tests;

#[cfg(test)]
mod integration_tests;