feat(vm): generic operation/scope consumption for invocation, run, reset and pool

Author: fffonion

build.rs

@@ -166,11 +166,18 @@ fn main() {
         catalog.retain(|entry| !entry.source_name.starts_with("sqlite::"));
     }
 
-    let host_sources = [SourceSpec {
-        path: "src/builtins/runtime/host.rs".to_string(),
-        module: "host".to_string(),
-        category: SourceCategory::DefaultHost,
-    }];
+    let host_sources = vec![
+        SourceSpec {
+            path: "src/builtins/runtime/host.rs".to_string(),
+            module: "host".to_string(),
+            category: SourceCategory::DefaultHost,
+        },
+        SourceSpec {
+            path: "src/builtins/runtime/context_host.rs".to_string(),
+            module: "context_host".to_string(),
+            category: SourceCategory::DefaultHost,
+        },
+    ];
     let async_enabled = env::var_os("CARGO_FEATURE_ASYNC").is_some();
     let target_arch = env::var("CARGO_CFG_TARGET_ARCH").expect("missing target architecture");
     let builtin_sources = builtin_source_specs(&namespaces, async_enabled, &target_arch);

crates/rustscript/tests/alias_smoke.rs

@@ -21,3 +21,31 @@ fn alias_exports_op_code() {
     let _ = rustscript::OpCode::Nop;
     let _ = rustscript::OpCode::Add;
 }
+
+#[cfg(feature = "runtime")]
+#[test]
+fn alias_exports_public_invocation_stream_contract() {
+    fn accept_item(_item: rustscript::InvocationItem) {}
+
+    accept_item(rustscript::InvocationItem::Complete(
+        rustscript::Value::Null,
+    ));
+    accept_item(rustscript::InvocationItem::Event(rustscript::Value::Bool(
+        true,
+    )));
+
+    fn accept_poll(_poll: rustscript::InvocationPoll) {}
+    accept_poll(rustscript::InvocationPoll::Pending);
+    accept_poll(rustscript::InvocationPoll::Ready(None));
+    accept_poll(rustscript::InvocationPoll::Ready(Some(Ok(
+        rustscript::InvocationItem::Complete(rustscript::Value::Null),
+    ))));
+
+    fn accept_error(_error: rustscript::InvocationError) {}
+    accept_error(rustscript::InvocationError::Cancelled(
+        rustscript::operation::OperationCancelReason::Requested,
+    ));
+    accept_error(rustscript::InvocationError::Host {
+        message: "boom".to_string(),
+    });
+}

docs/callable-runtime.md

@@ -43,6 +43,18 @@ Reset clears Program runtime values and rebinds root function items from Program
 
 PDRC recordings preserve full execution-frame metadata. Callable environments use identity-table encoding, so aliases still share one environment after decode.
 
+## Invocation item stream
+
+`Vm::start_invocation` starts one exported callable with ordinary `Value` arguments and returns an `Invocation` handle that behaves like a fused `Stream<Item = Result<InvocationItem, InvocationError>>`:
+
+- `InvocationItem::Event(value)` items arrive in order for each `stream::emit(value)` call; `stream::emit` still evaluates to `()` inside RSS.
+- exactly one `InvocationItem::Complete(value)` carries the callable return value; events never replace it;
+- cancellation, fuel exhaustion, epoch deadline expiry, runtime capability failures (including event payload bound violations), and host failures each produce exactly one typed `InvocationError` item;
+- every poll after `Complete` or the error item returns `Ready(None)` (fused end of stream);
+- `InvocationPoll::Pending` means the VM is paused on an outstanding host operation; drive it through the embedding-owned async bridge and poll again.
+
+Polling drives execution and provides backpressure: at most one event item is buffered between polls, and the VM does not produce items while the consumer is not polling. `stream::emit` validates only the configured per-item value bound (payload bytes and nesting depth); sequence assignment, receipts, persistence, and delivery policy belong to the embedding. At most one invocation is active per VM, `Invocation::cancel(reason)` cancels with a typed `OperationCancelReason`, dropping the handle retires the invocation synchronously for immediate VM reuse, and the low-level `Vm::run` pump is unchanged for custom drivers. VM reset uses the generic execution-scope close boundary; a pending close keeps the old scope installed and blocks reuse until `poll_reset_for_reuse` reports quiescence.
+
 ## Optimized backends
 
 Whole-program AOT and Trace JIT use the same builtin call path (static catalog IDs) for environment binding and native frame dispatch for `callvalue`. Script-frame entry and return preserve frame-relative locals and typed continuations.

examples/collection_rebind_bench.rs

@@ -222,7 +222,7 @@ fn measure(
     let mut samples = Vec::with_capacity(config.samples);
     let mut generic_builtin_calls = 0u64;
     for _ in 0..config.samples {
-        vm.reset_for_reuse();
+        let _ = vm.reset_for_reuse();
         let native_execs_before_sample = vm.jit_native_exec_count();
         let started = Instant::now();
         let status = vm

examples/mini_bench.rs

@@ -569,7 +569,7 @@ fn measure_runtime_mode(
         let mut vm = Vm::new(program.clone());
         configure_vm_for_mode(&mut vm, mode);
         warm_vm_for_mode(&mut vm, mode, expected_stack)?;
-        vm.reset_for_reuse();
+        let _ = vm.reset_for_reuse();
         let started = Instant::now();
         let status = vm
             .run()

pd-vm-wasm/src/runtime.rs

@@ -11,9 +11,9 @@ use std::time::Instant;
 
 use serde::Deserialize;
 use vm::{
-    CallOutcome, CallReturn, FunctionDecl, HostAsyncBridge, HostFunction, HostOpId, LocalInfo,
-    SourceFlavor, SourcePathError, Value, Vm, VmError, VmResult, VmStatus, VmYieldReason,
-    compile_source_with_flavor_and_options, format_value, render_vm_error,
+    CallOutcome, CallReturn, FunctionDecl, HostAsyncBridge, HostAsyncOpTerminal, HostFunction,
+    HostOpId, LocalInfo, SourceFlavor, SourcePathError, Value, Vm, VmError, VmResult, VmStatus,
+    VmYieldReason, compile_source_with_flavor_and_options, format_value, render_vm_error,
 };
 
 use crate::analyzer::{LintDiagnostic, lint_source_with_flavor, lint_success_diagnostics};
@@ -294,6 +294,34 @@ impl HostAsyncBridge for BrowserAsyncBridge {
             Poll::Pending
         }
     }
+
+    fn request_cancel_op(
+        &mut self,
+        op_id: HostOpId,
+        _reason: vm::operation::OperationCancelReason,
+    ) -> VmResult<()> {
+        let Ok(mut state) = self.state.lock() else {
+            return Err(VmError::HostError(
+                "browser async bridge state is unavailable".to_string(),
+            ));
+        };
+        state.deadlines_ms.remove(&op_id);
+        Ok(())
+    }
+
+    fn poll_cancel_op(&mut self, _op_id: HostOpId, _cx: &mut Context<'_>) -> Poll<VmResult<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn cleanup_op(&mut self, op_id: HostOpId, _terminal: HostAsyncOpTerminal) -> VmResult<()> {
+        let Ok(mut state) = self.state.lock() else {
+            return Err(VmError::HostError(
+                "browser async bridge state is unavailable".to_string(),
+            ));
+        };
+        state.deadlines_ms.remove(&op_id);
+        Ok(())
+    }
 }
 
 struct PlaygroundRuntimeSleepHostFunction {
@@ -1323,7 +1351,8 @@ fn register_functions(
         .any(|decl| decl.name == "runtime::sleep")
         .then(|| {
             let state = Arc::new(Mutex::new(BrowserAsyncState::default()));
-            vm.set_async_bridge(Box::new(BrowserAsyncBridge::new(Arc::clone(&state))));
+            vm.set_async_bridge(Box::new(BrowserAsyncBridge::new(Arc::clone(&state))))
+                .expect("browser async bridge should install");
             state
         });
     for decl in functions {

src/builtins/runtime/context.rs

@@ -0,0 +1,3 @@
+//! Compatibility re-exports for the adapter runtime context.
+
+pub(crate) use crate::vm::runtime::{RuntimeContext, RuntimeContextConfig, STREAM_EMIT_NAME};

src/builtins/runtime/context_host.rs

@@ -0,0 +1,12 @@
+use pd_host_function::pd_host_function;
+
+use super::AnyValue;
+use crate::vm::{CallOutcome, Vm, VmResult};
+
+/// Places one bounded event item on the active invocation stream and yields
+/// control to the invocation poller. `stream::emit` still evaluates to `()`
+/// inside RSS.
+#[pd_host_function(name = "stream::emit")]
+fn stream_emit_impl(vm: &mut Vm, value: AnyValue) -> VmResult<CallOutcome> {
+    vm.emit_stream_item(value)
+}

src/builtins/runtime/error.rs

@@ -0,0 +1,3 @@
+//! Compatibility re-exports for the adapter runtime error API.
+
+pub use crate::vm::runtime::{RuntimeError, RuntimeErrorCode, RuntimeResult};

src/builtins/runtime/event.rs

@@ -0,0 +1,7 @@
+//! Compatibility re-exports for invocation event validation.
+
+#[allow(unused_imports)]
+pub use crate::vm::runtime::{
+    DEFAULT_MAX_EVENT_DEPTH, DEFAULT_MAX_EVENT_PAYLOAD_BYTES, EventLimits, EventPayload,
+    estimate_value_size,
+};