301 lines
10 KiB
Rust
301 lines
10 KiB
Rust
//! 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,
|
|
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(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
|
|
}
|