use murmur_artifact::{
current_platform, 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, InterruptMethod, LaunchPlan,
LaunchRequest, 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.4.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, current_platform()) {
VersionFit::Unserved(mismatched) => Some(RuntimeError::WitLevelMismatch {
name: name.to_string(),
version: version.to_string(),
level: level.level,
mismatched,
}),
VersionFit::Fits | VersionFit::NoPayloadFor(_) => 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,
store: Store<ProcessDriverStoreState>,
bindings: ProcessDriverBindings,
call_deadlines: CallDeadlines,
cut: Option<(&'static str, u64)>,
}
impl ProcessDriver {
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);
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(),
store,
bindings,
call_deadlines,
cut: None,
})
}
pub async fn describe(&mut self) -> Result<Description, RuntimeError> {
self.enter("describe")
.map_err(|message| self.load_error(message))?;
let limits = self.reset_deadline();
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"describe",
&limits,
self.bindings
.murmur_driver_process()
.call_describe(&mut self.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.store.data().limits) {
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)
})?;
if let Some(bad) = unusable_env_name(&description) {
return Err(self.load_error(format!(
"describe() names {bad:?} in required-env, which is not a usable variable name"
)));
}
Ok(description)
}
pub async fn launch(
&mut self,
request: LaunchRequest,
) -> Result<Result<LaunchPlan, String>, RuntimeError> {
self.enter("launch")
.map_err(|message| self.failed_in("launch", message))?;
let limits = self.reset_deadline();
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"launch",
&limits,
self.bindings
.murmur_driver_process()
.call_launch(&mut self.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> {
self.enter("parse")
.map_err(|message| self.failed_in("parse", message))?;
let limits = self.reset_deadline();
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"parse",
&limits,
self.bindings
.murmur_driver_process()
.call_parse(&mut self.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> {
self.enter("classify-exit")
.map_err(|message| self.failed_in("classify-exit", message))?;
let limits = self.reset_deadline();
let called = self
.call_deadlines
.bound(
&self.name,
ComponentRole::ProcessDriver,
"classify-exit",
&limits,
self.bindings
.murmur_driver_process()
.call_classify_exit(&mut self.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))
}
fn reset_deadline(&mut self) -> ExecutionLimits {
let limits = self.store.data().limits.limits();
self.store.set_epoch_deadline(limits.deadline_ticks());
limits
}
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.store.data().limits) {
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)
}
const STANDALONE_NAME: &str = "<wasm>";
const STANDALONE_VERSION: &str = "";
struct StandaloneDriver {
engine: Engine,
component: Component,
rt: tokio::runtime::Runtime,
_ticker: EpochTicker,
}
impl StandaloneDriver {
fn compile(wasm: &[u8]) -> 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: STANDALONE_NAME.to_string(),
version: STANDALONE_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,
})
}
async fn instantiate(&self) -> Result<ProcessDriver, RuntimeError> {
ProcessDriver::instantiate(
&self.engine,
&self.component,
STANDALONE_NAME,
STANDALONE_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(&["murmur:driver/process@0.4.0"]);
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 3 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,
}
}
#[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:?}"),
}
}
}