1use murmur_artifact::{version_fit, FitSubject, VersionFit, WitContracts, WitLevel};
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::{
16 classify_guest_failure, CallDeadlines, ComponentRole, EpochTicker, ExecutionLimiter,
17 GuestFailure, WallDeadline,
18};
19use crate::ExecutionLimits;
20
21pub use crate::bindings::process_driver::exports::murmur::driver::process::{
22 Bridge, Description, DriverFile, Event, ExitStatus, FailureKind, HarnessNeeds, InterruptMethod,
23 LaunchPlan, LaunchRequest, LoginCheck, LoginOutput, LoginState, PathAccess, PathNeed,
24 QuotaState, QuotaWindow, RetryInfo, Session, SessionInfo, SessionMode, ToolCallInfo,
25 ToolCallProgress, ToolCallStart, ToolResultInfo, TurnFailure, Usage,
26};
27
28impl QuotaState {
29 pub fn wire_name(self) -> &'static str {
31 match self {
32 QuotaState::Ok => "ok",
33 QuotaState::Warning => "warning",
34 QuotaState::Exhausted => "exhausted",
35 }
36 }
37}
38
39pub const PROCESS_DRIVER_IFACE: &str = "murmur:driver/process@0.6.0";
41
42const PROCESS_DRIVER_IFACE_PREFIX: &str = "murmur:driver/process@";
45
46pub fn check_driver_interface(
59 transport: &str,
60 name: &str,
61 version: &str,
62 contracts: Option<&WitContracts>,
63 level: &WitLevel,
64) -> Result<(), RuntimeError> {
65 let process_exports: Vec<String> = contracts
66 .map(|c| {
67 c.exports
68 .iter()
69 .filter(|export| export.starts_with(PROCESS_DRIVER_IFACE_PREFIX))
70 .cloned()
71 .collect()
72 })
73 .unwrap_or_default();
74 let expected = level_interface(level, "murmur:driver", "process")
75 .unwrap_or_else(|| PROCESS_DRIVER_IFACE.to_string());
76 match transport {
77 "process" if process_exports.contains(&expected) => Ok(()),
78 "process" if process_exports.is_empty() => {
79 Err(RuntimeError::ProcessDriverInterfaceMissing {
80 name: name.to_string(),
81 version: version.to_string(),
82 expected,
83 })
84 }
85 "process" => Err(
88 level_mismatch(name, version, level, process_exports, Vec::new()).unwrap_or_else(
89 || RuntimeError::ProcessDriverInterfaceMissing {
90 name: name.to_string(),
91 version: version.to_string(),
92 expected,
93 },
94 ),
95 ),
96 "http" if !process_exports.is_empty() => {
97 Err(RuntimeError::HttpDriverExportsProcessInterface {
98 name: name.to_string(),
99 version: version.to_string(),
100 exported: process_exports,
101 })
102 }
103 _ => Ok(()),
104 }
105}
106
107const RETIRED_TEXT_IFACE_PREFIX: &str = "murmur:text/";
110
111const STREAM_IFACE_PREFIX: &str = "murmur:stream/";
113
114pub fn check_stream_interface(
129 name: &str,
130 version: &str,
131 contracts: Option<&WitContracts>,
132 level: &WitLevel,
133) -> Result<(), RuntimeError> {
134 let expected = level_interface(level, "murmur:stream", "events")
135 .unwrap_or_else(|| crate::runtime::WIT_STREAM_EVENTS_IFACE.to_string());
136 let imported: Vec<String> = contracts
137 .map(|c| {
138 c.imports
139 .iter()
140 .filter(|import| {
141 import.starts_with(RETIRED_TEXT_IFACE_PREFIX)
142 || (import.starts_with(STREAM_IFACE_PREFIX) && **import != expected)
143 })
144 .cloned()
145 .collect()
146 })
147 .unwrap_or_default();
148 if imported.is_empty() {
149 return Ok(());
150 }
151 if let Some(mismatch) = level_mismatch(name, version, level, Vec::new(), imported.clone()) {
152 return Err(mismatch);
153 }
154 Err(RuntimeError::DriverStreamInterfaceNotServed {
155 name: name.to_string(),
156 version: version.to_string(),
157 imported,
158 expected,
159 })
160}
161
162pub(crate) fn driver_refusal(
166 stream: Result<(), RuntimeError>,
167 driver: Result<(), RuntimeError>,
168) -> Result<(), RuntimeError> {
169 match (stream, driver) {
170 (
171 Err(RuntimeError::WitLevelMismatch {
172 mismatched: imports,
173 ..
174 }),
175 Err(RuntimeError::WitLevelMismatch {
176 name,
177 version,
178 level,
179 mismatched: mut exports,
180 }),
181 ) => {
182 exports.extend(imports);
183 Err(RuntimeError::WitLevelMismatch {
184 name,
185 version,
186 level,
187 mismatched: exports,
188 })
189 }
190 (Err(_), Err(driver @ RuntimeError::WitLevelMismatch { .. })) => Err(driver),
191 (Err(stream), _) => Err(stream),
192 (Ok(()), driver) => driver,
193 }
194}
195
196fn level_interface(level: &WitLevel, package: &str, interface: &str) -> Option<String> {
198 level
199 .version_of(package)
200 .map(|version| format!("{package}/{interface}@{version}"))
201}
202
203pub(crate) fn check_component_level(
207 name: &str,
208 version: &str,
209 contracts: WitContracts,
210 level: &WitLevel,
211) -> Result<(), RuntimeError> {
212 match level_mismatch(name, version, level, contracts.exports, contracts.imports) {
213 Some(mismatch) => Err(mismatch),
214 None => Ok(()),
215 }
216}
217
218fn level_mismatch(
221 name: &str,
222 version: &str,
223 level: &WitLevel,
224 exports: Vec<String>,
225 imports: Vec<String>,
226) -> Option<RuntimeError> {
227 let subject = FitSubject::component(WitContracts { exports, imports });
228 match version_fit(&subject, level) {
229 VersionFit::Unserved(mismatched) => Some(RuntimeError::WitLevelMismatch {
230 name: name.to_string(),
231 version: version.to_string(),
232 level: level.level,
233 mismatched,
234 }),
235 VersionFit::Fits => None,
236 }
237}
238
239pub fn check_required_env(
244 name: &str,
245 version: &str,
246 description: &Description,
247 env_allow: &[String],
248) -> Result<(), RuntimeError> {
249 let mut missing: Vec<String> = Vec::new();
250 for var in &description.required_env {
251 if !env_allow.contains(var) && !missing.contains(var) {
252 missing.push(var.clone());
253 }
254 }
255 if missing.is_empty() {
256 Ok(())
257 } else {
258 Err(RuntimeError::ProcessDriverRequiredEnvNotAllowed {
259 name: name.to_string(),
260 version: version.to_string(),
261 missing,
262 })
263 }
264}
265
266struct ProcessDriverStoreState {
268 wasi: WasiCtx,
269 table: ResourceTable,
270 limits: ExecutionLimiter,
271}
272
273impl WasiView for ProcessDriverStoreState {
274 fn ctx(&mut self) -> WasiCtxView<'_> {
275 WasiCtxView {
276 ctx: &mut self.wasi,
277 table: &mut self.table,
278 }
279 }
280}
281
282pub struct ProcessDriver {
285 name: String,
286 version: String,
287 body: DriverBody,
288 call_deadlines: CallDeadlines,
290 cut: Option<(&'static str, u64)>,
293}
294
295enum DriverBody {
298 Local {
299 store: Store<ProcessDriverStoreState>,
300 bindings: ProcessDriverBindings,
301 },
302 Remote(crate::guest_worker::RemoteProcessDriver),
303}
304
305#[derive(Debug, serde::Serialize, serde::Deserialize)]
307pub(crate) enum DriverCall {
308 Describe,
309 Launch(LaunchRequest),
310 Parse(Vec<String>),
311 ClassifyExit(ExitStatus),
312 ReadLogin(LoginOutput),
313}
314
315#[derive(Debug, serde::Serialize, serde::Deserialize)]
317pub(crate) enum DriverAnswer {
318 Opened,
319 Description(Description),
320 Launch(Result<LaunchPlan, String>),
321 Events(Vec<Event>),
322 Event(Event),
323 Login(LoginState),
324}
325
326#[derive(Debug, serde::Serialize, serde::Deserialize)]
329pub(crate) enum DriverError {
330 Load(String),
331 CallFailed { call: String, message: String },
332 Other(String),
333}
334
335impl DriverError {
336 pub(crate) fn from_runtime(error: RuntimeError) -> Self {
337 match error {
338 RuntimeError::ProcessDriverLoad { message, .. } => Self::Load(message),
339 RuntimeError::ProcessDriverCallFailed { call, message, .. } => {
340 Self::CallFailed { call, message }
341 }
342 other => Self::Other(other.to_string()),
343 }
344 }
345
346 pub(crate) fn into_runtime(self, name: &str, version: &str) -> RuntimeError {
347 match self {
348 Self::Load(message) => RuntimeError::ProcessDriverLoad {
349 name: name.to_string(),
350 version: version.to_string(),
351 message,
352 },
353 Self::CallFailed { call, message } => RuntimeError::ProcessDriverCallFailed {
354 name: name.to_string(),
355 version: version.to_string(),
356 call,
357 message,
358 },
359 Self::Other(message) => RuntimeError::Runtime(message),
360 }
361 }
362}
363
364impl ProcessDriver {
365 pub(crate) async fn open(
368 workers: Option<&std::sync::Arc<crate::guest_worker::GuestWorkers>>,
369 engine: &Engine,
370 component: &Component,
371 name: &str,
372 version: &str,
373 call_deadlines: CallDeadlines,
374 ) -> Result<Self, RuntimeError> {
375 let Some(workers) = workers else {
376 return Self::instantiate(engine, component, name, version, call_deadlines).await;
377 };
378 let remote = crate::guest_worker::RemoteProcessDriver::open(
379 workers,
380 component,
381 name,
382 version,
383 &call_deadlines,
384 )
385 .await?;
386 Ok(Self {
387 name: name.to_string(),
388 version: version.to_string(),
389 body: DriverBody::Remote(remote),
390 call_deadlines,
391 cut: None,
392 })
393 }
394
395 pub(crate) async fn call(&mut self, call: DriverCall) -> Result<DriverAnswer, RuntimeError> {
398 Ok(match call {
399 DriverCall::Describe => DriverAnswer::Description(self.describe().await?),
400 DriverCall::Launch(request) => DriverAnswer::Launch(self.launch(request).await?),
401 DriverCall::Parse(lines) => DriverAnswer::Events(self.parse(lines).await?),
402 DriverCall::ClassifyExit(exit) => DriverAnswer::Event(self.classify_exit(exit).await?),
403 DriverCall::ReadLogin(output) => DriverAnswer::Login(self.read_login(output).await?),
404 })
405 }
406
407 async fn remote_call(
409 &mut self,
410 call: DriverCall,
411 ) -> Option<Result<DriverAnswer, RuntimeError>> {
412 match &self.body {
413 DriverBody::Remote(remote) => Some(remote.call(call, &self.call_deadlines).await),
414 DriverBody::Local { .. } => None,
415 }
416 }
417
418 fn unexpected(&self, call: &str, answer: DriverAnswer) -> RuntimeError {
419 self.failed_in(call, format!("the guest worker answered {answer:?}"))
420 }
421
422 pub(crate) async fn instantiate(
432 engine: &Engine,
433 component: &Component,
434 name: &str,
435 version: &str,
436 call_deadlines: CallDeadlines,
437 ) -> Result<Self, RuntimeError> {
438 let load_error = |message: String| RuntimeError::ProcessDriverLoad {
439 name: name.to_string(),
440 version: version.to_string(),
441 message,
442 };
443 let limits = ExecutionLimits::default().for_hook_calls(None);
444 crate::guest_worker::check_store_placement(crate::guest_worker::StoreSite::ProcessDriver)?;
445
446 let mut linker: Linker<ProcessDriverStoreState> = Linker::new(engine);
447 wasmtime_wasi::p2::add_to_linker_async(&mut linker)
448 .map_err(|e| load_error(e.to_string()))?;
449
450 let state = ProcessDriverStoreState {
451 wasi: WasiCtxBuilder::new().build(),
454 table: ResourceTable::new(),
455 limits: limits.limiter(),
456 };
457 let mut store = Store::new(engine, state);
458 store.limiter(|state| &mut state.limits);
459 store.set_epoch_deadline(limits.deadline_ticks());
460
461 let bindings = call_deadlines
462 .bound(
463 name,
464 ComponentRole::ProcessDriver,
465 "instantiate",
466 &limits,
467 ProcessDriverBindings::instantiate_async(&mut store, component, &linker),
468 )
469 .await
470 .map_err(|_| load_error(format!("instantiation {}", wall_deadline_text(&limits))))?
471 .map_err(|e| load_error(format!("{e:#}")))?;
472 Ok(Self {
473 name: name.to_string(),
474 version: version.to_string(),
475 body: DriverBody::Local { store, bindings },
476 call_deadlines,
477 cut: None,
478 })
479 }
480
481 pub async fn describe(&mut self) -> Result<Description, RuntimeError> {
488 if let Some(answer) = self.remote_call(DriverCall::Describe).await {
489 return match answer? {
490 DriverAnswer::Description(description) => Ok(description),
491 other => Err(self.unexpected("describe", other)),
492 };
493 }
494 self.enter("describe")
495 .map_err(|message| self.load_error(message))?;
496 let limits = self.reset_deadline();
497 let DriverBody::Local { store, bindings } = &mut self.body else {
498 unreachable!("a remote driver answered above")
499 };
500 let called = self
501 .call_deadlines
502 .bound(
503 &self.name,
504 ComponentRole::ProcessDriver,
505 "describe",
506 &limits,
507 bindings.murmur_driver_process().call_describe(&mut *store),
508 )
509 .await;
510 let description = self
511 .settle("describe", &limits, called)
512 .map_err(|message| self.load_error(format!("describe() {message}")))?
513 .map_err(|e| {
514 let message = match classify_guest_failure(&e, self.limiter()) {
515 GuestFailure::DeadlineExceeded { seconds } => {
516 format!("describe() exceeded its {seconds}s deadline and was interrupted")
517 }
518 GuestFailure::ResourceLimit { message } => {
519 format!("describe() exceeded its resource limits: {message}")
520 }
521 GuestFailure::Other => format!("describe() trapped: {e:#}"),
522 };
523 self.load_error(message)
524 })?;
525 validate_description(&self.name, &self.version, &description)?;
526 Ok(description)
527 }
528
529 pub async fn launch(
532 &mut self,
533 request: LaunchRequest,
534 ) -> Result<Result<LaunchPlan, String>, RuntimeError> {
535 if let DriverBody::Remote(remote) = &self.body {
536 return match remote
537 .call(DriverCall::Launch(request), &self.call_deadlines)
538 .await?
539 {
540 DriverAnswer::Launch(plan) => Ok(plan),
541 other => Err(self.unexpected("launch", other)),
542 };
543 }
544 self.enter("launch")
545 .map_err(|message| self.failed_in("launch", message))?;
546 let limits = self.reset_deadline();
547 let DriverBody::Local { store, bindings } = &mut self.body else {
548 unreachable!("a remote driver answered above")
549 };
550 let called = self
551 .call_deadlines
552 .bound(
553 &self.name,
554 ComponentRole::ProcessDriver,
555 "launch",
556 &limits,
557 bindings
558 .murmur_driver_process()
559 .call_launch(&mut *store, &request),
560 )
561 .await;
562 self.settle("launch", &limits, called)
563 .map_err(|message| self.failed_in("launch", message))?
564 .map_err(|e| self.call_error("launch", &e))
565 }
566
567 pub async fn parse(&mut self, lines: Vec<String>) -> Result<Vec<Event>, RuntimeError> {
570 if let DriverBody::Remote(remote) = &self.body {
571 return match remote
572 .call(DriverCall::Parse(lines), &self.call_deadlines)
573 .await?
574 {
575 DriverAnswer::Events(events) => Ok(events),
576 other => Err(self.unexpected("parse", other)),
577 };
578 }
579 self.enter("parse")
580 .map_err(|message| self.failed_in("parse", message))?;
581 let limits = self.reset_deadline();
582 let DriverBody::Local { store, bindings } = &mut self.body else {
583 unreachable!("a remote driver answered above")
584 };
585 let called = self
586 .call_deadlines
587 .bound(
588 &self.name,
589 ComponentRole::ProcessDriver,
590 "parse",
591 &limits,
592 bindings
593 .murmur_driver_process()
594 .call_parse(&mut *store, &lines),
595 )
596 .await;
597 self.settle("parse", &limits, called)
598 .map_err(|message| self.failed_in("parse", message))?
599 .map_err(|e| self.call_error("parse", &e))
600 }
601
602 pub async fn classify_exit(&mut self, exit: ExitStatus) -> Result<Event, RuntimeError> {
604 if let DriverBody::Remote(remote) = &self.body {
605 return match remote
606 .call(DriverCall::ClassifyExit(exit), &self.call_deadlines)
607 .await?
608 {
609 DriverAnswer::Event(event) => Ok(event),
610 other => Err(self.unexpected("classify-exit", other)),
611 };
612 }
613 self.enter("classify-exit")
614 .map_err(|message| self.failed_in("classify-exit", message))?;
615 let limits = self.reset_deadline();
616 let DriverBody::Local { store, bindings } = &mut self.body else {
617 unreachable!("a remote driver answered above")
618 };
619 let called = self
620 .call_deadlines
621 .bound(
622 &self.name,
623 ComponentRole::ProcessDriver,
624 "classify-exit",
625 &limits,
626 bindings
627 .murmur_driver_process()
628 .call_classify_exit(&mut *store, &exit),
629 )
630 .await;
631 self.settle("classify-exit", &limits, called)
632 .map_err(|message| self.failed_in("classify-exit", message))?
633 .map_err(|e| self.call_error("classify-exit", &e))
634 }
635
636 pub async fn read_login(&mut self, output: LoginOutput) -> Result<LoginState, RuntimeError> {
638 if let DriverBody::Remote(remote) = &self.body {
639 return match remote
640 .call(DriverCall::ReadLogin(output), &self.call_deadlines)
641 .await?
642 {
643 DriverAnswer::Login(state) => Ok(state),
644 other => Err(self.unexpected("read-login", other)),
645 };
646 }
647 self.enter("read-login")
648 .map_err(|message| self.failed_in("read-login", message))?;
649 let limits = self.reset_deadline();
650 let DriverBody::Local { store, bindings } = &mut self.body else {
651 unreachable!("a remote driver answered above")
652 };
653 let called = self
654 .call_deadlines
655 .bound(
656 &self.name,
657 ComponentRole::ProcessDriver,
658 "read-login",
659 &limits,
660 bindings
661 .murmur_driver_process()
662 .call_read_login(&mut *store, &output),
663 )
664 .await;
665 self.settle("read-login", &limits, called)
666 .map_err(|message| self.failed_in("read-login", message))?
667 .map_err(|e| self.call_error("read-login", &e))
668 }
669
670 fn reset_deadline(&mut self) -> ExecutionLimits {
674 let DriverBody::Local { store, .. } = &mut self.body else {
675 return ExecutionLimits::default().for_hook_calls(None);
676 };
677 let limits = store.data().limits.limits();
678 store.set_epoch_deadline(limits.deadline_ticks());
679 limits
680 }
681
682 fn limiter(&self) -> &ExecutionLimiter {
684 match &self.body {
685 DriverBody::Local { store, .. } => &store.data().limits,
686 DriverBody::Remote(_) => unreachable!("a remote driver classifies in its worker"),
687 }
688 }
689
690 fn enter(&self, call: &str) -> Result<(), String> {
692 match self.cut {
693 None => Ok(()),
694 Some((earlier, seconds)) => Err(format!(
695 "{call} was not called: the earlier {earlier} call was stopped at its {seconds}s \
696 deadline, and a stopped driver is not entered again"
697 )),
698 }
699 }
700
701 fn settle<T>(
704 &mut self,
705 call: &'static str,
706 limits: &ExecutionLimits,
707 called: Result<T, WallDeadline>,
708 ) -> Result<T, String> {
709 called.map_err(|_| {
710 self.cut = Some((call, limits.deadline_seconds));
711 wall_deadline_text(limits)
712 })
713 }
714
715 fn failed_in(&self, call: &str, message: String) -> RuntimeError {
716 RuntimeError::ProcessDriverCallFailed {
717 name: self.name.clone(),
718 version: self.version.clone(),
719 call: call.to_string(),
720 message,
721 }
722 }
723
724 fn call_error(&self, call: &str, error: &wasmtime::Error) -> RuntimeError {
725 let message = match classify_guest_failure(error, self.limiter()) {
726 GuestFailure::DeadlineExceeded { seconds } => {
727 format!("exceeded its {seconds}s deadline and was interrupted")
728 }
729 GuestFailure::ResourceLimit { message } => {
730 format!("exceeded its resource limits: {message}")
731 }
732 GuestFailure::Other => format!("trapped: {error:#}"),
733 };
734 self.failed_in(call, message)
735 }
736
737 fn load_error(&self, message: String) -> RuntimeError {
738 RuntimeError::ProcessDriverLoad {
739 name: self.name.clone(),
740 version: self.version.clone(),
741 message,
742 }
743 }
744}
745
746fn wall_deadline_text(limits: &ExecutionLimits) -> String {
750 format!(
751 "waited on the host past its {}s deadline and was stopped",
752 limits.deadline_seconds
753 )
754}
755
756pub(crate) fn is_usable_env_name(name: &str) -> bool {
763 !name.is_empty() && !name.contains('=') && !name.contains('\0')
764}
765
766fn unusable_env_name(description: &Description) -> Option<&str> {
769 description
770 .required_env
771 .iter()
772 .find(|var| !is_usable_env_name(var))
773 .map(String::as_str)
774}
775
776fn validate_description(
780 name: &str,
781 version: &str,
782 description: &Description,
783) -> Result<(), RuntimeError> {
784 let load_error = |message: String| RuntimeError::ProcessDriverLoad {
785 name: name.to_string(),
786 version: version.to_string(),
787 message,
788 };
789 if let Some(bad) = unusable_env_name(description) {
790 return Err(load_error(format!(
791 "describe() names {bad:?} in required-env, which is not a usable variable name"
792 )));
793 }
794 if matches!(&description.login_check, LoginCheck::Command(args) if args.is_empty()) {
795 return Err(load_error(
796 "describe() names a login-check command with no arguments; the bare harness binary \
797 may start a session, so the login check must name the arguments that make it report \
798 its login"
799 .to_string(),
800 ));
801 }
802 if let Err(reason) = crate::harness_needs::validate_harness_needs(&description.harness_needs) {
803 return Err(load_error(format!("describe() {reason}")));
804 }
805 Ok(())
806}
807
808const STANDALONE_NAME: &str = "<wasm>";
811const STANDALONE_VERSION: &str = "";
812
813pub(crate) struct StandaloneDriver {
816 engine: Engine,
817 component: Component,
818 pub(crate) rt: tokio::runtime::Runtime,
819 _ticker: EpochTicker,
820 name: String,
821 version: String,
822 workers: Option<std::sync::Arc<crate::guest_worker::GuestWorkers>>,
824}
825
826impl StandaloneDriver {
827 fn compile(wasm: &[u8]) -> Result<Self, RuntimeError> {
828 Self::compile_named(wasm, STANDALONE_NAME, STANDALONE_VERSION)
829 }
830
831 pub(crate) fn compile_named(
833 wasm: &[u8],
834 name: &str,
835 version: &str,
836 ) -> Result<Self, RuntimeError> {
837 let engine = crate::runtime::build_engine()?;
838 let ticker = EpochTicker::spawn(&engine);
839 let component =
840 Component::new(&engine, wasm).map_err(|e| RuntimeError::ToolComponentCompile {
841 name: name.to_string(),
842 version: version.to_string(),
843 message: e.to_string(),
844 })?;
845 let rt = tokio::runtime::Builder::new_current_thread()
846 .enable_all()
847 .build()
848 .map_err(|e| {
849 RuntimeError::Runtime(format!("failed to build process driver runtime: {e}"))
850 })?;
851 Ok(StandaloneDriver {
852 engine,
853 component,
854 rt,
855 _ticker: ticker,
856 name: name.to_string(),
857 version: version.to_string(),
858 workers: crate::guest_worker::session_launcher()
859 .map(crate::guest_worker::GuestWorkers::new),
860 })
861 }
862
863 pub(crate) async fn instantiate(&self) -> Result<ProcessDriver, RuntimeError> {
864 ProcessDriver::open(
865 self.workers.as_ref(),
866 &self.engine,
867 &self.component,
868 &self.name,
869 &self.version,
870 CallDeadlines::default(),
872 )
873 .await
874 }
875}
876
877pub fn describe_process_driver_wasm(wasm: &[u8]) -> Result<Description, RuntimeError> {
884 let driver = StandaloneDriver::compile(wasm)?;
885 driver.rt.block_on(async {
886 let mut instance = driver.instantiate().await?;
887 instance.describe().await
888 })
889}
890
891pub fn launch_and_parse_process_driver_wasm(
898 wasm: &[u8],
899 request: LaunchRequest,
900 lines: Vec<String>,
901) -> Result<(Result<LaunchPlan, String>, Vec<Event>), RuntimeError> {
902 let driver = StandaloneDriver::compile(wasm)?;
903 driver.rt.block_on(async {
904 let mut instance = driver.instantiate().await?;
905 let plan = instance.launch(request).await?;
906 let events = instance.parse(lines).await?;
907 Ok((plan, events))
908 })
909}
910
911#[cfg(test)]
912mod tests {
913 use murmur_artifact::ContractDirection;
914
915 use super::*;
916
917 fn served() -> &'static WitLevel {
918 crate::served_wit_level()
919 }
920
921 fn contracts(exports: &[&str]) -> WitContracts {
922 WitContracts {
923 exports: exports.iter().map(|e| e.to_string()).collect(),
924 imports: Vec::new(),
925 }
926 }
927
928 #[test]
929 fn process_driver_interface_process_transport_accepts_the_process_export() {
930 let c = contracts(&[PROCESS_DRIVER_IFACE]);
931 assert!(check_driver_interface("process", "d", "1.0.0", Some(&c), served()).is_ok());
932 }
933
934 #[test]
935 fn process_driver_interface_process_transport_refuses_a_tool_export() {
936 let c = contracts(&["murmur:tool/run@0.2.0"]);
937 let err = check_driver_interface("process", "d", "1.0.0", Some(&c), served()).unwrap_err();
938 match &err {
939 RuntimeError::ProcessDriverInterfaceMissing { expected, .. } => {
940 assert_eq!(expected, PROCESS_DRIVER_IFACE);
941 }
942 other => panic!("unexpected error: {other:?}"),
943 }
944 let message = err.to_string();
945 assert!(message.contains("'d@1.0.0'"), "{message}");
946 assert!(message.contains(PROCESS_DRIVER_IFACE), "{message}");
947 }
948
949 #[test]
950 fn process_driver_interface_process_transport_refuses_another_version_as_a_level_mismatch() {
951 let c = contracts(&["murmur:driver/process@0.0.1"]);
952 let err = check_driver_interface("process", "d", "1.0.0", Some(&c), served()).unwrap_err();
953 match &err {
954 RuntimeError::WitLevelMismatch {
955 level, mismatched, ..
956 } => {
957 assert_eq!(*level, served().level);
958 assert_eq!(mismatched.len(), 1);
959 assert_eq!(mismatched[0].interface, "murmur:driver/process@0.0.1");
960 assert_eq!(mismatched[0].direction, ContractDirection::Export);
961 }
962 other => panic!("unexpected error: {other:?}"),
963 }
964 assert_eq!(
965 err.to_string(),
966 format!(
967 "artifact 'd@1.0.0' does not fit WIT level {0}, the level this mur serves: it \
968 exports murmur:driver/process@0.0.1, and level {0} has murmur:driver@{1}",
969 served().level,
970 served().version_of("murmur:driver").unwrap()
971 )
972 );
973 }
974
975 fn level_1() -> &'static WitLevel {
978 &crate::wit_versions::wit_levels()[0]
979 }
980
981 #[test]
982 fn process_driver_interface_follows_the_level_it_is_given() {
983 let c = contracts(&[PROCESS_DRIVER_IFACE]);
984 assert!(check_driver_interface("process", "d", "1.0.0", Some(&c), served()).is_ok());
985 match check_driver_interface("process", "d", "1.0.0", Some(&c), level_1()).unwrap_err() {
986 RuntimeError::WitLevelMismatch {
987 level, mismatched, ..
988 } => {
989 assert_eq!(level, 1);
990 assert_eq!(mismatched[0].direction, ContractDirection::Export);
991 assert_eq!(mismatched[0].package, "murmur:driver");
992 assert_eq!(mismatched[0].served_version.as_deref(), Some("0.3.0"));
993 }
994 other => panic!("unexpected error: {other:?}"),
995 }
996 let old = contracts(&["murmur:driver/process@0.3.0"]);
997 assert!(check_driver_interface("process", "d", "1.0.0", Some(&old), level_1()).is_ok());
998 let stream = importing(&["murmur:stream/events@0.1.0"]);
999 assert!(check_stream_interface("d", "1.0.0", Some(&stream), level_1()).is_ok());
1000 }
1001
1002 #[test]
1003 fn process_driver_interface_process_transport_refuses_no_contracts() {
1004 let err = check_driver_interface("process", "d", "1.0.0", None, served()).unwrap_err();
1005 assert!(matches!(
1006 err,
1007 RuntimeError::ProcessDriverInterfaceMissing { .. }
1008 ));
1009 }
1010
1011 #[test]
1012 fn process_driver_interface_http_transport_refuses_the_process_export() {
1013 let c = contracts(&[PROCESS_DRIVER_IFACE]);
1014 let err = check_driver_interface("http", "d", "1.0.0", Some(&c), served()).unwrap_err();
1015 match &err {
1016 RuntimeError::HttpDriverExportsProcessInterface { name, exported, .. } => {
1017 assert_eq!(name, "d");
1018 assert_eq!(exported, &vec![PROCESS_DRIVER_IFACE.to_string()]);
1019 }
1020 other => panic!("unexpected error: {other:?}"),
1021 }
1022 }
1023
1024 #[test]
1025 fn process_driver_interface_http_transport_accepts_a_tool_export() {
1026 let c = contracts(&["murmur:tool/run@0.2.0"]);
1027 assert!(check_driver_interface("http", "d", "1.0.0", Some(&c), served()).is_ok());
1028 }
1029
1030 fn importing(imports: &[&str]) -> WitContracts {
1031 WitContracts {
1032 exports: vec!["murmur:tool/run@0.2.0".to_string()],
1033 imports: imports.iter().map(|i| i.to_string()).collect(),
1034 }
1035 }
1036
1037 #[test]
1038 fn stream_interface_refuses_and_lists_a_murmur_text_import() {
1039 let c = importing(&[
1040 "wasi:cli/environment@0.2.12",
1041 "murmur:text/chunks@0.1.0",
1042 "murmur:task/task@0.1.0",
1043 ]);
1044 let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
1045 match &err {
1046 RuntimeError::WitLevelMismatch {
1047 name,
1048 version,
1049 mismatched,
1050 ..
1051 } => {
1052 assert_eq!((name.as_str(), version.as_str()), ("d", "1.0.0"));
1053 assert_eq!(mismatched.len(), 1);
1054 assert_eq!(mismatched[0].interface, "murmur:text/chunks@0.1.0");
1055 assert_eq!(mismatched[0].served_version, None);
1056 }
1057 other => panic!("unexpected error: {other:?}"),
1058 }
1059 let message = err.to_string();
1060 assert!(message.contains("'d@1.0.0'"), "{message}");
1061 assert!(
1062 message.contains(
1063 "it imports murmur:text/chunks@0.1.0, and level 6 has no version of murmur:text"
1064 ),
1065 "{message}"
1066 );
1067 }
1068
1069 #[test]
1070 fn stream_interface_refuses_another_version_of_murmur_stream() {
1071 let c = importing(&["murmur:stream/events@0.2.0"]);
1072 let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
1073 assert!(matches!(
1074 err,
1075 RuntimeError::WitLevelMismatch { ref mismatched, .. }
1076 if mismatched.len() == 1 && mismatched[0].interface == "murmur:stream/events@0.2.0"
1077 ));
1078 }
1079
1080 #[test]
1083 fn stream_interface_refuses_an_unknown_interface_at_the_served_version_as_not_served() {
1084 let c = importing(&["murmur:stream/other@0.1.0"]);
1085 let err = check_stream_interface("d", "1.0.0", Some(&c), served()).unwrap_err();
1086 match &err {
1087 RuntimeError::DriverStreamInterfaceNotServed {
1088 imported, expected, ..
1089 } => {
1090 assert_eq!(imported, &vec!["murmur:stream/other@0.1.0".to_string()]);
1091 assert_eq!(expected, crate::runtime::WIT_STREAM_EVENTS_IFACE);
1092 }
1093 other => panic!("unexpected error: {other:?}"),
1094 }
1095 let both = importing(&["murmur:stream/other@0.1.0", "murmur:text/chunks@0.1.0"]);
1097 assert!(matches!(
1098 check_stream_interface("d", "1.0.0", Some(&both), served()).unwrap_err(),
1099 RuntimeError::WitLevelMismatch { ref mismatched, .. }
1100 if mismatched.len() == 1 && mismatched[0].interface == "murmur:text/chunks@0.1.0"
1101 ));
1102 }
1103
1104 #[test]
1105 fn driver_refusal_reports_a_level_mismatch_over_any_other() {
1106 let not_served = || {
1107 check_stream_interface(
1108 "d",
1109 "1.0.0",
1110 Some(&importing(&["murmur:stream/other@0.1.0"])),
1111 served(),
1112 )
1113 };
1114 let old_export = || {
1115 check_driver_interface(
1116 "process",
1117 "d",
1118 "1.0.0",
1119 Some(&contracts(&["murmur:driver/process@0.2.0"])),
1120 served(),
1121 )
1122 };
1123 let text_import = || {
1124 check_stream_interface(
1125 "d",
1126 "1.0.0",
1127 Some(&importing(&["murmur:text/chunks@0.1.0"])),
1128 served(),
1129 )
1130 };
1131 let no_export = || check_driver_interface("process", "d", "1.0.0", None, served());
1132
1133 assert!(matches!(
1134 driver_refusal(not_served(), old_export()),
1135 Err(RuntimeError::WitLevelMismatch { .. })
1136 ));
1137 assert!(matches!(
1138 driver_refusal(text_import(), no_export()),
1139 Err(RuntimeError::WitLevelMismatch { .. })
1140 ));
1141 assert!(matches!(
1142 driver_refusal(not_served(), no_export()),
1143 Err(RuntimeError::DriverStreamInterfaceNotServed { .. })
1144 ));
1145 match driver_refusal(text_import(), old_export()) {
1146 Err(RuntimeError::WitLevelMismatch { mismatched, .. }) => {
1147 let names: Vec<&str> = mismatched.iter().map(|m| m.interface.as_str()).collect();
1148 assert_eq!(
1149 names,
1150 ["murmur:driver/process@0.2.0", "murmur:text/chunks@0.1.0"]
1151 );
1152 }
1153 other => panic!("unexpected result: {other:?}"),
1154 }
1155 assert!(driver_refusal(Ok(()), Ok(())).is_ok());
1156 }
1157
1158 #[test]
1159 fn stream_interface_accepts_exactly_the_served_instance() {
1160 let c = importing(&["murmur:stream/events@0.1.0", "wasi:http/types@0.2.12"]);
1161 assert!(check_stream_interface("d", "1.0.0", Some(&c), served()).is_ok());
1162 }
1163
1164 #[test]
1165 fn stream_interface_accepts_a_component_with_no_murmur_import() {
1166 let c = importing(&["wasi:cli/environment@0.2.12"]);
1167 assert!(check_stream_interface("d", "1.0.0", Some(&c), served()).is_ok());
1168 assert!(check_stream_interface("d", "1.0.0", Some(&importing(&[])), served()).is_ok());
1169 }
1170
1171 #[test]
1172 fn stream_interface_accepts_no_contracts() {
1173 assert!(check_stream_interface("d", "1.0.0", None, served()).is_ok());
1174 }
1175
1176 fn description(required_env: &[&str]) -> Description {
1177 Description {
1178 harness: "h".to_string(),
1179 binary: "b".to_string(),
1180 version_args: Vec::new(),
1181 tested_versions: Vec::new(),
1182 interrupt: InterruptMethod::Unsupported,
1183 required_env: required_env.iter().map(|v| v.to_string()).collect(),
1184 streams_text: false,
1185 reports_usage: false,
1186 login_check: LoginCheck::Unavailable("no login".to_string()),
1187 harness_needs: HarnessNeeds {
1188 paths: Vec::new(),
1189 hosts: Vec::new(),
1190 },
1191 }
1192 }
1193
1194 #[test]
1195 fn login_check_command_with_arguments_passes() {
1196 let mut described = description(&[]);
1197 described.login_check = LoginCheck::Command(vec!["auth".to_string(), "status".to_string()]);
1198 assert!(validate_description("d", "1.0.0", &described).is_ok());
1199 assert!(validate_description("d", "1.0.0", &description(&[])).is_ok());
1200 }
1201
1202 #[test]
1205 fn login_check_command_with_no_arguments_is_refused_at_load() {
1206 let mut described = description(&[]);
1207 described.login_check = LoginCheck::Command(Vec::new());
1208 let err = validate_description("d", "1.0.0", &described).unwrap_err();
1209 match &err {
1210 RuntimeError::ProcessDriverLoad { message, .. } => {
1211 assert!(
1212 message.contains("login-check command with no arguments"),
1213 "{message}"
1214 );
1215 }
1216 other => panic!("unexpected error: {other:?}"),
1217 }
1218 }
1219
1220 #[test]
1221 fn required_env_all_declared_passes() {
1222 let allow = vec!["A".to_string(), "B".to_string(), "C".to_string()];
1223 assert!(check_required_env("d", "1.0.0", &description(&["B", "A"]), &allow).is_ok());
1224 }
1225
1226 #[test]
1227 fn required_env_reports_missing_names_in_order_once_each() {
1228 let allow = vec!["B".to_string()];
1229 let err = check_required_env("d", "1.0.0", &description(&["Z", "B", "A", "Z"]), &allow)
1230 .unwrap_err();
1231 match &err {
1232 RuntimeError::ProcessDriverRequiredEnvNotAllowed { missing, .. } => {
1233 assert_eq!(missing, &vec!["Z".to_string(), "A".to_string()]);
1234 }
1235 other => panic!("unexpected error: {other:?}"),
1236 }
1237 let message = err.to_string();
1238 assert!(message.contains("'d@1.0.0'"), "{message}");
1239 assert!(message.contains("Z, A"), "{message}");
1240 }
1241
1242 #[test]
1243 fn required_env_usable_names_pass() {
1244 assert_eq!(
1245 unusable_env_name(&description(&["HOME", "PROFILE_2"])),
1246 None
1247 );
1248 assert_eq!(unusable_env_name(&description(&[])), None);
1249 }
1250
1251 #[test]
1252 fn required_env_refuses_a_name_that_is_not_a_variable() {
1253 for bad in ["", "A=B", "A\0B"] {
1254 assert_eq!(
1255 unusable_env_name(&description(&["HOME", bad, "TAIL"])),
1256 Some(bad),
1257 "{bad:?} should be refused"
1258 );
1259 }
1260 }
1261
1262 #[test]
1268 #[ignore = "a measurement, not an assertion; run with --release --ignored --nocapture"]
1269 fn process_driver_parse_cost() {
1270 use std::time::Instant;
1271
1272 const WASM: &[u8] = include_bytes!(concat!(
1273 env!("CARGO_MANIFEST_DIR"),
1274 "/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
1275 ));
1276 let engine = crate::runtime::build_engine().expect("engine");
1277 let _ticker = EpochTicker::spawn(&engine);
1278 let component = Component::new(&engine, WASM).expect("fixture compiles");
1279 let rt = tokio::runtime::Builder::new_current_thread()
1280 .enable_all()
1281 .build()
1282 .expect("runtime");
1283
1284 let instantiations = 50;
1285 let start = Instant::now();
1286 for _ in 0..instantiations {
1287 rt.block_on(ProcessDriver::instantiate(
1288 &engine,
1289 &component,
1290 "d",
1291 "0.1.0",
1292 CallDeadlines::default(),
1293 ))
1294 .expect("instantiates");
1295 }
1296 let per_instantiation = start.elapsed() / instantiations;
1297
1298 let mut driver = rt
1299 .block_on(ProcessDriver::instantiate(
1300 &engine,
1301 &component,
1302 "d",
1303 "0.1.0",
1304 CallDeadlines::default(),
1305 ))
1306 .expect("instantiates");
1307 for lines in [1usize, 64] {
1308 let batch: Vec<String> = (0..lines).map(|i| format!("text line {i}")).collect();
1309 let runs = 2_000;
1310 let mut samples = Vec::with_capacity(runs);
1311 for _ in 0..runs {
1312 let start = Instant::now();
1313 let events = rt.block_on(driver.parse(batch.clone())).expect("parses");
1314 samples.push(start.elapsed());
1315 assert_eq!(events.len(), lines);
1316 }
1317 samples.sort_unstable();
1318 let median = samples[samples.len() / 2];
1319 let p95 = samples[samples.len() * 95 / 100];
1320 crate::diagnostic::raw_to_stdout(&format!(
1321 "parse {lines}-line batch on one reused instance: median {median:?}, p95 {p95:?} \
1322 ({runs} calls)\n"
1323 ));
1324 }
1325 crate::diagnostic::raw_to_stdout(&format!(
1326 "instantiate one driver: {per_instantiation:?} (mean of {instantiations})\n"
1327 ));
1328 }
1329
1330 #[test]
1331 fn process_driver_granted_nothing() {
1332 let wat = r#"(component
1333 (import "murmur:stream/events@0.1.0" (instance
1334 (export "emit-chunk" (func (param "chunk" string)))
1335 ))
1336 )"#;
1337 let bytes = wat::parse_str(wat).expect("component WAT parses");
1338 let engine = crate::runtime::build_engine().expect("engine");
1339 let _ticker = EpochTicker::spawn(&engine);
1340 let component = Component::new(&engine, &bytes).expect("component compiles");
1341 let rt = tokio::runtime::Builder::new_current_thread()
1342 .enable_all()
1343 .build()
1344 .expect("runtime");
1345 let result = rt.block_on(ProcessDriver::instantiate(
1346 &engine,
1347 &component,
1348 "d",
1349 "1.0.0",
1350 CallDeadlines::default(),
1351 ));
1352 match result {
1353 Err(RuntimeError::ProcessDriverLoad { name, message, .. }) => {
1354 assert_eq!(name, "d");
1355 assert!(message.contains("murmur:stream/events@0.1.0"), "{message}");
1356 }
1357 Err(other) => panic!("unexpected error: {other:?}"),
1358 Ok(_) => panic!("a driver importing a Murmur interface instantiated"),
1359 }
1360 }
1361
1362 #[test]
1363 fn launch_and_parse_run_on_one_instance() {
1364 const WASM: &[u8] = include_bytes!(concat!(
1365 env!("CARGO_MANIFEST_DIR"),
1366 "/../murmur-cli/tests/fixtures/process-driver/tool/process-driver.wasm"
1367 ));
1368 let request = LaunchRequest {
1369 model: None,
1370 system_prompt: String::new(),
1371 config: None,
1372 bridge: Some(Bridge {
1373 server_name: "murmur".to_string(),
1374 url: "http://127.0.0.1:1/mcp".to_string(),
1375 bearer_token: "token".to_string(),
1376 tool_names: vec!["python3".to_string()],
1377 }),
1378 session: Session {
1379 id: "s1".to_string(),
1380 mode: SessionMode::New,
1381 },
1382 harness_version: None,
1383 task: "probe".to_string(),
1384 };
1385 let lines = vec![
1386 r#"tool c1 murmur__python3 {"command":"x"}"#.to_string(),
1387 "end done".to_string(),
1388 ];
1389 let (plan, events) =
1390 launch_and_parse_process_driver_wasm(WASM, request, lines).expect("driver runs");
1391 let plan = plan.expect("the fixture accepts a non-empty task");
1392 assert!(plan
1393 .env_set
1394 .iter()
1395 .any(|(name, value)| name == "FIXTURE_BRIDGE_TOKEN" && value == "token"));
1396 match &events[..] {
1399 [Event::ToolCall(call), Event::TurnEnd(summary)] => {
1400 assert_eq!(call.name, "python3");
1401 assert_eq!(call.input, r#"{"command":"x"}"#);
1402 assert_eq!(summary, "done");
1403 }
1404 other => panic!("unexpected events: {other:?}"),
1405 }
1406 }
1407}