1use murmur_artifact::WitContracts;
9use wasmtime::component::{Component, Linker, ResourceTable};
10use wasmtime::{Engine, Store};
11use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView};
12
13use crate::bindings::process_driver::ProcessDriver as ProcessDriverBindings;
14use crate::errors::RuntimeError;
15use crate::limits::{classify_guest_failure, EpochTicker, ExecutionLimiter, GuestFailure};
16use crate::ExecutionLimits;
17
18pub use crate::bindings::process_driver::exports::murmur::driver::process::{
19 Bridge, Description, DriverFile, Event, ExitStatus, FailureKind, InterruptMethod, LaunchPlan,
20 LaunchRequest, RetryInfo, Session, SessionInfo, SessionMode, ToolCallInfo, ToolCallProgress,
21 ToolCallStart, ToolResultInfo, TurnFailure, Usage,
22};
23
24pub const PROCESS_DRIVER_IFACE: &str = "murmur:driver/process@0.3.0";
26
27const PROCESS_DRIVER_IFACE_PREFIX: &str = "murmur:driver/process@";
30
31pub fn check_driver_interface(
39 transport: &str,
40 name: &str,
41 version: &str,
42 contracts: Option<&WitContracts>,
43) -> Result<(), RuntimeError> {
44 let process_exports: Vec<String> = contracts
45 .map(|c| {
46 c.exports
47 .iter()
48 .filter(|export| export.starts_with(PROCESS_DRIVER_IFACE_PREFIX))
49 .cloned()
50 .collect()
51 })
52 .unwrap_or_default();
53 match transport {
54 "process" if process_exports.iter().any(|e| e == PROCESS_DRIVER_IFACE) => Ok(()),
55 "process" => Err(RuntimeError::ProcessDriverInterfaceMissing {
56 name: name.to_string(),
57 version: version.to_string(),
58 expected: PROCESS_DRIVER_IFACE.to_string(),
59 found: process_exports,
60 }),
61 "http" if !process_exports.is_empty() => {
62 Err(RuntimeError::HttpDriverExportsProcessInterface {
63 name: name.to_string(),
64 version: version.to_string(),
65 exported: process_exports,
66 })
67 }
68 _ => Ok(()),
69 }
70}
71
72const RETIRED_TEXT_IFACE_PREFIX: &str = "murmur:text/";
75
76const STREAM_IFACE_PREFIX: &str = "murmur:stream/";
78
79pub fn check_stream_interface(
89 name: &str,
90 version: &str,
91 contracts: Option<&WitContracts>,
92) -> Result<(), RuntimeError> {
93 let expected = crate::runtime::WIT_STREAM_EVENTS_IFACE;
94 let imported: Vec<String> = contracts
95 .map(|c| {
96 c.imports
97 .iter()
98 .filter(|import| {
99 import.starts_with(RETIRED_TEXT_IFACE_PREFIX)
100 || (import.starts_with(STREAM_IFACE_PREFIX) && *import != expected)
101 })
102 .cloned()
103 .collect()
104 })
105 .unwrap_or_default();
106 if imported.is_empty() {
107 return Ok(());
108 }
109 Err(RuntimeError::DriverStreamInterfaceNotServed {
110 name: name.to_string(),
111 version: version.to_string(),
112 imported,
113 expected: expected.to_string(),
114 })
115}
116
117pub fn check_required_env(
122 name: &str,
123 version: &str,
124 description: &Description,
125 env_allow: &[String],
126) -> Result<(), RuntimeError> {
127 let mut missing: Vec<String> = Vec::new();
128 for var in &description.required_env {
129 if !env_allow.contains(var) && !missing.contains(var) {
130 missing.push(var.clone());
131 }
132 }
133 if missing.is_empty() {
134 Ok(())
135 } else {
136 Err(RuntimeError::ProcessDriverRequiredEnvNotAllowed {
137 name: name.to_string(),
138 version: version.to_string(),
139 missing,
140 })
141 }
142}
143
144struct ProcessDriverStoreState {
146 wasi: WasiCtx,
147 table: ResourceTable,
148 limits: ExecutionLimiter,
149}
150
151impl WasiView for ProcessDriverStoreState {
152 fn ctx(&mut self) -> WasiCtxView<'_> {
153 WasiCtxView {
154 ctx: &mut self.wasi,
155 table: &mut self.table,
156 }
157 }
158}
159
160pub struct ProcessDriver {
163 name: String,
164 version: String,
165 store: Store<ProcessDriverStoreState>,
166 bindings: ProcessDriverBindings,
167}
168
169impl ProcessDriver {
170 pub async fn instantiate(
179 engine: &Engine,
180 component: &Component,
181 name: &str,
182 version: &str,
183 ) -> Result<Self, RuntimeError> {
184 let load_error = |message: String| RuntimeError::ProcessDriverLoad {
185 name: name.to_string(),
186 version: version.to_string(),
187 message,
188 };
189 let limits = ExecutionLimits::default().for_hook_calls(None);
190
191 let mut linker: Linker<ProcessDriverStoreState> = Linker::new(engine);
192 wasmtime_wasi::p2::add_to_linker_async(&mut linker)
193 .map_err(|e| load_error(e.to_string()))?;
194
195 let state = ProcessDriverStoreState {
196 wasi: WasiCtxBuilder::new().build(),
199 table: ResourceTable::new(),
200 limits: limits.limiter(),
201 };
202 let mut store = Store::new(engine, state);
203 store.limiter(|state| &mut state.limits);
204 store.set_epoch_deadline(limits.deadline_ticks());
205
206 let bindings = ProcessDriverBindings::instantiate_async(&mut store, component, &linker)
207 .await
208 .map_err(|e| load_error(format!("{e:#}")))?;
209 Ok(Self {
210 name: name.to_string(),
211 version: version.to_string(),
212 store,
213 bindings,
214 })
215 }
216
217 pub async fn describe(&mut self) -> Result<Description, RuntimeError> {
222 self.reset_deadline();
223 let description = self
224 .bindings
225 .murmur_driver_process()
226 .call_describe(&mut self.store)
227 .await
228 .map_err(|e| {
229 let message = match classify_guest_failure(&e, &self.store.data().limits) {
230 GuestFailure::DeadlineExceeded { seconds } => {
231 format!("describe() exceeded its {seconds}s deadline and was interrupted")
232 }
233 GuestFailure::ResourceLimit { message } => {
234 format!("describe() exceeded its resource limits: {message}")
235 }
236 GuestFailure::Other => format!("describe() trapped: {e:#}"),
237 };
238 self.load_error(message)
239 })?;
240 if let Some(bad) = unusable_env_name(&description) {
241 return Err(self.load_error(format!(
242 "describe() names {bad:?} in required-env, which is not a usable variable name"
243 )));
244 }
245 Ok(description)
246 }
247
248 pub async fn launch(
251 &mut self,
252 request: LaunchRequest,
253 ) -> Result<Result<LaunchPlan, String>, RuntimeError> {
254 self.reset_deadline();
255 self.bindings
256 .murmur_driver_process()
257 .call_launch(&mut self.store, &request)
258 .await
259 .map_err(|e| self.call_error("launch", &e))
260 }
261
262 pub async fn parse(&mut self, lines: Vec<String>) -> Result<Vec<Event>, RuntimeError> {
265 self.reset_deadline();
266 self.bindings
267 .murmur_driver_process()
268 .call_parse(&mut self.store, &lines)
269 .await
270 .map_err(|e| self.call_error("parse", &e))
271 }
272
273 pub async fn classify_exit(&mut self, exit: ExitStatus) -> Result<Event, RuntimeError> {
275 self.reset_deadline();
276 self.bindings
277 .murmur_driver_process()
278 .call_classify_exit(&mut self.store, &exit)
279 .await
280 .map_err(|e| self.call_error("classify-exit", &e))
281 }
282
283 fn reset_deadline(&mut self) {
287 let deadline = self.store.data().limits.limits().deadline_ticks();
288 self.store.set_epoch_deadline(deadline);
289 }
290
291 fn call_error(&self, call: &str, error: &wasmtime::Error) -> RuntimeError {
292 let message = match classify_guest_failure(error, &self.store.data().limits) {
293 GuestFailure::DeadlineExceeded { seconds } => {
294 format!("exceeded its {seconds}s deadline and was interrupted")
295 }
296 GuestFailure::ResourceLimit { message } => {
297 format!("exceeded its resource limits: {message}")
298 }
299 GuestFailure::Other => format!("trapped: {error:#}"),
300 };
301 RuntimeError::ProcessDriverCallFailed {
302 name: self.name.clone(),
303 version: self.version.clone(),
304 call: call.to_string(),
305 message,
306 }
307 }
308
309 fn load_error(&self, message: String) -> RuntimeError {
310 RuntimeError::ProcessDriverLoad {
311 name: self.name.clone(),
312 version: self.version.clone(),
313 message,
314 }
315 }
316}
317
318pub(crate) fn is_usable_env_name(name: &str) -> bool {
325 !name.is_empty() && !name.contains('=') && !name.contains('\0')
326}
327
328fn unusable_env_name(description: &Description) -> Option<&str> {
331 description
332 .required_env
333 .iter()
334 .find(|var| !is_usable_env_name(var))
335 .map(String::as_str)
336}
337
338const STANDALONE_NAME: &str = "<wasm>";
341const STANDALONE_VERSION: &str = "";
342
343struct StandaloneDriver {
346 engine: Engine,
347 component: Component,
348 rt: tokio::runtime::Runtime,
349 _ticker: EpochTicker,
350}
351
352impl StandaloneDriver {
353 fn compile(wasm: &[u8]) -> Result<Self, RuntimeError> {
354 let engine = crate::runtime::build_engine()?;
355 let ticker = EpochTicker::spawn(&engine);
356 let component =
357 Component::new(&engine, wasm).map_err(|e| RuntimeError::ToolComponentCompile {
358 name: STANDALONE_NAME.to_string(),
359 version: STANDALONE_VERSION.to_string(),
360 message: e.to_string(),
361 })?;
362 let rt = tokio::runtime::Builder::new_current_thread()
363 .enable_all()
364 .build()
365 .map_err(|e| {
366 RuntimeError::Runtime(format!("failed to build process driver runtime: {e}"))
367 })?;
368 Ok(StandaloneDriver {
369 engine,
370 component,
371 rt,
372 _ticker: ticker,
373 })
374 }
375
376 async fn instantiate(&self) -> Result<ProcessDriver, RuntimeError> {
377 ProcessDriver::instantiate(
378 &self.engine,
379 &self.component,
380 STANDALONE_NAME,
381 STANDALONE_VERSION,
382 )
383 .await
384 }
385}
386
387pub fn describe_process_driver_wasm(wasm: &[u8]) -> Result<Description, RuntimeError> {
394 let driver = StandaloneDriver::compile(wasm)?;
395 driver.rt.block_on(async {
396 let mut instance = driver.instantiate().await?;
397 instance.describe().await
398 })
399}
400
401pub fn launch_and_parse_process_driver_wasm(
408 wasm: &[u8],
409 request: LaunchRequest,
410 lines: Vec<String>,
411) -> Result<(Result<LaunchPlan, String>, Vec<Event>), RuntimeError> {
412 let driver = StandaloneDriver::compile(wasm)?;
413 driver.rt.block_on(async {
414 let mut instance = driver.instantiate().await?;
415 let plan = instance.launch(request).await?;
416 let events = instance.parse(lines).await?;
417 Ok((plan, events))
418 })
419}
420
421#[cfg(test)]
422mod tests {
423 use super::*;
424
425 fn contracts(exports: &[&str]) -> WitContracts {
426 WitContracts {
427 exports: exports.iter().map(|e| e.to_string()).collect(),
428 imports: Vec::new(),
429 }
430 }
431
432 #[test]
433 fn process_driver_interface_process_transport_accepts_the_process_export() {
434 let c = contracts(&[PROCESS_DRIVER_IFACE]);
435 assert!(check_driver_interface("process", "d", "1.0.0", Some(&c)).is_ok());
436 }
437
438 #[test]
439 fn process_driver_interface_process_transport_refuses_a_tool_export() {
440 let c = contracts(&["murmur:tool/run@0.1.0"]);
441 let err = check_driver_interface("process", "d", "1.0.0", Some(&c)).unwrap_err();
442 match &err {
443 RuntimeError::ProcessDriverInterfaceMissing {
444 expected, found, ..
445 } => {
446 assert_eq!(expected, PROCESS_DRIVER_IFACE);
447 assert!(found.is_empty());
448 }
449 other => panic!("unexpected error: {other:?}"),
450 }
451 let message = err.to_string();
452 assert!(message.contains("'d@1.0.0'"), "{message}");
453 assert!(message.contains(PROCESS_DRIVER_IFACE), "{message}");
454 }
455
456 #[test]
457 fn process_driver_interface_process_transport_names_the_version_found() {
458 let c = contracts(&["murmur:driver/process@0.0.1"]);
459 let err = check_driver_interface("process", "d", "1.0.0", Some(&c)).unwrap_err();
460 match &err {
461 RuntimeError::ProcessDriverInterfaceMissing { found, .. } => {
462 assert_eq!(found, &vec!["murmur:driver/process@0.0.1".to_string()]);
463 }
464 other => panic!("unexpected error: {other:?}"),
465 }
466 let message = err.to_string();
467 assert!(message.contains("murmur:driver/process@0.0.1"), "{message}");
468 assert!(message.contains("rebuild it"), "{message}");
469 }
470
471 #[test]
472 fn process_driver_interface_process_transport_refuses_no_contracts() {
473 let err = check_driver_interface("process", "d", "1.0.0", None).unwrap_err();
474 assert!(matches!(
475 err,
476 RuntimeError::ProcessDriverInterfaceMissing { ref found, .. } if found.is_empty()
477 ));
478 }
479
480 #[test]
481 fn process_driver_interface_http_transport_refuses_the_process_export() {
482 let c = contracts(&[PROCESS_DRIVER_IFACE]);
483 let err = check_driver_interface("http", "d", "1.0.0", Some(&c)).unwrap_err();
484 match &err {
485 RuntimeError::HttpDriverExportsProcessInterface { name, exported, .. } => {
486 assert_eq!(name, "d");
487 assert_eq!(exported, &vec![PROCESS_DRIVER_IFACE.to_string()]);
488 }
489 other => panic!("unexpected error: {other:?}"),
490 }
491 }
492
493 #[test]
494 fn process_driver_interface_http_transport_accepts_a_tool_export() {
495 let c = contracts(&["murmur:tool/run@0.1.0"]);
496 assert!(check_driver_interface("http", "d", "1.0.0", Some(&c)).is_ok());
497 }
498
499 fn importing(imports: &[&str]) -> WitContracts {
500 WitContracts {
501 exports: vec!["murmur:tool/run@0.1.0".to_string()],
502 imports: imports.iter().map(|i| i.to_string()).collect(),
503 }
504 }
505
506 #[test]
507 fn stream_interface_refuses_and_lists_a_murmur_text_import() {
508 let c = importing(&[
509 "wasi:cli/environment@0.2.12",
510 "murmur:text/chunks@0.1.0",
511 "murmur:task/task@0.1.0",
512 ]);
513 let err = check_stream_interface("d", "1.0.0", Some(&c)).unwrap_err();
514 match &err {
515 RuntimeError::DriverStreamInterfaceNotServed {
516 name,
517 version,
518 imported,
519 expected,
520 } => {
521 assert_eq!((name.as_str(), version.as_str()), ("d", "1.0.0"));
522 assert_eq!(imported, &vec!["murmur:text/chunks@0.1.0".to_string()]);
523 assert_eq!(expected, crate::runtime::WIT_STREAM_EVENTS_IFACE);
524 }
525 other => panic!("unexpected error: {other:?}"),
526 }
527 let message = err.to_string();
528 assert!(message.contains("'d@1.0.0'"), "{message}");
529 assert!(message.contains("murmur:text/chunks@0.1.0"), "{message}");
530 assert!(message.contains("murmur:stream/events@0.1.0"), "{message}");
531 }
532
533 #[test]
534 fn stream_interface_refuses_another_version_of_murmur_stream() {
535 let c = importing(&["murmur:stream/events@0.2.0"]);
536 let err = check_stream_interface("d", "1.0.0", Some(&c)).unwrap_err();
537 assert!(matches!(
538 err,
539 RuntimeError::DriverStreamInterfaceNotServed { ref imported, .. }
540 if imported == &vec!["murmur:stream/events@0.2.0".to_string()]
541 ));
542 }
543
544 #[test]
545 fn stream_interface_accepts_exactly_the_served_instance() {
546 let c = importing(&["murmur:stream/events@0.1.0", "wasi:http/types@0.2.12"]);
547 assert!(check_stream_interface("d", "1.0.0", Some(&c)).is_ok());
548 }
549
550 #[test]
551 fn stream_interface_accepts_a_component_with_no_murmur_import() {
552 let c = importing(&["wasi:cli/environment@0.2.12"]);
553 assert!(check_stream_interface("d", "1.0.0", Some(&c)).is_ok());
554 assert!(check_stream_interface("d", "1.0.0", Some(&importing(&[]))).is_ok());
555 }
556
557 #[test]
558 fn stream_interface_accepts_no_contracts() {
559 assert!(check_stream_interface("d", "1.0.0", None).is_ok());
560 }
561
562 fn description(required_env: &[&str]) -> Description {
563 Description {
564 harness: "h".to_string(),
565 binary: "b".to_string(),
566 version_args: Vec::new(),
567 tested_versions: Vec::new(),
568 interrupt: InterruptMethod::Unsupported,
569 required_env: required_env.iter().map(|v| v.to_string()).collect(),
570 streams_text: false,
571 reports_usage: false,
572 }
573 }
574
575 #[test]
576 fn required_env_all_declared_passes() {
577 let allow = vec!["A".to_string(), "B".to_string(), "C".to_string()];
578 assert!(check_required_env("d", "1.0.0", &description(&["B", "A"]), &allow).is_ok());
579 }
580
581 #[test]
582 fn required_env_reports_missing_names_in_order_once_each() {
583 let allow = vec!["B".to_string()];
584 let err = check_required_env("d", "1.0.0", &description(&["Z", "B", "A", "Z"]), &allow)
585 .unwrap_err();
586 match &err {
587 RuntimeError::ProcessDriverRequiredEnvNotAllowed { missing, .. } => {
588 assert_eq!(missing, &vec!["Z".to_string(), "A".to_string()]);
589 }
590 other => panic!("unexpected error: {other:?}"),
591 }
592 let message = err.to_string();
593 assert!(message.contains("'d@1.0.0'"), "{message}");
594 assert!(message.contains("Z, A"), "{message}");
595 }
596
597 #[test]
598 fn required_env_usable_names_pass() {
599 assert_eq!(
600 unusable_env_name(&description(&["HOME", "PROFILE_2"])),
601 None
602 );
603 assert_eq!(unusable_env_name(&description(&[])), None);
604 }
605
606 #[test]
607 fn required_env_refuses_a_name_that_is_not_a_variable() {
608 for bad in ["", "A=B", "A\0B"] {
609 assert_eq!(
610 unusable_env_name(&description(&["HOME", bad, "TAIL"])),
611 Some(bad),
612 "{bad:?} should be refused"
613 );
614 }
615 }
616
617 #[test]
623 #[ignore = "a measurement, not an assertion; run with --release --ignored --nocapture"]
624 fn process_driver_parse_cost() {
625 use std::time::Instant;
626
627 const WASM: &[u8] = include_bytes!(concat!(
628 env!("CARGO_MANIFEST_DIR"),
629 "/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
630 ));
631 let engine = crate::runtime::build_engine().expect("engine");
632 let _ticker = EpochTicker::spawn(&engine);
633 let component = Component::new(&engine, WASM).expect("fixture compiles");
634 let rt = tokio::runtime::Builder::new_current_thread()
635 .enable_all()
636 .build()
637 .expect("runtime");
638
639 let instantiations = 50;
640 let start = Instant::now();
641 for _ in 0..instantiations {
642 rt.block_on(ProcessDriver::instantiate(
643 &engine, &component, "d", "0.1.0",
644 ))
645 .expect("instantiates");
646 }
647 let per_instantiation = start.elapsed() / instantiations;
648
649 let mut driver = rt
650 .block_on(ProcessDriver::instantiate(
651 &engine, &component, "d", "0.1.0",
652 ))
653 .expect("instantiates");
654 for lines in [1usize, 64] {
655 let batch: Vec<String> = (0..lines).map(|i| format!("text line {i}")).collect();
656 let runs = 2_000;
657 let mut samples = Vec::with_capacity(runs);
658 for _ in 0..runs {
659 let start = Instant::now();
660 let events = rt.block_on(driver.parse(batch.clone())).expect("parses");
661 samples.push(start.elapsed());
662 assert_eq!(events.len(), lines);
663 }
664 samples.sort_unstable();
665 let median = samples[samples.len() / 2];
666 let p95 = samples[samples.len() * 95 / 100];
667 crate::diagnostic::raw_to_stdout(&format!(
668 "parse {lines}-line batch on one reused instance: median {median:?}, p95 {p95:?} \
669 ({runs} calls)\n"
670 ));
671 }
672 crate::diagnostic::raw_to_stdout(&format!(
673 "instantiate one driver: {per_instantiation:?} (mean of {instantiations})\n"
674 ));
675 }
676
677 #[test]
678 fn process_driver_granted_nothing() {
679 let wat = r#"(component
680 (import "murmur:stream/events@0.1.0" (instance
681 (export "emit-chunk" (func (param "chunk" string)))
682 ))
683 )"#;
684 let bytes = wat::parse_str(wat).expect("component WAT parses");
685 let engine = crate::runtime::build_engine().expect("engine");
686 let _ticker = EpochTicker::spawn(&engine);
687 let component = Component::new(&engine, &bytes).expect("component compiles");
688 let rt = tokio::runtime::Builder::new_current_thread()
689 .enable_all()
690 .build()
691 .expect("runtime");
692 let result = rt.block_on(ProcessDriver::instantiate(
693 &engine, &component, "d", "1.0.0",
694 ));
695 match result {
696 Err(RuntimeError::ProcessDriverLoad { name, message, .. }) => {
697 assert_eq!(name, "d");
698 assert!(message.contains("murmur:stream/events@0.1.0"), "{message}");
699 }
700 Err(other) => panic!("unexpected error: {other:?}"),
701 Ok(_) => panic!("a driver importing a Murmur interface instantiated"),
702 }
703 }
704
705 #[test]
706 fn launch_and_parse_run_on_one_instance() {
707 const WASM: &[u8] = include_bytes!(concat!(
708 env!("CARGO_MANIFEST_DIR"),
709 "/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
710 ));
711 let request = LaunchRequest {
712 model: None,
713 system_prompt: String::new(),
714 config: None,
715 bridge: Some(Bridge {
716 server_name: "murmur".to_string(),
717 url: "http://127.0.0.1:1/mcp".to_string(),
718 bearer_token: "token".to_string(),
719 tool_names: vec!["python3".to_string()],
720 }),
721 session: Session {
722 id: "s1".to_string(),
723 mode: SessionMode::New,
724 },
725 harness_version: None,
726 task: "probe".to_string(),
727 };
728 let lines = vec![
729 r#"tool c1 murmur__python3 {"command":"x"}"#.to_string(),
730 "end done".to_string(),
731 ];
732 let (plan, events) =
733 launch_and_parse_process_driver_wasm(WASM, request, lines).expect("driver runs");
734 let plan = plan.expect("the fixture accepts a non-empty task");
735 assert!(plan
736 .env_set
737 .iter()
738 .any(|(name, value)| name == "FIXTURE_BRIDGE_TOKEN" && value == "token"));
739 match &events[..] {
742 [Event::ToolCall(call), Event::TurnEnd(summary)] => {
743 assert_eq!(call.name, "python3");
744 assert_eq!(call.input, r#"{"command":"x"}"#);
745 assert_eq!(summary, "done");
746 }
747 other => panic!("unexpected events: {other:?}"),
748 }
749 }
750}