feat(vm): expose cancellable invocation item streams

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 builtin_sources = builtin_source_specs(&namespaces);
     let core_sources = [SourceSpec {
         path: "src/builtins/runtime/core.rs".to_string(),

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.
+
 ## 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

@@ -0,0 +1,74 @@
+//! Run-scoped invocation stream configuration.
+//!
+//! The [`RuntimeContext`] carries only the per-item event bound applied by
+//! `stream::emit`. 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.
+
+use super::error::RuntimeResult;
+use super::event::EventLimits;
+
+/// The authoritative `stream::emit` builtin identity.
+#[allow(dead_code)]
+pub const STREAM_EMIT_NAME: &str = "stream::emit";
+
+/// Configuration for one VM/run-scoped invocation stream.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct RuntimeContextConfig {
+    event_limits: EventLimits,
+}
+
+#[allow(dead_code)]
+impl RuntimeContextConfig {
+    pub fn new(event_limits: EventLimits) -> Self {
+        Self { event_limits }
+    }
+
+    #[allow(dead_code)]
+    pub const fn event_limits(self) -> EventLimits {
+        self.event_limits
+    }
+}
+
+/// Run-scoped invocation stream configuration.
+#[derive(Debug, Default)]
+pub struct RuntimeContext {
+    event_limits: EventLimits,
+}
+
+#[allow(dead_code)]
+impl RuntimeContext {
+    pub fn with_config(config: RuntimeContextConfig) -> RuntimeResult<Self> {
+        Ok(Self {
+            event_limits: config.event_limits,
+        })
+    }
+
+    pub fn config(&self) -> RuntimeContextConfig {
+        RuntimeContextConfig::new(self.event_limits)
+    }
+
+    pub fn event_limits(&self) -> EventLimits {
+        self.event_limits
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{EventLimits, RuntimeContext, RuntimeContextConfig, STREAM_EMIT_NAME};
+
+    #[test]
+    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

@@ -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,138 @@
+//! Structured runtime error types shared by the invocation stream.
+//!
+//! A [`RuntimeError`] carries a stable machine-readable [`RuntimeErrorCode`],
+//! the offending builtin operation name, and optional numeric limit/value
+//! fields. The invocation stream preserves these instead of flattening them
+//! to a string, so an embedding can branch on the code and inspect the
+//! numeric state (payload bytes, depth) without string matching.
+
+use std::fmt;
+
+/// Result alias used by runtime builtin surfaces.
+pub type RuntimeResult<T> = Result<T, RuntimeError>;
+
+/// Stable machine-readable runtime error codes.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RuntimeErrorCode {
+    InvalidConfiguration,
+    EventPayloadTooLarge,
+    EventDepthExceeded,
+    ResourceLimitExceeded,
+    InvalidResourceHandle,
+    ResourceHandleWrongTable,
+    OperationFailed,
+    OperationAlreadyTerminal,
+    OperationCancelled,
+    SyncResourceUnavailable,
+    CloseFailed,
+}
+
+impl RuntimeErrorCode {
+    /// Stable snake_case string form, used for transport and tests.
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::InvalidConfiguration => "invalid_configuration",
+            Self::EventPayloadTooLarge => "event_payload_too_large",
+            Self::EventDepthExceeded => "event_depth_exceeded",
+            Self::ResourceLimitExceeded => "resource_limit_exceeded",
+            Self::InvalidResourceHandle => "invalid_resource_handle",
+            Self::ResourceHandleWrongTable => "resource_handle_wrong_table",
+            Self::OperationFailed => "operation_failed",
+            Self::OperationAlreadyTerminal => "operation_already_terminal",
+            Self::OperationCancelled => "operation_cancelled",
+            Self::SyncResourceUnavailable => "sync_resource_unavailable",
+            Self::CloseFailed => "close_failed",
+        }
+    }
+}
+
+/// A structured runtime error with a stable code and optional numeric state.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RuntimeError {
+    code: RuntimeErrorCode,
+    operation: String,
+    message: String,
+    limit: Option<u64>,
+    value: Option<u64>,
+}
+
+impl RuntimeError {
+    pub fn new(code: RuntimeErrorCode, operation: &str, message: impl Into<String>) -> Self {
+        Self {
+            code,
+            operation: operation.to_string(),
+            message: message.into(),
+            limit: None,
+            value: None,
+        }
+    }
+
+    /// Attaches the configured bound that was violated.
+    pub fn with_limit(mut self, limit: usize) -> Self {
+        self.limit = Some(limit as u64);
+        self
+    }
+
+    /// Attaches the offending value (for example the measured payload size).
+    pub fn with_value(mut self, value: usize) -> Self {
+        self.value = Some(value as u64);
+        self
+    }
+
+    pub fn code(&self) -> RuntimeErrorCode {
+        self.code
+    }
+
+    pub fn operation(&self) -> &str {
+        &self.operation
+    }
+
+    pub fn limit(&self) -> Option<u64> {
+        self.limit
+    }
+
+    pub fn value(&self) -> Option<u64> {
+        self.value
+    }
+
+    pub fn message(&self) -> &str {
+        &self.message
+    }
+}
+
+impl fmt::Display for RuntimeError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}: {}", self.code.as_str(), self.message)?;
+        if let Some(limit) = self.limit {
+            write!(f, " (limit {limit})")?;
+        }
+        if let Some(value) = self.value {
+            write!(f, " (value {value})")?;
+        }
+        Ok(())
+    }
+}
+
+impl std::error::Error for RuntimeError {}
+
+#[cfg(test)]
+mod tests {
+    use super::{RuntimeError, RuntimeErrorCode};
+
+    #[test]
+    fn structured_error_preserves_code_and_fields() {
+        let error = RuntimeError::new(
+            RuntimeErrorCode::EventPayloadTooLarge,
+            "stream::emit",
+            "event payload exceeds the configured bound",
+        )
+        .with_limit(32)
+        .with_value(64);
+
+        assert_eq!(error.code(), RuntimeErrorCode::EventPayloadTooLarge);
+        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"));
+    }
+}