sausage/host/src/harness/mod.rs

//! Durable V8 harness, host-side SQLite storage, and world tool operations.
pub mod config;
mod credentials;
mod environment;
mod http;
pub mod input;
pub(crate) mod loader;
use crate::{exec::Exec, world::World};
use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
use deno_error::JsErrorBox;
use rusqlite::{Connection, types::Value as SqlValue};
use serde_json::{Value, json};
use std::path::Path;
use std::sync::Arc;

#[op2]
#[serde]
fn op_sql(
    state: &mut OpState,
    #[string] kind: String,
    #[string] sql: String,
    #[serde] params: Vec<serde_json::Value>,
) -> Result<serde_json::Value, JsErrorBox> {
    let work = || -> Result<Value, Box<dyn std::error::Error>> {
        if kind == "open" {
            let db = Connection::open(sql)?;
            db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
            state.put(db);
            return Ok(Value::Null);
        }
        if kind == "close" {
            state.take::<Connection>();
            return Ok(Value::Null);
        }
        let db = state.borrow::<Connection>();
        if kind == "exec" {
            db.execute_batch(&sql)?;
            return Ok(Value::Null);
        }
        let params: Vec<SqlValue> = params
            .into_iter()
            .map(|v| match v {
                Value::Null => SqlValue::Null,
                Value::Number(n) => n
                    .as_i64()
                    .map(SqlValue::Integer)
                    .unwrap_or_else(|| SqlValue::Real(n.as_f64().unwrap())),
                Value::String(s) => SqlValue::Text(s),
                Value::Array(a) => {
                    SqlValue::Blob(a.into_iter().map(|n| n.as_u64().unwrap() as u8).collect())
                }
                _ => panic!("unsupported SQL parameter"),
            })
            .collect();
        let mut stmt = db.prepare_cached(&sql)?;
        if kind == "run" {
            let changes = stmt.execute(rusqlite::params_from_iter(params))?;
            return Ok(json!({"changes": changes, "lastInsertRowid": db.last_insert_rowid()}));
        }
        let names: Vec<String> = stmt.column_names().into_iter().map(str::to_owned).collect();
        let mut rows = stmt.query(rusqlite::params_from_iter(params))?;
        let mut result = vec![];
        while let Some(row) = rows.next()? {
            let mut obj = serde_json::Map::new();
            for (i, name) in names.iter().enumerate() {
                let v = match row.get::<_, SqlValue>(i)? {
                    SqlValue::Null => Value::Null,
                    SqlValue::Integer(n) => json!(n),
                    SqlValue::Real(n) => json!(n),
                    SqlValue::Text(s) => json!(s),
                    SqlValue::Blob(b) => json!(b),
                };
                obj.insert(name.clone(), v);
            }
            result.push(Value::Object(obj));
        }
        Ok(json!(result))
    };
    work().map_err(|e| JsErrorBox::generic(e.to_string()))
}

deno_core::extension!(
    storage,
    ops = [
        op_sql,
        credentials::op_credential_requests,
        credentials::op_model_providers,
        config::op_model_catalog,
        credentials::op_credential_next,
        environment::op_file,
        environment::op_exec,
        environment::op_exec_next,
        input::op_input,
        http::op_harness_reply,
        http::op_harness_closed,
        environment::op_watch,
        environment::op_watch_next,
        loader::op_modules,
        loader::op_extension_status,
        environment::op_output
    ]
);

/// Start the console with the process implementation supplied by the host.
pub async fn run_with_exec(world: World, exec: Arc<dyn Exec>) -> anyhow::Result<()> {
    let cred = Arc::new(crate::cred::Cred::open(&world.host_dir)?);
    run_with_shared_credentials(world, exec, cred).await
}
async fn run_with_shared_credentials(
    world: World,
    exec: Arc<dyn Exec>,
    cred: Arc<crate::cred::Cred>,
) -> anyhow::Result<()> {
    // Deno's unsynchronized operations require a current-thread Tokio runtime.
    tokio::task::spawn_blocking(move || {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?
            .block_on(run_console(world, exec, cred))
    })
    .await?
}

/// Evaluate a bundled harness module with the same storage and tool operations as the console.
/// The caller must use a current-thread Tokio runtime.
pub async fn run_module(world: World, exec: Arc<dyn Exec>, source: &str) -> anyhow::Result<()> {
    let cred = Arc::new(crate::cred::store::Cred::open(&world.host_dir)?);
    run_module_with_credentials(world, exec, cred, source).await
}

/// Run a harness with the credentials handle shared by the host API.
pub async fn run_module_with_credentials(
    world: World,
    exec: Arc<dyn Exec>,
    cred: Arc<crate::cred::store::Cred>,
    source: &str,
) -> anyhow::Result<()> {
    let (_, receiver) = input::channel();
    run_module_with_input(world, exec, cred, receiver, source).await
}

/// Run without the console. Submissions and stop requests arrive through the channel.
/// The caller must use a current-thread Tokio runtime.
pub async fn run_module_with_input(
    world: World,
    exec: Arc<dyn Exec>,
    cred: Arc<crate::cred::store::Cred>,
    receiver: input::Receiver,
    source: &str,
) -> anyhow::Result<()> {
    let config_root = world.root.clone();
    let provider_config = config::Config::read(&config_root).await?;
    let loader = std::rc::Rc::new(loader::Loader {
        root: world.root.clone(),
        modules: Default::default(),
    });
    let mut runtime = JsRuntime::new(RuntimeOptions {
        module_loader: Some(loader.clone()),
        extensions: vec![
            deno_telemetry::deno_telemetry::init(),
            deno_webidl::deno_webidl::init(),
            deno_web::deno_web::init(
                Arc::new(deno_web::BlobStore::default()),
                None,
                false,
                Default::default(),
            ),
            deno_crypto::deno_crypto::init(None),
            deno_net::deno_net::init(None, None),
            deno_fetch::deno_fetch::init(Default::default()),
            storage::init(),
        ],
        extension_transpiler: Some(std::rc::Rc::new(|name, source| {
            Ok((loader::transpile(&name, source.to_string())?.into(), None))
        })),
        ..Default::default()
    });
    runtime
        .op_state()
        .borrow_mut()
        .put(config::CatalogCache(world.host_dir.join("models-dev.json")));
    runtime.op_state().borrow_mut().put(loader);
    runtime
        .op_state()
        .borrow_mut()
        .put(http::Replies::default());
    runtime
        .op_state()
        .borrow_mut()
        .put(loader::status(&world.host_dir));
    runtime
        .op_state()
        .borrow_mut()
        .put(deno_permissions::PermissionsContainer::allow_all(Arc::new(
            deno_permissions::RuntimePermissionDescriptorParser::new(sys_traits::impls::RealSys),
        )));
    runtime.execute_script("bootstrap.js", include_str!("../../js/bootstrap.js"))?;
    runtime
        .op_state()
        .borrow_mut()
        .put(credentials::Notifications::new(&cred));
    runtime.op_state().borrow_mut().put(cred);
    runtime
        .op_state()
        .borrow_mut()
        .put(std::rc::Rc::new(input::Inbox(tokio::sync::Mutex::new(
            receiver,
        ))));
    runtime
        .op_state()
        .borrow_mut()
        .put(environment::Environment {
            root: world.root,
            exec,
        });
    runtime.execute_script(
        "config.js",
        format!(
            "globalThis.storagePath = {};",
            serde_json::to_string(&world.host_dir.join("harness.sqlite"))?
        ),
    )?;
    let specifier = deno_core::ModuleSpecifier::parse("file:///harness.js")?;
    let module = runtime
        .load_main_es_module_from_code(&specifier, source.to_owned())
        .await?;
    let result = runtime.mod_evaluate(module);
    let result = async {
        runtime
            .with_event_loop_promise(result, Default::default())
            .await?;
        runtime.run_event_loop(Default::default()).await?;
        Ok::<_, anyhow::Error>(())
    }
    .await;
    let current_config = config::Config::read(&config_root).await.unwrap_or_default();
    result.map_err(|error| {
        let mut text = error.to_string();
        for provider in provider_config
            .providers
            .values()
            .chain(current_config.providers.values())
        {
            if !provider.api_key.is_empty() {
                text = text.replace(&provider.api_key, "[redacted]");
            }
        }
        anyhow::anyhow!(text)
    })
}

/// Run the headless harness with console input and signal clients.
/// The caller must use a current-thread Tokio runtime.
pub async fn run_console(
    world: World,
    exec: Arc<dyn Exec>,
    cred: Arc<crate::cred::store::Cred>,
) -> anyhow::Result<()> {
    let (sender, receiver) = input::channel();
    run_console_with_input(world, exec, cred, sender, receiver).await
}

/// Run the console with the inbox shared by the host HTTP API.
pub async fn run_console_with_input(
    world: World,
    exec: Arc<dyn Exec>,
    cred: Arc<crate::cred::store::Cred>,
    sender: input::Sender,
    receiver: input::Receiver,
) -> anyhow::Result<()> {
    input::console(sender.clone());
    let signals = tokio::spawn(input::stop_signal(sender)?);
    let source = concat!(
        "globalThis.consoleEnabled = true;\n",
        include_str!("../../js/bundle.js")
    );
    let result = run_module_with_input(world, exec, cred, receiver, source).await;
    signals.abort();
    result
}

/// Open a world and resume its durable root conversation.
pub async fn run(path: &Path) -> anyhow::Result<()> {
    run_with_userland(path, None).await
}

/// Start the console with the shared process table and an optional userland bundle.
pub async fn run_with_userland(path: &Path, userland: Option<&Path>) -> anyhow::Result<()> {
    let world = World::open_with_userland(path, userland)?;
    let cred = Arc::new(crate::cred::Cred::open(&world.host_dir)?);
    let runtime = crate::process::Runtime::new_with_hooks(
        World {
            root: world.root.clone(),
            host_dir: world.host_dir.clone(),
            processes: world.processes.clone(),
        },
        cred.clone(),
        crate::process::default_script(),
        None,
    )?;
    let services = crate::process::supervise(runtime.clone());
    let result = run_with_shared_credentials(world, Arc::new(runtime), cred).await;
    services.shutdown().await;
    result
}