Skip to main content

capsule_runtime/
process_driver.rs

1//! Loading a process driver: the artifact a `transport: process` manifest names under
2//! `inference.driver`, which knows how to drive one harness CLI.
3//!
4//! A process driver is pure translation and is granted nothing. [`ProcessDriver`] instantiates
5//! it on an empty WASI context with no Murmur host import, so a driver that reaches for anything
6//! else fails to load rather than running with less than it expected.
7
8use 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    /// The WIT spelling of the state, which every record and frame carrying it uses.
31    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
40/// The instance name every process driver exports.
41pub const PROCESS_DRIVER_IFACE: &str = "murmur:driver/process@0.4.0";
42
43/// Every version of the process interface starts with this, so an export built against another
44/// version is still recognised as a process driver.
45const PROCESS_DRIVER_IFACE_PREFIX: &str = "murmur:driver/process@";
46
47/// Checks the inference driver's exports against the transport that names it and the WIT level
48/// this host serves.
49///
50/// Under `process` the driver's exports must include the process interface at the version
51/// `level` lists for `murmur:driver` (for the served level, [`PROCESS_DRIVER_IFACE`]) under that
52/// exact name; other instances alongside it are ignored. A driver exporting the process interface
53/// only at other versions does not fit the level and is refused with
54/// [`RuntimeError::WitLevelMismatch`]; one exporting no version of it is not a process driver and
55/// is refused with [`RuntimeError::ProcessDriverInterfaceMissing`]. Under `http` it must export
56/// no version of the process interface. `contracts` is what
57/// [`murmur_artifact::extract_wit_contracts`] read out of the driver's wasm; `None` (a core
58/// module, or bytes it could not read) counts as exporting nothing. Any other transport passes.
59pub 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        // Refuse even if every export fits the level: none is `expected`, so the driver must not
87        // pass.
88        "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
108/// Every instance of the retired `murmur:text` package starts with this. No version of it is
109/// served: `murmur:stream/events` replaced it.
110const RETIRED_TEXT_IFACE_PREFIX: &str = "murmur:text/";
111
112/// Every instance of the `murmur:stream` package starts with this.
113const STREAM_IFACE_PREFIX: &str = "murmur:stream/";
114
115/// Checks a `runtime: driver` artifact's imports against the stream interface of the WIT level
116/// this host serves.
117///
118/// Refuses a component importing any `murmur:text/*` instance, or any `murmur:stream/*` instance
119/// other than exactly `murmur:stream/events` at the version `level` lists (for the served level,
120/// [`WIT_STREAM_EVENTS_IFACE`]): the host serves neither, so the driver would fail to instantiate
121/// on its first turn. The imports among those that do not fit `level` are refused with
122/// [`RuntimeError::WitLevelMismatch`]; only when every one is at the level's version of its
123/// package, under an interface name that does not exist, is it
124/// [`RuntimeError::DriverStreamInterfaceNotServed`]. `contracts` is what
125/// [`murmur_artifact::extract_wit_contracts`] read out of the driver's wasm; `None`, and a
126/// component importing neither package, pass.
127///
128/// [`WIT_STREAM_EVENTS_IFACE`]: crate::runtime::WIT_STREAM_EVENTS_IFACE
129pub 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
163/// The one refusal of a driver both [`check_stream_interface`] and [`check_driver_interface`]
164/// checked: a level mismatch over any other, both checks' mismatches together (exports first)
165/// when each found one, and otherwise the stream check's refusal before the driver check's.
166pub(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
197/// `<package>/<interface>@<version>` at the version `level` lists for `package`, if it lists one.
198fn 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
204/// Refuses a WASM tool, driver or script capsule whose `murmur:*` exports and imports do not all
205/// fit `level`, with [`RuntimeError::WitLevelMismatch`] naming each one that does not. A component
206/// built for an earlier level would otherwise stage and fail on its first call.
207pub(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
219/// [`RuntimeError::WitLevelMismatch`] for the refused names among `exports` and `imports` that do
220/// not fit `level`, or `None` when every one of them does.
221fn 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    // One untagged payload: the platform never decides a component's fit.
230    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
241/// Checks that every variable `description.required_env` names is declared in
242/// `capabilities.env.allow`, given as `env_allow`. Names match exactly.
243///
244/// The refusal lists every undeclared name, in the driver's order, each once.
245pub 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
268/// The store a process driver runs in: an empty WASI context and the execution limits.
269struct 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
284/// One instantiated process driver. The runtime keeps one for a whole run, so the driver may
285/// carry what `launch` was given into `parse`.
286pub struct ProcessDriver {
287    name: String,
288    version: String,
289    store: Store<ProcessDriverStoreState>,
290    bindings: ProcessDriverBindings,
291    /// The session's latch, where a call the wall-clock bound stops is recorded.
292    call_deadlines: CallDeadlines,
293    /// The call the wall-clock bound stopped, once one has, and its deadline in seconds. The
294    /// stopped call's future was dropped mid-guest, so the instance is never entered again.
295    cut: Option<(&'static str, u64)>,
296}
297
298impl ProcessDriver {
299    /// Instantiates `component` with no grants: a WASI context with no environment, arguments,
300    /// preopened directories, inherited stdio or network, and no Murmur host import.
301    ///
302    /// Each call into the driver is bounded by the hook call limits
303    /// ([`ExecutionLimits::for_hook_calls`] with no declared deadline), so `engine` must have an
304    /// epoch ticker running, and to the same wall-clock bound, recorded in `call_deadlines`.
305    /// `name` and `version` identify the driver in errors. Any failure, including an import the
306    /// empty context does not provide, is [`RuntimeError::ProcessDriverLoad`] carrying wasmtime's
307    /// message.
308    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            // The builder's defaults are the grant: no environment, arguments, preopens or
328            // stdio, no name lookup, and a socket address check that refuses every address.
329            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    /// Calls the driver's `describe`.
359    ///
360    /// A trap, an exceeded deadline, or a `required-env` entry that is not a usable variable
361    /// name (empty, or containing `=` or NUL) is [`RuntimeError::ProcessDriverLoad`].
362    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    /// Asks the driver to plan one harness run. The inner `Err` is the driver's own refusal,
402    /// reported to the operator verbatim; the outer one is a trap or an exceeded deadline.
403    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    /// Reads one batch of complete stdout lines into events. The same instance serves a whole
428    /// run, so the driver may answer from what `launch` was given.
429    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    /// Asks the driver what a run whose output ended without a terminal event amounts to.
451    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    /// Gives the next call its own deadline, and returns the limits it runs under. Without this
473    /// the ticks the previous call spent would count against it, so a long run would eventually
474    /// interrupt a driver that had done nothing wrong.
475    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    /// Refuses a call into an instance the wall-clock bound stopped earlier, naming that call.
482    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    /// A bound call's outcome: the call's own result, or, when the wall-clock bound stopped it,
493    /// the instance marked cut and the text saying so.
494    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
537/// What a process driver phase the wall-clock bound stopped reports, in the wording of the
538/// epoch's "exceeded its …s deadline and was interrupted": the driver's deadline is the hook
539/// default, which no manifest key sets.
540fn 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
547/// Whether `name` can be set as an environment variable.
548///
549/// A name carrying `=` would set a second variable when an environment is built, and one carrying
550/// NUL truncates it; an empty name is neither. Both places a driver names a variable — its
551/// `required-env`, checked at load, and its launch plan's `env-set`, checked at spawn — hold to
552/// this one rule.
553pub(crate) fn is_usable_env_name(name: &str) -> bool {
554    !name.is_empty() && !name.contains('=') && !name.contains('\0')
555}
556
557/// The first `required-env` name that could not be set as an environment variable, if any.
558/// Rejected where the driver's word is first taken, rather than wherever it is later spent.
559fn 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
567/// The name and version the `*_process_driver_wasm` helpers give a driver in errors, since bare
568/// bytes carry no artifact name.
569const STANDALONE_NAME: &str = "<wasm>";
570const STANDALONE_VERSION: &str = "";
571
572/// `wasm` compiled on a fresh engine with its own epoch ticker, and a current-thread runtime to
573/// drive one instance of it on: everything the `*_process_driver_wasm` helpers share.
574struct 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            // No session trace exists to drain a firing into.
612            CallDeadlines::default(),
613        )
614        .await
615    }
616}
617
618/// Compiles `wasm` as a process driver on a fresh engine, instantiates it with no grants, and
619/// returns what its `describe` reports.
620///
621/// For tests and tools that hold a driver's bytes outside any capsule. It blocks on its own
622/// current-thread runtime, so it must not be called from inside one. Errors name the driver as
623/// `<wasm>` with an empty version, since the bytes carry no artifact name.
624pub 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
632/// Compiles `wasm` as a process driver the way [`describe_process_driver_wasm`] does, then calls
633/// `launch` with `request` and `parse` with `lines` on that one instance, as a run would.
634///
635/// Returns the driver's launch answer — its own refusal is the inner `Err` — and the events
636/// `parse` read. `parse` is called whether or not `launch` refused, so a driver's line reading
637/// can be checked on its own. The same blocking and error-naming rules apply.
638pub 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    /// Level 1, the committed level before the one this host serves, which has
717    /// `murmur:driver@0.3.0`.
718    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    /// An import at the served version of `murmur:stream` under an interface the package does not
822    /// define fits the level, so it is not a level mismatch: the interface does not exist.
823    #[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        // With a level mismatch beside it, the mismatch is what is reported.
837        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    /// What one `parse` call costs, and what instantiating a driver costs.
973    ///
974    /// Ignored: it measures rather than asserts, and the numbers only mean anything in a release
975    /// build. Run it with
976    /// `cargo test -p capsule-runtime --release process_driver_parse_cost -- --ignored --nocapture`.
977    #[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        // The fixture strips the `<server-name>__` prefix it remembered in `launch`, so a bare
1107        // name here shows `parse` ran on the instance `launch` configured.
1108        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}