feat(vm): expose cancellable invocation item streams

Author: fffonion

crates/rustscript/tests/alias_smoke.rs

@@ -27,17 +27,30 @@ fn alias_exports_op_code() {
 
 #[cfg(feature = "runtime")]
 #[test]
-fn alias_exports_public_runtime_event_contract() {
-    fn accept_sink<S: rustscript::EventSink>(_sink: S) {}
-
-    struct Sink;
-    impl rustscript::EventSink for Sink {
-        fn emit(&mut self, _payload: rustscript::EventPayload) -> rustscript::RuntimeResult<()> {
-            Ok(())
-        }
-    }
+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),
+    ))));
 
-    accept_sink(Sink);
+    fn accept_error(_error: rustscript::InvocationError) {}
+    accept_error(rustscript::InvocationError::Cancelled(
+        rustscript::CancellationReason::Requested,
+    ));
+    accept_error(rustscript::InvocationError::Host {
+        message: "boom".to_string(),
+    });
 }
 
 #[cfg(feature = "sqlite")]

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, 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; 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 `CancellationReason`, and the low-level `Vm::run` pump is unchanged for custom drivers.
+
 ## 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.

src/builtins/runtime/context.rs

@@ -1,15 +1,10 @@
-use super::error::{RuntimeError, RuntimeErrorCode, RuntimeResult};
-use super::event::{EventEmitter, EventLimits, EventReceipt, EventSink};
-use crate::vm::{Value, VmResult};
+use super::error::RuntimeResult;
+use super::event::EventLimits;
 
-pub const RUNTIME_INPUT_NAME: &str = "runtime::input";
 #[allow(dead_code)]
-pub const RUNTIME_EMIT_NAME: &str = "runtime::emit";
+pub const STREAM_EMIT_NAME: &str = "stream::emit";
 
-#[allow(dead_code)]
-pub type RuntimeEventSink = dyn EventSink;
-
-/// Configuration for one VM/run-scoped generic runtime context.
+/// Configuration for one VM/run-scoped invocation stream.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct RuntimeContextConfig {
     event_limits: EventLimits,
@@ -31,74 +26,29 @@ impl Default for RuntimeContextConfig {
     }
 }
 
-/// Run-scoped input and generic event transport hooks.
+/// Run-scoped invocation stream configuration.
 ///
-/// The context stores values as VM [`Value`]s and delegates event persistence/delivery to the
-/// embedding. It has no knowledge of sessions, providers, platforms, or event names.
+/// The context carries only the per-item event bound. Event values are owned by
+/// the active invocation's single pending-event slot; there is no ambient
+/// input, no embedding event sink, and no sequence or persistence policy here.
 pub struct RuntimeContext {
-    input: Option<Value>,
-    events: EventEmitter,
+    event_limits: EventLimits,
 }
 
 #[allow(dead_code)]
 impl RuntimeContext {
     pub fn with_config(config: RuntimeContextConfig) -> RuntimeResult<Self> {
         Ok(Self {
-            input: None,
-            events: EventEmitter::new(config.event_limits()),
+            event_limits: config.event_limits(),
         })
     }
 
     pub fn config(&self) -> RuntimeContextConfig {
-        RuntimeContextConfig::new(self.events.limits())
-    }
-
-    pub fn set_input(&mut self, value: Value) -> RuntimeResult<()> {
-        self.input = Some(value);
-        Ok(())
-    }
-
-    pub fn clear_input(&mut self) {
-        self.input = None;
-    }
-
-    pub fn reset_for_reuse(&mut self) {
-        self.input = None;
-        self.events.reset_for_reuse();
-    }
-
-    pub fn input(&self) -> RuntimeResult<Value> {
-        self.input.clone().ok_or_else(|| {
-            RuntimeError::new(
-                RuntimeErrorCode::InputUnavailable,
-                RUNTIME_INPUT_NAME,
-                "run input has not been configured",
-            )
-        })
-    }
-
-    pub fn set_event_sink<S>(&mut self, sink: S) -> RuntimeResult<()>
-    where
-        S: EventSink + 'static,
-    {
-        self.events.set_sink(sink);
-        Ok(())
-    }
-
-    pub fn clear_event_sink(&mut self) {
-        self.events.clear_sink();
-    }
-
-    pub fn emit(&mut self, value: Value) -> RuntimeResult<EventReceipt> {
-        self.events.emit(value)
-    }
-
-    pub fn emitted_events(&self) -> u64 {
-        self.events.emitted_events()
+        RuntimeContextConfig::new(self.event_limits)
     }
 
     pub fn event_limits(&self) -> EventLimits {
-        self.events.limits()
+        self.event_limits
     }
 }
 
@@ -109,29 +59,22 @@ impl Default for RuntimeContext {
     }
 }
 
-/// Parent registration helper for the zero-argument `runtime::input()` host function.
-pub fn runtime_input(context: &RuntimeContext) -> VmResult<Value> {
-    context
-        .input()
-        .map_err(|error| crate::vm::VmError::HostError(error.to_string()))
-}
-
-/// Parent registration helper for the one-argument `runtime::emit(value)` host function.
-pub fn runtime_emit(context: &mut RuntimeContext, value: Value) -> VmResult<()> {
-    context
-        .emit(value)
-        .map(|_| ())
-        .map_err(|error| crate::vm::VmError::HostError(error.to_string()))
-}
-
 #[cfg(test)]
 mod tests {
-    use super::{RUNTIME_EMIT_NAME, RUNTIME_INPUT_NAME, RuntimeContext};
+    use super::{EventLimits, RuntimeContext, RuntimeContextConfig, STREAM_EMIT_NAME};
 
     #[test]
-    fn host_names_are_generic_and_stable() {
-        assert_eq!(RUNTIME_INPUT_NAME, "runtime::input");
-        assert_eq!(RUNTIME_EMIT_NAME, "runtime::emit");
+    fn host_name_is_generic_and_stable() {
+        assert_eq!(STREAM_EMIT_NAME, "stream::emit");
         assert!(std::mem::size_of::<RuntimeContext>() > 0);
     }
+
+    #[test]
+    fn per_item_event_limits_are_configurable() {
+        let limits = EventLimits::new(128, 4).expect("limits should be valid");
+        let context = RuntimeContext::with_config(RuntimeContextConfig::new(limits))
+            .expect("context should be constructible");
+        assert_eq!(context.event_limits(), limits);
+        assert_eq!(context.config().event_limits(), limits);
+    }
 }

src/builtins/runtime/context_host.rs

@@ -1,29 +1,12 @@
 use pd_host_function::pd_host_function;
 
 use super::AnyValue;
-use crate::vm::{Value, Vm, VmResult};
-
-/// Returns the embedding-provided input for the current run.
-#[pd_host_function(name = "runtime::input")]
-fn runtime_input_impl(vm: &mut Vm) -> VmResult<AnyValue> {
-    vm.runtime_input_value()
-}
-
-/// Returns the run-scoped input encoded with the runtime's strict JSON contract.
-#[pd_host_function(name = "runtime::input_json")]
-fn runtime_input_json_impl(vm: &mut Vm) -> VmResult<String> {
-    let value = vm.runtime_input_value()?;
-    super::json::encode_value_to_string(&value)
-}
-
-/// Emits one bounded event without changing the script return value.
-#[pd_host_function(name = "runtime::emit")]
-fn runtime_emit_impl(vm: &mut Vm, value: AnyValue) -> VmResult<()> {
-    vm.emit_runtime_event(value)
-}
-
-/// Emits one JSON text event for strict RSS boundary adapters.
-#[pd_host_function(name = "runtime::emit_json")]
-fn runtime_emit_json_impl(vm: &mut Vm, value: &str) -> VmResult<()> {
-    vm.emit_runtime_event(Value::string(value))
+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

@@ -8,12 +8,8 @@ pub type RuntimeResult<T> = Result<T, RuntimeError>;
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum RuntimeErrorCode {
     InvalidConfiguration,
-    InputUnavailable,
-    EventSinkUnavailable,
     EventPayloadTooLarge,
     EventDepthExceeded,
-    EventSequenceExhausted,
-    EventSinkRejected,
     ResourceLimitExceeded,
     InvalidResourceHandle,
     ResourceHandleWrongTable,
@@ -35,12 +31,8 @@ impl RuntimeErrorCode {
     pub const fn as_str(self) -> &'static str {
         match self {
             Self::InvalidConfiguration => "invalid_configuration",
-            Self::InputUnavailable => "input_unavailable",
-            Self::EventSinkUnavailable => "event_sink_unavailable",
             Self::EventPayloadTooLarge => "event_payload_too_large",
             Self::EventDepthExceeded => "event_depth_exceeded",
-            Self::EventSequenceExhausted => "event_sequence_exhausted",
-            Self::EventSinkRejected => "event_sink_rejected",
             Self::ResourceLimitExceeded => "resource_limit_exceeded",
             Self::InvalidResourceHandle => "invalid_resource_handle",
             Self::ResourceHandleWrongTable => "resource_handle_wrong_table",
@@ -150,14 +142,14 @@ mod tests {
     fn structured_error_preserves_code_and_fields() {
         let error = RuntimeError::new(
             RuntimeErrorCode::EventPayloadTooLarge,
-            "runtime::emit",
+            "stream::emit",
             "event payload exceeds the configured bound",
         )
         .with_limit(32)
         .with_value(64);
 
         assert_eq!(error.code(), RuntimeErrorCode::EventPayloadTooLarge);
-        assert_eq!(error.operation(), "runtime::emit");
+        assert_eq!(error.operation(), "stream::emit");
         assert_eq!(error.limit(), Some(32));
         assert_eq!(error.value(), Some(64));
         assert!(error.to_string().contains("event_payload_too_large"));