1use murmur_artifact::{
9 current_platform, version_fit, FitSubject, VersionFit, WitContracts, WitLevel,
10};
11use wasmtime::component::{Component, Linker, ResourceTable};
12use wasmtime::{Engine, Store};
13use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView};
14
15use crate::bindings::process_driver::ProcessDriver as ProcessDriverBindings;
16use crate::errors::RuntimeError;
17use crate::limits::{
18 classify_guest_failure, CallDeadlines, ComponentRole, EpochTicker, ExecutionLimiter,
19 GuestFailure, WallDeadline,
20};
21use crate::ExecutionLimits;
22
23pub use crate::bindings::process_driver::exports::murmur::driver::process::{
24 Bridge, Description, DriverFile, Event, ExitStatus, FailureKind, InterruptMethod, LaunchPlan,
25 LaunchRequest, QuotaState, QuotaWindow, RetryInfo, Session, SessionInfo, SessionMode,
26 ToolCallInfo, ToolCallProgress, ToolCallStart, ToolResultInfo, TurnFailure, Usage,
27};
28
29impl QuotaState {
30 pub fn wire_name(self) -> &'static str {
32 match self {
33 QuotaState::Ok => "ok",
34 QuotaState::Warning => "warning",
35 QuotaState::Exhausted => "exhausted",
36 }
37 }
38}
39
40pub const PROCESS_DRIVER_IFACE: &str = "murmur:driver/process@0.4.0";
42
43const PROCESS_DRIVER_IFACE_PREFIX: &str = "murmur:driver/process@";
46
47pub fn check_driver_interface(
60 transport: &str,
61 name: &str,
62 version: &str,
63 contracts: Option<&WitContracts>,
64 level: &WitLevel,
65) -> Result<(), RuntimeError> {
66 let process_exports: Vec<String> = contracts
67 .map(|c| {
68 c.exports
69 .iter()
70 .filter(|export| export.starts_with(PROCESS_DRIVER_IFACE_PREFIX))
71 .cloned()
72 .collect()
73 })
74 .unwrap_or_default();
75 let expected = level_interface(level, "murmur:driver", "process")
76 .unwrap_or_else(|| PROCESS_DRIVER_IFACE.to_string());
77 match transport {
78 "process" if process_exports.contains(&expected) => Ok(()),
79 "process" if process_exports.is_empty() => {
80 Err(RuntimeError::ProcessDriverInterfaceMissing {
81 name: name.to_string(),
82 version: version.to_string(),
83 expected,
84 })
85 }
86 "process" => Err(
89 level_mismatch(name, version, level, process_exports, Vec::new()).unwrap_or_else(
90 || RuntimeError::ProcessDriverInterfaceMissing {
91 name: name.to_string(),
92 version: version.to_string(),
93 expected,
94 },
95 ),
96 ),
97 "http" if !process_exports.is_empty() => {
98 Err(RuntimeError::HttpDriverExportsProcessInterface {
99 name: name.to_string(),
100 version: version.to_string(),
101 exported: process_exports,
102 })
103 }
104 _ => Ok(()),
105 }
106}
107
108const RETIRED_TEXT_IFACE_PREFIX: &str = "murmur:text/";
111
112const STREAM_IFACE_PREFIX: &str = "murmur:stream/";
114
115pub fn check_stream_interface(
130 name: &str,
131 version: &str,
132 contracts: Option<&WitContracts>,
133 level: &WitLevel,
134) -> Result<(), RuntimeError> {
135 let expected = level_interface(level, "murmur:stream", "events")
136 .unwrap_or_else(|| crate::runtime::WIT_STREAM_EVENTS_IFACE.to_string());
137 let imported: Vec<String> = contracts
138 .map(|c| {
139 c.imports
140 .iter()
141 .filter(|import| {
142 import.starts_with(RETIRED_TEXT_IFACE_PREFIX)
143 || (import.starts_with(STREAM_IFACE_PREFIX) && **import != expected)
144 })
145 .cloned()
146 .collect()
147 })
148 .unwrap_or_default();
149 if imported.is_empty() {
150 return Ok(());
151 }
152 if let Some(mismatch) = level_mismatch(name, version, level, Vec::new(), imported.clone()) {
153 return Err(mismatch);
154 }
155 Err(RuntimeError::DriverStreamInterfaceNotServed {
156 name: name.to_string(),
157 version: version.to_string(),
158 imported,
159 expected,
160 })
161}
162
163pub(crate) fn driver_refusal(
167 stream: Result<(), RuntimeError>,
168 driver: Result<(), RuntimeError>,
169) -> Result<(), RuntimeError> {
170 match (stream, driver) {
171 (
172 Err(RuntimeError::WitLevelMismatch {
173 mismatched: imports,
174 ..
175 }),
176 Err(RuntimeError::WitLevelMismatch {
177 name,
178 version,
179 level,
180 mismatched: mut exports,
181 }),
182 ) => {
183 exports.extend(imports);
184 Err(RuntimeError::WitLevelMismatch {
185 name,
186 version,
187 level,
188 mismatched: exports,
189 })
190 }
191 (Err(_), Err(driver @ RuntimeError::WitLevelMismatch { .. })) => Err(driver),
192 (Err(stream), _) => Err(stream),
193 (Ok(()), driver) => driver,
194 }
195}
196
197fn level_interface(level: &WitLevel, package: &str, interface: &str) -> Option<String> {
199 level
200 .version_of(package)
201 .map(|version| format!("{package}/{interface}@{version}"))
202}
203
204pub(crate) fn check_component_level(
208 name: &str,
209 version: &str,
210 contracts: WitContracts,
211 level: &WitLevel,
212) -> Result<(), RuntimeError> {
213 match level_mismatch(name, version, level, contracts.exports, contracts.imports) {
214 Some(mismatch) => Err(mismatch),
215 None => Ok(()),
216 }
217}
218
219fn level_mismatch(
222 name: &str,
223 version: &str,
224 level: &WitLevel,
225 exports: Vec<String>,
226 imports: Vec<String>,
227) -> Option<RuntimeError> {
228 let subject = FitSubject::component(WitContracts { exports, imports });
229 match version_fit(&subject, level, current_platform()) {
231 VersionFit::Unserved(mismatched) => Some(RuntimeError::WitLevelMismatch {
232 name: name.to_string(),
233 version: version.to_string(),
234 level: level.level,
235 mismatched,
236 }),
237 VersionFit::Fits | VersionFit::NoPayloadFor(_) => None,
238 }
239}
240
241pub fn check_required_env(
246 name: &str,
247 version: &str,
248 description: &Description,
249 env_allow: &[String],
250) -> Result<(), RuntimeError> {
251 let mut missing: Vec<String> = Vec::new();
252 for var in &description.required_env {
253 if !env_allow.contains(var) && !missing.contains(var) {
254 missing.push(var.clone());
255 }
256 }
257 if missing.is_empty() {
258 Ok(())
259 } else {
260 Err(RuntimeError::ProcessDriverRequiredEnvNotAllowed {
261 name: name.to_string(),
262 version: version.to_string(),
263 missing,
264 })
265 }
266}
267
268struct ProcessDriverStoreState {
270 wasi: WasiCtx,
271 table: ResourceTable,
272 limits: ExecutionLimiter,
273}
274
275impl WasiView for ProcessDriverStoreState {
276 fn ctx(&mut self) -> WasiCtxView<'_> {
277 WasiCtxView {
278 ctx: &mut self.wasi,
279 table: &mut self.table,
280 }
281 }
282}
283
284pub struct ProcessDriver {
287 name: String,
288 version: String,
289 store: Store<ProcessDriverStoreState>,
290 bindings: ProcessDriverBindings,
291 call_deadlines: CallDeadlines,
293 cut: Option<(&'static str, u64)>,
296}
297
298impl ProcessDriver {
299 pub(crate) async fn instantiate(
309 engine: &Engine,
310 component: &Component,
311 name: &str,
312 version: &str,
313 call_deadlines: CallDeadlines,
314 ) -> Result<Self, RuntimeError> {
315 let load_error = |message: String| RuntimeError::ProcessDriverLoad {
316 name: name.to_string(),
317 version: version.to_string(),
318 message,
319 };
320 let limits = ExecutionLimits::default().for_hook_calls(None);
321
322 let mut linker: Linker<ProcessDriverStoreState> = Linker::new(engine);
323 wasmtime_wasi::p2::add_to_linker_async(&mut linker)
324 .map_err(|e| load_error(e.to_string()))?;
325
326 let state = ProcessDriverStoreState {
327 wasi: WasiCtxBuilder::new().build(),
330 table: ResourceTable::new(),
331 limits: limits.limiter(),
332 };
333 let mut store = Store::new(engine, state);
334 store.limiter(|state| &mut state.limits);
335 store.set_epoch_deadline(limits.deadline_ticks());
336
337 let bindings = call_deadlines
338 .bound(
339 name,
340 ComponentRole::ProcessDriver,
341 "instantiate",
342 &limits,
343 ProcessDriverBindings::instantiate_async(&mut store, component, &linker),
344 )
345 .await
346 .map_err(|_| load_error(format!("instantiation {}", wall_deadline_text(&limits))))?
347 .map_err(|e| load_error(format!("{e:#}")))?;
348 Ok(Self {
349 name: name.to_string(),
350 version: version.to_string(),
351 store,
352 bindings,
353 call_deadlines,
354 cut: None,
355 })
356 }
357
358 pub async fn describe(&mut self) -> Result<Description, RuntimeError> {
363 self.enter("describe")
364 .map_err(|message| self.load_error(message))?;
365 let limits = self.reset_deadline();
366 let called = self
367 .call_deadlines
368 .bound(
369 &self.name,
370 ComponentRole::ProcessDriver,
371 "describe",
372 &limits,
373 self.bindings
374 .murmur_driver_process()
375 .call_describe(&mut self.store),
376 )
377 .await;
378 let description = self
379 .settle("describe", &limits, called)
380 .map_err(|message| self.load_error(format!("describe() {message}")))?
381 .map_err(|e| {
382 let message = match classify_guest_failure(&e, &self.store.data().limits) {
383 GuestFailure::DeadlineExceeded { seconds } => {
384 format!("describe() exceeded its {seconds}s deadline and was interrupted")
385 }
386 GuestFailure::ResourceLimit { message } => {
387 format!("describe() exceeded its resource limits: {message}")
388 }
389 GuestFailure::Other => format!("describe() trapped: {e:#}"),
390 };
391 self.load_error(message)
392 })?;
393 if let Some(bad) = unusable_env_name(&description) {
394 return Err(self.load_error(format!(
395 "describe() names {bad:?} in required-env, which is not a usable variable name"
396 )));
397 }
398 Ok(description)
399 }
400
401 pub async fn launch(
404 &mut self,
405 request: LaunchRequest,
406 ) -> Result<Result<LaunchPlan, String>, RuntimeError> {
407 self.enter("launch")
408 .map_err(|message| self.failed_in("launch", message))?;
409 let limits = self.reset_deadline();
410 let called = self
411 .call_deadlines
412 .bound(
413 &self.name,
414 ComponentRole::ProcessDriver,
415 "launch",
416 &limits,
417 self.bindings
418 .murmur_driver_process()
419 .call_launch(&mut self.store, &request),
420 )
421 .await;
422 self.settle("launch", &limits, called)
423 .map_err(|message| self.failed_in("launch", message))?
424 .map_err(|e| self.call_error("launch", &e))
425 }
426
427 pub async fn parse(&mut self, lines: Vec<String>) -> Result<Vec<Event>, RuntimeError> {
430 self.enter("parse")
431 .map_err(|message| self.failed_in("parse", message))?;
432 let limits = self.reset_deadline();
433 let called = self
434 .call_deadlines
435 .bound(
436 &self.name,
437 ComponentRole::ProcessDriver,
438 "parse",
439 &limits,
440 self.bindings
441 .murmur_driver_process()
442 .call_parse(&mut self.store, &lines),
443 )
444 .await;
445 self.settle("parse", &limits, called)
446 .map_err(|message| self.failed_in("parse", message))?
447 .map_err(|e| self.call_error("parse", &e))
448 }
449
450 pub async fn classify_exit(&mut self, exit: ExitStatus) -> Result<Event, RuntimeError> {
452 self.enter("classify-exit")
453 .map_err(|message| self.failed_in("classify-exit", message))?;
454 let limits = self.reset_deadline();
455 let called = self
456 .call_deadlines
457 .bound(
458 &self.name,
459 ComponentRole::ProcessDriver,
460 "classify-exit",
461 &limits,
462 self.bindings
463 .murmur_driver_process()
464 .call_classify_exit(&mut self.store, &exit),
465 )
466 .await;
467 self.settle("classify-exit", &limits, called)
468 .map_err(|message| self.failed_in("classify-exit", message))?
469 .map_err(|e| self.call_error("classify-exit", &e))
470 }
471
472 fn reset_deadline(&mut self) -> ExecutionLimits {
476 let limits = self.store.data().limits.limits();
477 self.store.set_epoch_deadline(limits.deadline_ticks());
478 limits
479 }
480
481 fn enter(&self, call: &str) -> Result<(), String> {
483 match self.cut {
484 None => Ok(()),
485 Some((earlier, seconds)) => Err(format!(
486 "{call} was not called: the earlier {earlier} call was stopped at its {seconds}s \
487 deadline, and a stopped driver is not entered again"
488 )),
489 }
490 }
491
492 fn settle<T>(
495 &mut self,
496 call: &'static str,
497 limits: &ExecutionLimits,
498 called: Result<T, WallDeadline>,
499 ) -> Result<T, String> {
500 called.map_err(|_| {
501 self.cut = Some((call, limits.deadline_seconds));
502 wall_deadline_text(limits)
503 })
504 }
505
506 fn failed_in(&self, call: &str, message: String) -> RuntimeError {
507 RuntimeError::ProcessDriverCallFailed {
508 name: self.name.clone(),
509 version: self.version.clone(),
510 call: call.to_string(),
511 message,
512 }
513 }
514
515 fn call_error(&self, call: &str, error: &wasmtime::Error) -> RuntimeError {
516 let message = match classify_guest_failure(error, &self.store.data().limits) {
517 GuestFailure::DeadlineExceeded { seconds } => {
518 format!("exceeded its {seconds}s deadline and was interrupted")
519 }
520 GuestFailure::ResourceLimit { message } => {
521 format!("exceeded its resource limits: {message}")
522 }
523 GuestFailure::Other => format!("trapped: {error:#}"),
524 };
525 self.failed_in(call, message)
526 }
527
528 fn load_error(&self, message: String) -> RuntimeError {
529 RuntimeError::ProcessDriverLoad {
530 name: self.name.clone(),
531 version: self.version.clone(),
532 message,
533 }
534 }
535}
536
537fn wall_deadline_text(limits: &ExecutionLimits) -> String {
541 format!(
542 "waited on the host past its {}s deadline and was stopped",
543 limits.deadline_seconds
544 )
545}
546
547pub(crate) fn is_usable_env_name(name: &str) -> bool {
554 !name.is_empty() && !name.contains('=') && !name.contains('\0')
555}
556
557fn unusable_env_name(description: &Description) -> Option<&str> {
560 description
561 .required_env
562 .iter()
563 .find(|var| !is_usable_env_name(var))
564 .map(String::as_str)
565}
566
567const STANDALONE_NAME: &str = "<wasm>";
570const STANDALONE_VERSION: &str = "";
571
572struct StandaloneDriver {
575 engine: Engine,
576 component: Component,
577 rt: tokio::runtime::Runtime,
578 _ticker: EpochTicker,
579}
580
581impl StandaloneDriver {
582 fn compile(wasm: &[u8]) -> Result<Self, RuntimeError> {
583 let engine = crate::runtime::build_engine()?;
584 let ticker = EpochTicker::spawn(&engine);
585 let component =
586 Component::new(&engine, wasm).map_err(|e| RuntimeError::ToolComponentCompile {
587 name: STANDALONE_NAME.to_string(),
588 version: STANDALONE_VERSION.to_string(),
589 message: e.to_string(),
590 })?;
591 let rt = tokio::runtime::Builder::new_current_thread()
592 .enable_all()
593 .build()
594 .map_err(|e| {
595 RuntimeError::Runtime(format!("failed to build process driver runtime: {e}"))
596 })?;
597 Ok(StandaloneDriver {
598 engine,
599 component,
600 rt,
601 _ticker: ticker,
602 })
603 }
604
605 async fn instantiate(&self) -> Result<ProcessDriver, RuntimeError> {
606 ProcessDriver::instantiate(
607 &self.engine,
608 &self.component,
609 STANDALONE_NAME,
610 STANDALONE_VERSION,
611 CallDeadlines::default(),
613 )
614 .await
615 }
616}
617
618pub fn describe_process_driver_wasm(wasm: &[u8]) -> Result<Description, RuntimeError> {
625 let driver = StandaloneDriver::compile(wasm)?;
626 driver.rt.block_on(async {
627 let mut instance = driver.instantiate().await?;
628 instance.describe().await
629 })
630}
631
632pub fn launch_and_parse_process_driver_wasm(
639 wasm: &[u8],
640 request: LaunchRequest,
641 lines: Vec<String>,
642) -> Result<(Result<LaunchPlan, String>, Vec<Event>), RuntimeError> {
643 let driver = StandaloneDriver::compile(wasm)?;
644 driver.rt.block_on(async {
645 let mut instance = driver.instantiate().await?;
646 let plan = instance.launch(request).await?;
647 let events = instance.parse(lines).await?;
648 Ok((plan, events))
649 })
650}
651
652#[cfg(test)]
653mod tests {
654 use murmur_artifact::ContractDirection;
655
656 use super::*;
657
658 fn served() -> &'static WitLevel {
659 crate::served_wit_level()
660 }
661
662 fn contracts(exports: &[&str]) -> WitContracts {
663 WitContracts {
664 exports: exports.iter().map(|e| e.to_string()).collect(),
665 imports: Vec::new(),
666 }
667 }
668
669 #[test]
670 fn process_driver_interface_process_transport_accepts_the_process_export() {
671 let c = contracts(&[PROCESS_DRIVER_IFACE]);
672 assert!(check_driver_interface("process", "d", "1.0.0", Some(&c), served()).is_ok());
673 }
674
675 #[test]
676 fn process_driver_interface_process_transport_refuses_a_tool_export() {
677 let c = contracts(&["murmur:tool/run@0.2.0"]);
678 let err = check_driver_interface("process", "d", "1.0.0", Some(&c), served()).unwrap_err();
679 match &err {
680 RuntimeError::ProcessDriverInterfaceMissing { expected, .. } => {
681 assert_eq!(expected, PROCESS_DRIVER_IFACE);
682 }
683 other => panic!("unexpected error: {other:?}"),
684 }
685 let message = err.to_string();
686 assert!(message.contains("'d@1.0.0'"), "{message}");
687 assert!(message.contains(PROCESS_DRIVER_IFACE), "{message}");
688 }
689
690 #[test]
691 fn process_driver_interface_process_transport_refuses_another_version_as_a_level_mismatch() {
692 let c = contracts(&["murmur:driver/process@0.0.1"]);
693 let err = check_driver_interface("process", "d", "1.0.0", Some(&c), served()).unwrap_err();
694 match &err {
695 RuntimeError::WitLevelMismatch {
696 level, mismatched, ..
697 } => {
698 assert_eq!(*level, served().level);
699 assert_eq!(mismatched.len(), 1);
700 assert_eq!(mismatched[0].interface, "murmur:driver/process@0.0.1");
701 assert_eq!(mismatched[0].direction, ContractDirection::Export);
702 }
703 other => panic!("unexpected error: {other:?}"),
704 }
705 assert_eq!(
706 err.to_string(),
707 format!(
708 "artifact 'd@1.0.0' does not fit WIT level {0}, the level this mur serves: it \
709 exports murmur:driver/process@0.0.1, and level {0} has murmur:driver@{1}",
710 served().level,
711 served().version_of("murmur:driver").unwrap()
712 )
713 );
714 }
715
716 fn level_1() -> &'static WitLevel {
719 &crate::wit_versions::wit_levels()[0]
720 }
721
722 #[test]
723 fn process_driver_interface_follows_the_level_it_is_given() {
724 let c = contracts(&["murmur:driver/process@0.4.0"]);
725 assert!(check_driver_interface("process", "d", "1.0.0", Some(&c), served()).is_ok());
726 match check_driver_interface("process", "d", "1.0.0", Some(&c), level_1()).unwrap_err() {
727 RuntimeError::WitLevelMismatch {
728 level, mismatched, ..
729 } => {
730 assert_eq!(level, 1);
731 assert_eq!(mismatched[0].direction, ContractDirection::Export);
732 assert_eq!(mismatched[0].package, "murmur:driver");
733 assert_eq!(mismatched[0].served_version.as_deref(), Some("0.3.0"));
734 }
735 other => panic!("unexpected error: {other:?}"),
736 }
737 let old = contracts(&["murmur:driver/process@0.3.0"]);
738 assert!(check_driver_interface("process", "d", "1.0.0", Some(&old), level_1()).is_ok());
739 let stream = importing(&["murmur:stream/events@0.1.0"]);
740 assert!(check_stream_interface("d", "1.0.0", Some(&stream), level_1()).is_ok());
741 }
742
743 #[test]
744 fn process_driver_interface_process_transport_refuses_no_contracts() {
745 let err = check_driver_interface("process", "d", "1.0.0", None, served()).unwrap_err();
746 assert!(matches!(
747 err,
748 RuntimeError::ProcessDriverInterfaceMissing { .. }
749 ));
750 }
751
752 #[test]
753 fn process_driver_interface_http_transport_refuses_the_process_export() {
754 let c = contracts(&[PROCESS_DRIVER_IFACE]);
755 let err = check_driver_interface("http", "d", "1.0.0", Some(&c), served()).unwrap_err();
756 match &err {
757 RuntimeError::HttpDriverExportsProcessInterface { name, exported, .. } => {
758 assert_eq!(name, "d");
759 assert_eq!(exported, &vec![PROCESS_DRIVER_IFACE.to_string()]);
760 }
761 other => panic!("unexpected error: {other:?}"),
762 }
763 }
764
765 #[test]
766 fn process_driver_interface_http_transport_accepts_a_tool_export() {
767 let c = contracts(&["murmur:tool/run@0.2.0"]);
768 assert!(check_driver_interface("http", "d", "1.0.0", Some(&c), served()).is_ok());
769 }
770
771 fn importing(imports: &[&str]) -> WitContracts {
772 WitContracts {
773 exports: vec!["murmur:tool/run@0.2.0".to_string()],
774 imports: imports.iter().map(|i| i.to_string()).collect(),
775 }
776 }
777
778 #[test]
779 fn stream_interface_refuses_and_lists_a_murmur_text_import() {
780 let c = importing(&[
781 "wasi:cli/environment@0.2.12",
782 "murmur:text/chunks@0.1.0",
783 "murmur:task/task@0.1.0",
784 ]);
785 let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
786 match &err {
787 RuntimeError::WitLevelMismatch {
788 name,
789 version,
790 mismatched,
791 ..
792 } => {
793 assert_eq!((name.as_str(), version.as_str()), ("d", "1.0.0"));
794 assert_eq!(mismatched.len(), 1);
795 assert_eq!(mismatched[0].interface, "murmur:text/chunks@0.1.0");
796 assert_eq!(mismatched[0].served_version, None);
797 }
798 other => panic!("unexpected error: {other:?}"),
799 }
800 let message = err.to_string();
801 assert!(message.contains("'d@1.0.0'"), "{message}");
802 assert!(
803 message.contains(
804 "it imports murmur:text/chunks@0.1.0, and level 3 has no version of murmur:text"
805 ),
806 "{message}"
807 );
808 }
809
810 #[test]
811 fn stream_interface_refuses_another_version_of_murmur_stream() {
812 let c = importing(&["murmur:stream/events@0.2.0"]);
813 let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
814 assert!(matches!(
815 err,
816 RuntimeError::WitLevelMismatch { ref mismatched, .. }
817 if mismatched.len() == 1 && mismatched[0].interface == "murmur:stream/events@0.2.0"
818 ));
819 }
820
821 #[test]
824 fn stream_interface_refuses_an_unknown_interface_at_the_served_version_as_not_served() {
825 let c = importing(&["murmur:stream/other@0.1.0"]);
826 let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
827 match &err {
828 RuntimeError::DriverStreamInterfaceNotServed {
829 imported, expected, ..
830 } => {
831 assert_eq!(imported, &vec!["murmur:stream/other@0.1.0".to_string()]);
832 assert_eq!(expected, crate::runtime::WIT_STREAM_EVENTS_IFACE);
833 }
834 other => panic!("unexpected error: {other:?}"),
835 }
836 let both = importing(&["murmur:stream/other@0.1.0", "murmur:text/chunks@0.1.0"]);
838 assert!(matches!(
839 check_stream_interface("d", "1.0.0", Some(&both), served()).unwrap_err(),
840 RuntimeError::WitLevelMismatch { ref mismatched, .. }
841 if mismatched.len() == 1 && mismatched[0].interface == "murmur:text/chunks@0.1.0"
842 ));
843 }
844
845 #[test]
846 fn driver_refusal_reports_a_level_mismatch_over_any_other() {
847 let not_served = || {
848 check_stream_interface(
849 "d",
850 "1.0.0",
851 Some(&importing(&["murmur:stream/other@0.1.0"])),
852 served(),
853 )
854 };
855 let old_export = || {
856 check_driver_interface(
857 "process",
858 "d",
859 "1.0.0",
860 Some(&contracts(&["murmur:driver/process@0.2.0"])),
861 served(),
862 )
863 };
864 let text_import = || {
865 check_stream_interface(
866 "d",
867 "1.0.0",
868 Some(&importing(&["murmur:text/chunks@0.1.0"])),
869 served(),
870 )
871 };
872 let no_export = || check_driver_interface("process", "d", "1.0.0", None, served());
873
874 assert!(matches!(
875 driver_refusal(not_served(), old_export()),
876 Err(RuntimeError::WitLevelMismatch { .. })
877 ));
878 assert!(matches!(
879 driver_refusal(text_import(), no_export()),
880 Err(RuntimeError::WitLevelMismatch { .. })
881 ));
882 assert!(matches!(
883 driver_refusal(not_served(), no_export()),
884 Err(RuntimeError::DriverStreamInterfaceNotServed { .. })
885 ));
886 match driver_refusal(text_import(), old_export()) {
887 Err(RuntimeError::WitLevelMismatch { mismatched, .. }) => {
888 let names: Vec<&str> = mismatched.iter().map(|m| m.interface.as_str()).collect();
889 assert_eq!(
890 names,
891 ["murmur:driver/process@0.2.0", "murmur:text/chunks@0.1.0"]
892 );
893 }
894 other => panic!("unexpected result: {other:?}"),
895 }
896 assert!(driver_refusal(Ok(()), Ok(())).is_ok());
897 }
898
899 #[test]
900 fn stream_interface_accepts_exactly_the_served_instance() {
901 let c = importing(&["murmur:stream/events@0.1.0", "wasi:http/types@0.2.12"]);
902 assert!(check_stream_interface("d", "1.0.0", Some(&c), served()).is_ok());
903 }
904
905 #[test]
906 fn stream_interface_accepts_a_component_with_no_murmur_import() {
907 let c = importing(&["wasi:cli/environment@0.2.12"]);
908 assert!(check_stream_interface("d", "1.0.0", Some(&c), served()).is_ok());
909 assert!(check_stream_interface("d", "1.0.0", Some(&importing(&[])), served()).is_ok());
910 }
911
912 #[test]
913 fn stream_interface_accepts_no_contracts() {
914 assert!(check_stream_interface("d", "1.0.0", None, served()).is_ok());
915 }
916
917 fn description(required_env: &[&str]) -> Description {
918 Description {
919 harness: "h".to_string(),
920 binary: "b".to_string(),
921 version_args: Vec::new(),
922 tested_versions: Vec::new(),
923 interrupt: InterruptMethod::Unsupported,
924 required_env: required_env.iter().map(|v| v.to_string()).collect(),
925 streams_text: false,
926 reports_usage: false,
927 }
928 }
929
930 #[test]
931 fn required_env_all_declared_passes() {
932 let allow = vec!["A".to_string(), "B".to_string(), "C".to_string()];
933 assert!(check_required_env("d", "1.0.0", &description(&["B", "A"]), &allow).is_ok());
934 }
935
936 #[test]
937 fn required_env_reports_missing_names_in_order_once_each() {
938 let allow = vec!["B".to_string()];
939 let err = check_required_env("d", "1.0.0", &description(&["Z", "B", "A", "Z"]), &allow)
940 .unwrap_err();
941 match &err {
942 RuntimeError::ProcessDriverRequiredEnvNotAllowed { missing, .. } => {
943 assert_eq!(missing, &vec!["Z".to_string(), "A".to_string()]);
944 }
945 other => panic!("unexpected error: {other:?}"),
946 }
947 let message = err.to_string();
948 assert!(message.contains("'d@1.0.0'"), "{message}");
949 assert!(message.contains("Z, A"), "{message}");
950 }
951
952 #[test]
953 fn required_env_usable_names_pass() {
954 assert_eq!(
955 unusable_env_name(&description(&["HOME", "PROFILE_2"])),
956 None
957 );
958 assert_eq!(unusable_env_name(&description(&[])), None);
959 }
960
961 #[test]
962 fn required_env_refuses_a_name_that_is_not_a_variable() {
963 for bad in ["", "A=B", "A\0B"] {
964 assert_eq!(
965 unusable_env_name(&description(&["HOME", bad, "TAIL"])),
966 Some(bad),
967 "{bad:?} should be refused"
968 );
969 }
970 }
971
972 #[test]
978 #[ignore = "a measurement, not an assertion; run with --release --ignored --nocapture"]
979 fn process_driver_parse_cost() {
980 use std::time::Instant;
981
982 const WASM: &[u8] = include_bytes!(concat!(
983 env!("CARGO_MANIFEST_DIR"),
984 "/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
985 ));
986 let engine = crate::runtime::build_engine().expect("engine");
987 let _ticker = EpochTicker::spawn(&engine);
988 let component = Component::new(&engine, WASM).expect("fixture compiles");
989 let rt = tokio::runtime::Builder::new_current_thread()
990 .enable_all()
991 .build()
992 .expect("runtime");
993
994 let instantiations = 50;
995 let start = Instant::now();
996 for _ in 0..instantiations {
997 rt.block_on(ProcessDriver::instantiate(
998 &engine,
999 &component,
1000 "d",
1001 "0.1.0",
1002 CallDeadlines::default(),
1003 ))
1004 .expect("instantiates");
1005 }
1006 let per_instantiation = start.elapsed() / instantiations;
1007
1008 let mut driver = rt
1009 .block_on(ProcessDriver::instantiate(
1010 &engine,
1011 &component,
1012 "d",
1013 "0.1.0",
1014 CallDeadlines::default(),
1015 ))
1016 .expect("instantiates");
1017 for lines in [1usize, 64] {
1018 let batch: Vec<String> = (0..lines).map(|i| format!("text line {i}")).collect();
1019 let runs = 2_000;
1020 let mut samples = Vec::with_capacity(runs);
1021 for _ in 0..runs {
1022 let start = Instant::now();
1023 let events = rt.block_on(driver.parse(batch.clone())).expect("parses");
1024 samples.push(start.elapsed());
1025 assert_eq!(events.len(), lines);
1026 }
1027 samples.sort_unstable();
1028 let median = samples[samples.len() / 2];
1029 let p95 = samples[samples.len() * 95 / 100];
1030 crate::diagnostic::raw_to_stdout(&format!(
1031 "parse {lines}-line batch on one reused instance: median {median:?}, p95 {p95:?} \
1032 ({runs} calls)\n"
1033 ));
1034 }
1035 crate::diagnostic::raw_to_stdout(&format!(
1036 "instantiate one driver: {per_instantiation:?} (mean of {instantiations})\n"
1037 ));
1038 }
1039
1040 #[test]
1041 fn process_driver_granted_nothing() {
1042 let wat = r#"(component
1043 (import "murmur:stream/events@0.1.0" (instance
1044 (export "emit-chunk" (func (param "chunk" string)))
1045 ))
1046 )"#;
1047 let bytes = wat::parse_str(wat).expect("component WAT parses");
1048 let engine = crate::runtime::build_engine().expect("engine");
1049 let _ticker = EpochTicker::spawn(&engine);
1050 let component = Component::new(&engine, &bytes).expect("component compiles");
1051 let rt = tokio::runtime::Builder::new_current_thread()
1052 .enable_all()
1053 .build()
1054 .expect("runtime");
1055 let result = rt.block_on(ProcessDriver::instantiate(
1056 &engine,
1057 &component,
1058 "d",
1059 "1.0.0",
1060 CallDeadlines::default(),
1061 ));
1062 match result {
1063 Err(RuntimeError::ProcessDriverLoad { name, message, .. }) => {
1064 assert_eq!(name, "d");
1065 assert!(message.contains("murmur:stream/events@0.1.0"), "{message}");
1066 }
1067 Err(other) => panic!("unexpected error: {other:?}"),
1068 Ok(_) => panic!("a driver importing a Murmur interface instantiated"),
1069 }
1070 }
1071
1072 #[test]
1073 fn launch_and_parse_run_on_one_instance() {
1074 const WASM: &[u8] = include_bytes!(concat!(
1075 env!("CARGO_MANIFEST_DIR"),
1076 "/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
1077 ));
1078 let request = LaunchRequest {
1079 model: None,
1080 system_prompt: String::new(),
1081 config: None,
1082 bridge: Some(Bridge {
1083 server_name: "murmur".to_string(),
1084 url: "http://127.0.0.1:1/mcp".to_string(),
1085 bearer_token: "token".to_string(),
1086 tool_names: vec!["python3".to_string()],
1087 }),
1088 session: Session {
1089 id: "s1".to_string(),
1090 mode: SessionMode::New,
1091 },
1092 harness_version: None,
1093 task: "probe".to_string(),
1094 };
1095 let lines = vec![
1096 r#"tool c1 murmur__python3 {"command":"x"}"#.to_string(),
1097 "end done".to_string(),
1098 ];
1099 let (plan, events) =
1100 launch_and_parse_process_driver_wasm(WASM, request, lines).expect("driver runs");
1101 let plan = plan.expect("the fixture accepts a non-empty task");
1102 assert!(plan
1103 .env_set
1104 .iter()
1105 .any(|(name, value)| name == "FIXTURE_BRIDGE_TOKEN" && value == "token"));
1106 match &events[..] {
1109 [Event::ToolCall(call), Event::TurnEnd(summary)] => {
1110 assert_eq!(call.name, "python3");
1111 assert_eq!(call.input, r#"{"command":"x"}"#);
1112 assert_eq!(summary, "done");
1113 }
1114 other => panic!("unexpected events: {other:?}"),
1115 }
1116 }
1117}