use murmur_artifact::{version_fit, FitSubject, VersionFit, WitContracts, WitLevel};
use wasmtime::component::{Component, Linker, ResourceTable};
use wasmtime::{Engine, Store};
use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView};
use crate::bindings::process_driver::ProcessDriver as ProcessDriverBindings;
use crate::errors::RuntimeError;
use crate::limits::{
classify_guest_failure, CallDeadlines, ComponentRole, EpochTicker, ExecutionLimiter,
GuestFailure, WallDeadline,
};
use crate::ExecutionLimits;
pub use crate::bindings::process_driver::exports::murmur::driver::process::{
Bridge, Description, DriverFile, Event, ExitStatus, FailureKind, HarnessNeeds, InterruptMethod,
LaunchPlan, LaunchRequest, LoginCheck, LoginOutput, LoginState, PathAccess, PathNeed,
QuotaState, QuotaWindow, RetryInfo, Session, SessionInfo, SessionMode, ToolCallInfo,
ToolCallProgress, ToolCallStart, ToolResultInfo, TurnFailure, Usage,
};
impl QuotaState {
pub fn wire_name(self) -> &'static str {
match self {
QuotaState::Ok => "ok",
QuotaState::Warning => "warning",
QuotaState::Exhausted => "exhausted",
}
}
}
pub const PROCESS_DRIVER_IFACE: &str = "murmur:driver/process@0.6.0";
const PROCESS_DRIVER_IFACE_PREFIX: &str = "murmur:driver/process@";
pub fn check_driver_interface(
transport: &str,
name: &str,
version: &str,
contracts: Option<&WitContracts>,
level: &WitLevel,
) -> Result<(), RuntimeError> {
let process_exports: Vec<String> = contracts
.map(|c| {
c.exports
.iter()
.filter(|export| export.starts_with(PROCESS_DRIVER_IFACE_PREFIX))
.cloned()
.collect()
})
.unwrap_or_default();
let expected = level_interface(level, "murmur:driver", "process")
.unwrap_or_else(|| PROCESS_DRIVER_IFACE.to_string());
match transport {
"process" if process_exports.contains(&expected) => Ok(()),
"process" if process_exports.is_empty() => {
Err(RuntimeError::ProcessDriverInterfaceMissing {
name: name.to_string(),
version: version.to_string(),
expected,
})
}
"process" => Err(
level_mismatch(name, version, level, process_exports, Vec::new()).unwrap_or_else(
|| RuntimeError::ProcessDriverInterfaceMissing {
name: name.to_string(),
version: version.to_string(),
expected,
},
),
),
"http" if !process_exports.is_empty() => {
Err(RuntimeError::HttpDriverExportsProcessInterface {
name: name.to_string(),
version: version.to_string(),
exported: process_exports,
})
}
_ => Ok(()),
}
}
const RETIRED_TEXT_IFACE_PREFIX: &str = "murmur:text/";
const STREAM_IFACE_PREFIX: &str = "murmur:stream/";
pub fn check_stream_interface(
name: &str,
version: &str,
contracts: Option<&WitContracts>,
level: &WitLevel,
) -> Result<(), RuntimeError> {
let expected = level_interface(level, "murmur:stream", "events")
.unwrap_or_else(|| crate::runtime::WIT_STREAM_EVENTS_IFACE.to_string());
let imported: Vec<String> = contracts
.map(|c| {
c.imports
.iter()
.filter(|import| {
import.starts_with(RETIRED_TEXT_IFACE_PREFIX)
|| (import.starts_with(STREAM_IFACE_PREFIX) && **import != expected)
})
.cloned()
.collect()
})
.unwrap_or_default();
if imported.is_empty() {
return Ok(());
}
if let Some(mismatch) = level_mismatch(name, version, level, Vec::new(), imported.clone()) {
return Err(mismatch);
}
Err(RuntimeError::DriverStreamInterfaceNotServed {
name: name.to_string(),
version: version.to_string(),
imported,
expected,
})
}
pub(crate) fn driver_refusal(
stream: Result<(), RuntimeError>,
driver: Result<(), RuntimeError>,
) -> Result<(), RuntimeError> {
match (stream, driver) {
(
Err(RuntimeError::WitLevelMismatch {
mismatched: imports,
..
}),
Err(RuntimeError::WitLevelMismatch {
name,
version,
level,
mismatched: mut exports,
}),
) => {
exports.extend(imports);
Err(RuntimeError::WitLevelMismatch {
name,
version,
level,
mismatched: exports,
})
}
(Err(_), Err(driver @ RuntimeError::WitLevelMismatch { .. })) => Err(driver),
(Err(stream), _) => Err(stream),
(Ok(()), driver) => driver,
}
}
fn level_interface(level: &WitLevel, package: &str, interface: &str) -> Option<String> {
level
.version_of(package)
.map(|version| format!("{package}/{interface}@{version}"))
}
pub(crate) fn check_component_level(
name: &str,
version: &str,
contracts: WitContracts,
level: &WitLevel,
) -> Result<(), RuntimeError> {
match level_mismatch(name, version, level, contracts.exports, contracts.imports) {
Some(mismatch) => Err(mismatch),
None => Ok(()),
}
}
fn level_mismatch(
name: &str,
version: &str,
level: &WitLevel,
exports: Vec<String>,
imports: Vec<String>,
) -> Option<RuntimeError> {
let subject = FitSubject::component(WitContracts { exports, imports });
match version_fit(&subject, level) {
VersionFit::Unserved(mismatched) => Some(RuntimeError::WitLevelMismatch {
name: name.to_string(),
version: version.to_string(),
level: level.level,
mismatched,
}),
VersionFit::Fits => None,
}
}
pub fn check_required_env(
name: &str,
version: &str,
description: &Description,
env_allow: &[String],
) -> Result<(), RuntimeError> {
let mut missing: Vec<String> = Vec::new();
for var in &description.required_env {
if !env_allow.contains(var) && !missing.contains(var) {
missing.push(var.clone());
}
}
if missing.is_empty() {
Ok(())
} else {
Err(RuntimeError::ProcessDriverRequiredEnvNotAllowed {
name: name.to_string(),
version: version.to_string(),
missing,
})
}
}
struct ProcessDriverStoreState {
wasi: WasiCtx,
table: ResourceTable,
limits: ExecutionLimiter,
}
impl WasiView for ProcessDriverStoreState {
fn ctx(&mut self) -> WasiCtxView<'_> {
WasiCtxView {
ctx: &mut self.wasi,
table: &mut self.table,
}
}
}
pub struct ProcessDriver {
name: String,
version: String,
body: DriverBody,
call_deadlines: CallDeadlines,
cut: Option<(&'static str, u64)>,
}
enum DriverBody {
Local {
store: Store<ProcessDriverStoreState>,
bindings: ProcessDriverBindings,
},
Remote(crate::guest_worker::RemoteProcessDriver),
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) enum DriverCall {
Describe,
Launch(LaunchRequest),
Parse(Vec<String>),
ClassifyExit(ExitStatus),
ReadLogin(LoginOutput),
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) enum DriverAnswer {
Opened,
Description(Description),
Launch(Result<LaunchPlan, String>),
Events(Vec<Event>),
Event(Event),
Login(LoginState),
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) enum DriverError {
Load(String),
CallFailed { call: String, message: String },
Other(String),
}
impl DriverError {
pub(crate) fn from_runtime(error: RuntimeError) -> Self {
match error {
RuntimeError::ProcessDriverLoad { message, .. } => Self::Load(message),
RuntimeError::ProcessDriverCallFailed { call, message, .. } => {
Self::CallFailed { call, message }
}
other => Self::Other(other.to_string()),
}
}
pub(crate) fn into_runtime(self, name: &str, version: &str) -> RuntimeError {
match self {
Self::Load(message) => RuntimeError::ProcessDriverLoad {
name: name.to_string(),
version: version.to_string(),
message,
},
Self::CallFailed { call, message } => RuntimeError::ProcessDriverCallFailed {
name: name.to_string(),
version: version.to_string(),
call,
message,
},
Self::Other(message) => RuntimeError::Runtime(message),
}
}
}
impl ProcessDriver {
pub(crate) async fn open(
workers: Option<&std::sync::Arc<crate::guest_worker::GuestWorkers>>,
engine: &Engine,
component: &Component,
name: &str,
version: &str,
call_deadlines: CallDeadlines,
) -> Result<Self, RuntimeError> {
let Some(workers) = workers else {
return Self::instantiate(engine, component, name, version, call_deadlines).await;
};
let remote = crate::guest_worker::RemoteProcessDriver::open(
workers,
component,
name,
version,
&call_deadlines,
)
.await?;
Ok(Self {
name: name.to_string(),
version: version.to_string(),
body: DriverBody::Remote(remote),
call_deadlines,
cut: None,
})
}
pub(crate) async fn call(&mut self, call: DriverCall) -> Result<DriverAnswer, RuntimeError> {
Ok(match call {
DriverCall::Describe => DriverAnswer::Description(self.describe().await?),
DriverCall::Launch(request) => DriverAnswer::Launch(self.launch(request).await?),
DriverCall::Parse(lines) => DriverAnswer::Events(self.parse(lines).await?),
DriverCall::ClassifyExit(exit) => DriverAnswer::Event(self.classify_exit(exit).await?),
DriverCall::ReadLogin(output) => DriverAnswer::Login(self.read_login(output).await?),
})
}
async fn remote_call(
&mut self,
call: DriverCall,
) -> Option<Result<DriverAnswer, RuntimeError>> {
match &self.body {
DriverBody::Remote(remote) => Some(remote.call(call, &self.call_deadlines).await),
DriverBody::Local { .. } => None,
}
}
fn unexpected(&self, call: &str, answer: DriverAnswer) -> RuntimeError {
self.failed_in(call, format!("the guest worker answered {answer:?}"))
}
pub(crate) async fn instantiate(
engine: &Engine,
component: &Component,
name: &str,
version: &str,
call_deadlines: CallDeadlines,
) -> Result<Self, RuntimeError> {
let load_error = |message: String| RuntimeError::ProcessDriverLoad {
name: name.to_string(),
version: version.to_string(),
message,
};
let limits = ExecutionLimits::default().for_hook_calls(None);
crate::guest_worker::check_store_placement(crate::guest_worker::StoreSite::ProcessDriver)?;
let mut linker: Linker<ProcessDriverStoreState> = Linker::new(engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker)
.map_err(|e| load_error(e.to_string()))?;
let state = ProcessDriverStoreState {
wasi: WasiCtxBuilder::new().build(),
table: ResourceTable::new(),
limits: limits.limiter(),
};
let mut store = Store::new(engine, state);
store.limiter(|state| &mut state.limits);
store.set_epoch_deadline(limits.deadline_ticks());
let bindings = call_deadlines
.bound(
name,
ComponentRole::ProcessDriver,
"instantiate",
&limits,
ProcessDriverBindings::instantiate_async(&mut store, component, &linker),
)
.await
.map_err(|_| load_error(format!("instantiation {}", wall_deadline_text(&limits))))?
.map_err(|e| load_error(format!("{e:#}")))?;
Ok(Self {
name: name.to_string(),
version: version.to_string(),
body: DriverBody::Local { store, bindings },
call_deadlines,
cut: None,
})
}
pub async fn describe(&mut self) -> Result<Description, RuntimeError> {
if let Some(answer) = self.remote_call(DriverCall::Describe).await {
return match answer? {
DriverAnswer::Description(description) => Ok(description),
other => Err(self.unexpected("describe", other)),
};
}
self.enter("describe")
.map_err(|message| self.load_error(message))?;
let limits = self.reset_deadline();
let DriverBody::Local { store, bindings } = &mut self.body else {
unreachable!("a remote driver answered above")
};
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"describe",
&limits,
bindings.murmur_driver_process().call_describe(&mut *store),
)
.await;
let description = self
.settle("describe", &limits, called)
.map_err(|message| self.load_error(format!("describe() {message}")))?
.map_err(|e| {
let message = match classify_guest_failure(&e, self.limiter()) {
GuestFailure::DeadlineExceeded { seconds } => {
format!("describe() exceeded its {seconds}s deadline and was interrupted")
}
GuestFailure::ResourceLimit { message } => {
format!("describe() exceeded its resource limits: {message}")
}
GuestFailure::Other => format!("describe() trapped: {e:#}"),
};
self.load_error(message)
})?;
validate_description(&self.name, &self.version, &description)?;
Ok(description)
}
pub async fn launch(
&mut self,
request: LaunchRequest,
) -> Result<Result<LaunchPlan, String>, RuntimeError> {
if let DriverBody::Remote(remote) = &self.body {
return match remote
.call(DriverCall::Launch(request), &self.call_deadlines)
.await?
{
DriverAnswer::Launch(plan) => Ok(plan),
other => Err(self.unexpected("launch", other)),
};
}
self.enter("launch")
.map_err(|message| self.failed_in("launch", message))?;
let limits = self.reset_deadline();
let DriverBody::Local { store, bindings } = &mut self.body else {
unreachable!("a remote driver answered above")
};
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"launch",
&limits,
bindings
.murmur_driver_process()
.call_launch(&mut *store, &request),
)
.await;
self.settle("launch", &limits, called)
.map_err(|message| self.failed_in("launch", message))?
.map_err(|e| self.call_error("launch", &e))
}
pub async fn parse(&mut self, lines: Vec<String>) -> Result<Vec<Event>, RuntimeError> {
if let DriverBody::Remote(remote) = &self.body {
return match remote
.call(DriverCall::Parse(lines), &self.call_deadlines)
.await?
{
DriverAnswer::Events(events) => Ok(events),
other => Err(self.unexpected("parse", other)),
};
}
self.enter("parse")
.map_err(|message| self.failed_in("parse", message))?;
let limits = self.reset_deadline();
let DriverBody::Local { store, bindings } = &mut self.body else {
unreachable!("a remote driver answered above")
};
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"parse",
&limits,
bindings
.murmur_driver_process()
.call_parse(&mut *store, &lines),
)
.await;
self.settle("parse", &limits, called)
.map_err(|message| self.failed_in("parse", message))?
.map_err(|e| self.call_error("parse", &e))
}
pub async fn classify_exit(&mut self, exit: ExitStatus) -> Result<Event, RuntimeError> {
if let DriverBody::Remote(remote) = &self.body {
return match remote
.call(DriverCall::ClassifyExit(exit), &self.call_deadlines)
.await?
{
DriverAnswer::Event(event) => Ok(event),
other => Err(self.unexpected("classify-exit", other)),
};
}
self.enter("classify-exit")
.map_err(|message| self.failed_in("classify-exit", message))?;
let limits = self.reset_deadline();
let DriverBody::Local { store, bindings } = &mut self.body else {
unreachable!("a remote driver answered above")
};
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"classify-exit",
&limits,
bindings
.murmur_driver_process()
.call_classify_exit(&mut *store, &exit),
)
.await;
self.settle("classify-exit", &limits, called)
.map_err(|message| self.failed_in("classify-exit", message))?
.map_err(|e| self.call_error("classify-exit", &e))
}
pub async fn read_login(&mut self, output: LoginOutput) -> Result<LoginState, RuntimeError> {
if let DriverBody::Remote(remote) = &self.body {
return match remote
.call(DriverCall::ReadLogin(output), &self.call_deadlines)
.await?
{
DriverAnswer::Login(state) => Ok(state),
other => Err(self.unexpected("read-login", other)),
};
}
self.enter("read-login")
.map_err(|message| self.failed_in("read-login", message))?;
let limits = self.reset_deadline();
let DriverBody::Local { store, bindings } = &mut self.body else {
unreachable!("a remote driver answered above")
};
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"read-login",
&limits,
bindings
.murmur_driver_process()
.call_read_login(&mut *store, &output),
)
.await;
self.settle("read-login", &limits, called)
.map_err(|message| self.failed_in("read-login", message))?
.map_err(|e| self.call_error("read-login", &e))
}
fn reset_deadline(&mut self) -> ExecutionLimits {
let DriverBody::Local { store, .. } = &mut self.body else {
return ExecutionLimits::default().for_hook_calls(None);
};
let limits = store.data().limits.limits();
store.set_epoch_deadline(limits.deadline_ticks());
limits
}
fn limiter(&self) -> &ExecutionLimiter {
match &self.body {
DriverBody::Local { store, .. } => &store.data().limits,
DriverBody::Remote(_) => unreachable!("a remote driver classifies in its worker"),
}
}
fn enter(&self, call: &str) -> Result<(), String> {
match self.cut {
None => Ok(()),
Some((earlier, seconds)) => Err(format!(
"{call} was not called: the earlier {earlier} call was stopped at its {seconds}s \
deadline, and a stopped driver is not entered again"
)),
}
}
fn settle<T>(
&mut self,
call: &'static str,
limits: &ExecutionLimits,
called: Result<T, WallDeadline>,
) -> Result<T, String> {
called.map_err(|_| {
self.cut = Some((call, limits.deadline_seconds));
wall_deadline_text(limits)
})
}
fn failed_in(&self, call: &str, message: String) -> RuntimeError {
RuntimeError::ProcessDriverCallFailed {
name: self.name.clone(),
version: self.version.clone(),
call: call.to_string(),
message,
}
}
fn call_error(&self, call: &str, error: &wasmtime::Error) -> RuntimeError {
let message = match classify_guest_failure(error, self.limiter()) {
GuestFailure::DeadlineExceeded { seconds } => {
format!("exceeded its {seconds}s deadline and was interrupted")
}
GuestFailure::ResourceLimit { message } => {
format!("exceeded its resource limits: {message}")
}
GuestFailure::Other => format!("trapped: {error:#}"),
};
self.failed_in(call, message)
}
fn load_error(&self, message: String) -> RuntimeError {
RuntimeError::ProcessDriverLoad {
name: self.name.clone(),
version: self.version.clone(),
message,
}
}
}
fn wall_deadline_text(limits: &ExecutionLimits) -> String {
format!(
"waited on the host past its {}s deadline and was stopped",
limits.deadline_seconds
)
}
pub(crate) fn is_usable_env_name(name: &str) -> bool {
!name.is_empty() && !name.contains('=') && !name.contains('\0')
}
fn unusable_env_name(description: &Description) -> Option<&str> {
description
.required_env
.iter()
.find(|var| !is_usable_env_name(var))
.map(String::as_str)
}
fn validate_description(
name: &str,
version: &str,
description: &Description,
) -> Result<(), RuntimeError> {
let load_error = |message: String| RuntimeError::ProcessDriverLoad {
name: name.to_string(),
version: version.to_string(),
message,
};
if let Some(bad) = unusable_env_name(description) {
return Err(load_error(format!(
"describe() names {bad:?} in required-env, which is not a usable variable name"
)));
}
if matches!(&description.login_check, LoginCheck::Command(args) if args.is_empty()) {
return Err(load_error(
"describe() names a login-check command with no arguments; the bare harness binary \
may start a session, so the login check must name the arguments that make it report \
its login"
.to_string(),
));
}
if let Err(reason) = crate::harness_needs::validate_harness_needs(&description.harness_needs) {
return Err(load_error(format!("describe() {reason}")));
}
Ok(())
}
const STANDALONE_NAME: &str = "<wasm>";
const STANDALONE_VERSION: &str = "";
pub(crate) struct StandaloneDriver {
engine: Engine,
component: Component,
pub(crate) rt: tokio::runtime::Runtime,
_ticker: EpochTicker,
name: String,
version: String,
workers: Option<std::sync::Arc<crate::guest_worker::GuestWorkers>>,
}
impl StandaloneDriver {
fn compile(wasm: &[u8]) -> Result<Self, RuntimeError> {
Self::compile_named(wasm, STANDALONE_NAME, STANDALONE_VERSION)
}
pub(crate) fn compile_named(
wasm: &[u8],
name: &str,
version: &str,
) -> Result<Self, RuntimeError> {
let engine = crate::runtime::build_engine()?;
let ticker = EpochTicker::spawn(&engine);
let component =
Component::new(&engine, wasm).map_err(|e| RuntimeError::ToolComponentCompile {
name: name.to_string(),
version: version.to_string(),
message: e.to_string(),
})?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| {
RuntimeError::Runtime(format!("failed to build process driver runtime: {e}"))
})?;
Ok(StandaloneDriver {
engine,
component,
rt,
_ticker: ticker,
name: name.to_string(),
version: version.to_string(),
workers: crate::guest_worker::session_launcher()
.map(crate::guest_worker::GuestWorkers::new),
})
}
pub(crate) async fn instantiate(&self) -> Result<ProcessDriver, RuntimeError> {
ProcessDriver::open(
self.workers.as_ref(),
&self.engine,
&self.component,
&self.name,
&self.version,
CallDeadlines::default(),
)
.await
}
}
pub fn describe_process_driver_wasm(wasm: &[u8]) -> Result<Description, RuntimeError> {
let driver = StandaloneDriver::compile(wasm)?;
driver.rt.block_on(async {
let mut instance = driver.instantiate().await?;
instance.describe().await
})
}
pub fn launch_and_parse_process_driver_wasm(
wasm: &[u8],
request: LaunchRequest,
lines: Vec<String>,
) -> Result<(Result<LaunchPlan, String>, Vec<Event>), RuntimeError> {
let driver = StandaloneDriver::compile(wasm)?;
driver.rt.block_on(async {
let mut instance = driver.instantiate().await?;
let plan = instance.launch(request).await?;
let events = instance.parse(lines).await?;
Ok((plan, events))
})
}
#[cfg(test)]
mod tests {
use murmur_artifact::ContractDirection;
use super::*;
fn served() -> &'static WitLevel {
crate::served_wit_level()
}
fn contracts(exports: &[&str]) -> WitContracts {
WitContracts {
exports: exports.iter().map(|e| e.to_string()).collect(),
imports: Vec::new(),
}
}
#[test]
fn process_driver_interface_process_transport_accepts_the_process_export() {
let c = contracts(&[PROCESS_DRIVER_IFACE]);
assert!(check_driver_interface("process", "d", "1.0.0", Some(&c), served()).is_ok());
}
#[test]
fn process_driver_interface_process_transport_refuses_a_tool_export() {
let c = contracts(&["murmur:tool/run@0.2.0"]);
let err = check_driver_interface("process", "d", "1.0.0", Some(&c), served()).unwrap_err();
match &err {
RuntimeError::ProcessDriverInterfaceMissing { expected, .. } => {
assert_eq!(expected, PROCESS_DRIVER_IFACE);
}
other => panic!("unexpected error: {other:?}"),
}
let message = err.to_string();
assert!(message.contains("'d@1.0.0'"), "{message}");
assert!(message.contains(PROCESS_DRIVER_IFACE), "{message}");
}
#[test]
fn process_driver_interface_process_transport_refuses_another_version_as_a_level_mismatch() {
let c = contracts(&["murmur:driver/process@0.0.1"]);
let err = check_driver_interface("process", "d", "1.0.0", Some(&c), served()).unwrap_err();
match &err {
RuntimeError::WitLevelMismatch {
level, mismatched, ..
} => {
assert_eq!(*level, served().level);
assert_eq!(mismatched.len(), 1);
assert_eq!(mismatched[0].interface, "murmur:driver/process@0.0.1");
assert_eq!(mismatched[0].direction, ContractDirection::Export);
}
other => panic!("unexpected error: {other:?}"),
}
assert_eq!(
err.to_string(),
format!(
"artifact 'd@1.0.0' does not fit WIT level {0}, the level this mur serves: it \
exports murmur:driver/process@0.0.1, and level {0} has murmur:driver@{1}",
served().level,
served().version_of("murmur:driver").unwrap()
)
);
}
fn level_1() -> &'static WitLevel {
&crate::wit_versions::wit_levels()[0]
}
#[test]
fn process_driver_interface_follows_the_level_it_is_given() {
let c = contracts(&[PROCESS_DRIVER_IFACE]);
assert!(check_driver_interface("process", "d", "1.0.0", Some(&c), served()).is_ok());
match check_driver_interface("process", "d", "1.0.0", Some(&c), level_1()).unwrap_err() {
RuntimeError::WitLevelMismatch {
level, mismatched, ..
} => {
assert_eq!(level, 1);
assert_eq!(mismatched[0].direction, ContractDirection::Export);
assert_eq!(mismatched[0].package, "murmur:driver");
assert_eq!(mismatched[0].served_version.as_deref(), Some("0.3.0"));
}
other => panic!("unexpected error: {other:?}"),
}
let old = contracts(&["murmur:driver/process@0.3.0"]);
assert!(check_driver_interface("process", "d", "1.0.0", Some(&old), level_1()).is_ok());
let stream = importing(&["murmur:stream/events@0.1.0"]);
assert!(check_stream_interface("d", "1.0.0", Some(&stream), level_1()).is_ok());
}
#[test]
fn process_driver_interface_process_transport_refuses_no_contracts() {
let err = check_driver_interface("process", "d", "1.0.0", None, served()).unwrap_err();
assert!(matches!(
err,
RuntimeError::ProcessDriverInterfaceMissing { .. }
));
}
#[test]
fn process_driver_interface_http_transport_refuses_the_process_export() {
let c = contracts(&[PROCESS_DRIVER_IFACE]);
let err = check_driver_interface("http", "d", "1.0.0", Some(&c), served()).unwrap_err();
match &err {
RuntimeError::HttpDriverExportsProcessInterface { name, exported, .. } => {
assert_eq!(name, "d");
assert_eq!(exported, &vec![PROCESS_DRIVER_IFACE.to_string()]);
}
other => panic!("unexpected error: {other:?}"),
}
}
#[test]
fn process_driver_interface_http_transport_accepts_a_tool_export() {
let c = contracts(&["murmur:tool/run@0.2.0"]);
assert!(check_driver_interface("http", "d", "1.0.0", Some(&c), served()).is_ok());
}
fn importing(imports: &[&str]) -> WitContracts {
WitContracts {
exports: vec!["murmur:tool/run@0.2.0".to_string()],
imports: imports.iter().map(|i| i.to_string()).collect(),
}
}
#[test]
fn stream_interface_refuses_and_lists_a_murmur_text_import() {
let c = importing(&[
"wasi:cli/environment@0.2.12",
"murmur:text/chunks@0.1.0",
"murmur:task/task@0.1.0",
]);
let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
match &err {
RuntimeError::WitLevelMismatch {
name,
version,
mismatched,
..
} => {
assert_eq!((name.as_str(), version.as_str()), ("d", "1.0.0"));
assert_eq!(mismatched.len(), 1);
assert_eq!(mismatched[0].interface, "murmur:text/chunks@0.1.0");
assert_eq!(mismatched[0].served_version, None);
}
other => panic!("unexpected error: {other:?}"),
}
let message = err.to_string();
assert!(message.contains("'d@1.0.0'"), "{message}");
assert!(
message.contains(
"it imports murmur:text/chunks@0.1.0, and level 6 has no version of murmur:text"
),
"{message}"
);
}
#[test]
fn stream_interface_refuses_another_version_of_murmur_stream() {
let c = importing(&["murmur:stream/events@0.2.0"]);
let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
assert!(matches!(
err,
RuntimeError::WitLevelMismatch { ref mismatched, .. }
if mismatched.len() == 1 && mismatched[0].interface == "murmur:stream/events@0.2.0"
));
}
#[test]
fn stream_interface_refuses_an_unknown_interface_at_the_served_version_as_not_served() {
let c = importing(&["murmur:stream/other@0.1.0"]);
let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
match &err {
RuntimeError::DriverStreamInterfaceNotServed {
imported, expected, ..
} => {
assert_eq!(imported, &vec!["murmur:stream/other@0.1.0".to_string()]);
assert_eq!(expected, crate::runtime::WIT_STREAM_EVENTS_IFACE);
}
other => panic!("unexpected error: {other:?}"),
}
let both = importing(&["murmur:stream/other@0.1.0", "murmur:text/chunks@0.1.0"]);
assert!(matches!(
check_stream_interface("d", "1.0.0", Some(&both), served()).unwrap_err(),
RuntimeError::WitLevelMismatch { ref mismatched, .. }
if mismatched.len() == 1 && mismatched[0].interface == "murmur:text/chunks@0.1.0"
));
}
#[test]
fn driver_refusal_reports_a_level_mismatch_over_any_other() {
let not_served = || {
check_stream_interface(
"d",
"1.0.0",
Some(&importing(&["murmur:stream/other@0.1.0"])),
served(),
)
};
let old_export = || {
check_driver_interface(
"process",
"d",
"1.0.0",
Some(&contracts(&["murmur:driver/process@0.2.0"])),
served(),
)
};
let text_import = || {
check_stream_interface(
"d",
"1.0.0",
Some(&importing(&["murmur:text/chunks@0.1.0"])),
served(),
)
};
let no_export = || check_driver_interface("process", "d", "1.0.0", None, served());
assert!(matches!(
driver_refusal(not_served(), old_export()),
Err(RuntimeError::WitLevelMismatch { .. })
));
assert!(matches!(
driver_refusal(text_import(), no_export()),
Err(RuntimeError::WitLevelMismatch { .. })
));
assert!(matches!(
driver_refusal(not_served(), no_export()),
Err(RuntimeError::DriverStreamInterfaceNotServed { .. })
));
match driver_refusal(text_import(), old_export()) {
Err(RuntimeError::WitLevelMismatch { mismatched, .. }) => {
let names: Vec<&str> = mismatched.iter().map(|m| m.interface.as_str()).collect();
assert_eq!(
names,
["murmur:driver/process@0.2.0", "murmur:text/chunks@0.1.0"]
);
}
other => panic!("unexpected result: {other:?}"),
}
assert!(driver_refusal(Ok(()), Ok(())).is_ok());
}
#[test]
fn stream_interface_accepts_exactly_the_served_instance() {
let c = importing(&["murmur:stream/events@0.1.0", "wasi:http/types@0.2.12"]);
assert!(check_stream_interface("d", "1.0.0", Some(&c), served()).is_ok());
}
#[test]
fn stream_interface_accepts_a_component_with_no_murmur_import() {
let c = importing(&["wasi:cli/environment@0.2.12"]);
assert!(check_stream_interface("d", "1.0.0", Some(&c), served()).is_ok());
assert!(check_stream_interface("d", "1.0.0", Some(&importing(&[])), served()).is_ok());
}
#[test]
fn stream_interface_accepts_no_contracts() {
assert!(check_stream_interface("d", "1.0.0", None, served()).is_ok());
}
fn description(required_env: &[&str]) -> Description {
Description {
harness: "h".to_string(),
binary: "b".to_string(),
version_args: Vec::new(),
tested_versions: Vec::new(),
interrupt: InterruptMethod::Unsupported,
required_env: required_env.iter().map(|v| v.to_string()).collect(),
streams_text: false,
reports_usage: false,
login_check: LoginCheck::Unavailable("no login".to_string()),
harness_needs: HarnessNeeds {
paths: Vec::new(),
hosts: Vec::new(),
},
}
}
#[test]
fn login_check_command_with_arguments_passes() {
let mut described = description(&[]);
described.login_check = LoginCheck::Command(vec!["auth".to_string(), "status".to_string()]);
assert!(validate_description("d", "1.0.0", &described).is_ok());
assert!(validate_description("d", "1.0.0", &description(&[])).is_ok());
}
#[test]
fn login_check_command_with_no_arguments_is_refused_at_load() {
let mut described = description(&[]);
described.login_check = LoginCheck::Command(Vec::new());
let err = validate_description("d", "1.0.0", &described).unwrap_err();
match &err {
RuntimeError::ProcessDriverLoad { message, .. } => {
assert!(
message.contains("login-check command with no arguments"),
"{message}"
);
}
other => panic!("unexpected error: {other:?}"),
}
}
#[test]
fn required_env_all_declared_passes() {
let allow = vec!["A".to_string(), "B".to_string(), "C".to_string()];
assert!(check_required_env("d", "1.0.0", &description(&["B", "A"]), &allow).is_ok());
}
#[test]
fn required_env_reports_missing_names_in_order_once_each() {
let allow = vec!["B".to_string()];
let err = check_required_env("d", "1.0.0", &description(&["Z", "B", "A", "Z"]), &allow)
.unwrap_err();
match &err {
RuntimeError::ProcessDriverRequiredEnvNotAllowed { missing, .. } => {
assert_eq!(missing, &vec!["Z".to_string(), "A".to_string()]);
}
other => panic!("unexpected error: {other:?}"),
}
let message = err.to_string();
assert!(message.contains("'d@1.0.0'"), "{message}");
assert!(message.contains("Z, A"), "{message}");
}
#[test]
fn required_env_usable_names_pass() {
assert_eq!(
unusable_env_name(&description(&["HOME", "PROFILE_2"])),
None
);
assert_eq!(unusable_env_name(&description(&[])), None);
}
#[test]
fn required_env_refuses_a_name_that_is_not_a_variable() {
for bad in ["", "A=B", "A\0B"] {
assert_eq!(
unusable_env_name(&description(&["HOME", bad, "TAIL"])),
Some(bad),
"{bad:?} should be refused"
);
}
}
#[test]
#[ignore = "a measurement, not an assertion; run with --release --ignored --nocapture"]
fn process_driver_parse_cost() {
use std::time::Instant;
const WASM: &[u8] = include_bytes!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
));
let engine = crate::runtime::build_engine().expect("engine");
let _ticker = EpochTicker::spawn(&engine);
let component = Component::new(&engine, WASM).expect("fixture compiles");
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let instantiations = 50;
let start = Instant::now();
for _ in 0..instantiations {
rt.block_on(ProcessDriver::instantiate(
&engine,
&component,
"d",
"0.1.0",
CallDeadlines::default(),
))
.expect("instantiates");
}
let per_instantiation = start.elapsed() / instantiations;
let mut driver = rt
.block_on(ProcessDriver::instantiate(
&engine,
&component,
"d",
"0.1.0",
CallDeadlines::default(),
))
.expect("instantiates");
for lines in [1usize, 64] {
let batch: Vec<String> = (0..lines).map(|i| format!("text line {i}")).collect();
let runs = 2_000;
let mut samples = Vec::with_capacity(runs);
for _ in 0..runs {
let start = Instant::now();
let events = rt.block_on(driver.parse(batch.clone())).expect("parses");
samples.push(start.elapsed());
assert_eq!(events.len(), lines);
}
samples.sort_unstable();
let median = samples[samples.len() / 2];
let p95 = samples[samples.len() * 95 / 100];
crate::diagnostic::raw_to_stdout(&format!(
"parse {lines}-line batch on one reused instance: median {median:?}, p95 {p95:?} \
({runs} calls)\n"
));
}
crate::diagnostic::raw_to_stdout(&format!(
"instantiate one driver: {per_instantiation:?} (mean of {instantiations})\n"
));
}
#[test]
fn process_driver_granted_nothing() {
let wat = r#"(component
(import "murmur:stream/events@0.1.0" (instance
(export "emit-chunk" (func (param "chunk" string)))
))
)"#;
let bytes = wat::parse_str(wat).expect("component WAT parses");
let engine = crate::runtime::build_engine().expect("engine");
let _ticker = EpochTicker::spawn(&engine);
let component = Component::new(&engine, &bytes).expect("component compiles");
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let result = rt.block_on(ProcessDriver::instantiate(
&engine,
&component,
"d",
"1.0.0",
CallDeadlines::default(),
));
match result {
Err(RuntimeError::ProcessDriverLoad { name, message, .. }) => {
assert_eq!(name, "d");
assert!(message.contains("murmur:stream/events@0.1.0"), "{message}");
}
Err(other) => panic!("unexpected error: {other:?}"),
Ok(_) => panic!("a driver importing a Murmur interface instantiated"),
}
}
#[test]
fn launch_and_parse_run_on_one_instance() {
const WASM: &[u8] = include_bytes!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
));
let request = LaunchRequest {
model: None,
system_prompt: String::new(),
config: None,
bridge: Some(Bridge {
server_name: "murmur".to_string(),
url: "http://127.0.0.1:1/mcp".to_string(),
bearer_token: "token".to_string(),
tool_names: vec!["python3".to_string()],
}),
session: Session {
id: "s1".to_string(),
mode: SessionMode::New,
},
harness_version: None,
task: "probe".to_string(),
};
let lines = vec![
r#"tool c1 murmur__python3 {"command":"x"}"#.to_string(),
"end done".to_string(),
];
let (plan, events) =
launch_and_parse_process_driver_wasm(WASM, request, lines).expect("driver runs");
let plan = plan.expect("the fixture accepts a non-empty task");
assert!(plan
.env_set
.iter()
.any(|(name, value)| name == "FIXTURE_BRIDGE_TOKEN" && value == "token"));
match &events[..] {
[Event::ToolCall(call), Event::TurnEnd(summary)] => {
assert_eq!(call.name, "python3");
assert_eq!(call.input, r#"{"command":"x"}"#);
assert_eq!(summary, "done");
}
other => panic!("unexpected events: {other:?}"),
}
}
}