diff --git a/crates/rustscript/tests/alias_smoke.rs b/crates/rustscript/tests/alias_smoke.rs index af2b24a8..5d6f0494 100644 --- a/crates/rustscript/tests/alias_smoke.rs +++ b/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(_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 = "http-client")] diff --git a/docs/callable-runtime.md b/docs/callable-runtime.md index d71033d8..66aa519f 100644 --- a/docs/callable-runtime.md +++ b/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>`: + +- `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. diff --git a/src/builtins/runtime/context.rs b/src/builtins/runtime/context.rs index 2588e23b..5fa4bfa9 100644 --- a/src/builtins/runtime/context.rs +++ b/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, - events: EventEmitter, + event_limits: EventLimits, } #[allow(dead_code)] impl RuntimeContext { pub fn with_config(config: RuntimeContextConfig) -> RuntimeResult { 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 { - self.input.clone().ok_or_else(|| { - RuntimeError::new( - RuntimeErrorCode::InputUnavailable, - RUNTIME_INPUT_NAME, - "run input has not been configured", - ) - }) - } - - pub fn set_event_sink(&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 { - 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 { - 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::() > 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); + } } diff --git a/src/builtins/runtime/context_host.rs b/src/builtins/runtime/context_host.rs index d4c32b0f..0cc3ba32 100644 --- a/src/builtins/runtime/context_host.rs +++ b/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 { - 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 { - 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 { + vm.emit_stream_item(value) } diff --git a/src/builtins/runtime/error.rs b/src/builtins/runtime/error.rs index 506ccf52..82ab3c31 100644 --- a/src/builtins/runtime/error.rs +++ b/src/builtins/runtime/error.rs @@ -8,12 +8,8 @@ pub type RuntimeResult = Result; #[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")); diff --git a/src/builtins/runtime/event.rs b/src/builtins/runtime/event.rs index cc4a3e94..c91a95ec 100644 --- a/src/builtins/runtime/event.rs +++ b/src/builtins/runtime/event.rs @@ -4,46 +4,32 @@ use super::error::{RuntimeError, RuntimeErrorCode, RuntimeResult}; pub const DEFAULT_MAX_EVENT_PAYLOAD_BYTES: usize = 64 * 1024; pub const DEFAULT_MAX_EVENT_DEPTH: usize = 64; -pub const DEFAULT_MAX_EVENTS: u64 = 1_024; -pub const DEFAULT_MAX_EVENT_BYTES: usize = 16 * 1024 * 1024; -/// Bounds applied before an event is handed to an embedding-owned sink. +/// Per-item bounds applied to one `stream::emit(value)` call. +/// +/// The core validates only this per-item value bound before placing the value +/// in the active invocation's single pending-event slot. Sequence assignment, +/// cumulative byte accounting, event receipts, and embedding-owned sinks are +/// not part of the core contract; delivery policy belongs to the embedding. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct EventLimits { max_payload_bytes: usize, max_depth: usize, - max_events: u64, - max_total_bytes: usize, } +#[allow(dead_code)] impl EventLimits { pub fn new(max_payload_bytes: usize, max_depth: usize) -> RuntimeResult { - Self::with_budget( - max_payload_bytes, - max_depth, - DEFAULT_MAX_EVENTS, - DEFAULT_MAX_EVENT_BYTES, - ) - } - - pub fn with_budget( - max_payload_bytes: usize, - max_depth: usize, - max_events: u64, - max_total_bytes: usize, - ) -> RuntimeResult { - if max_payload_bytes == 0 || max_depth == 0 || max_events == 0 || max_total_bytes == 0 { + if max_payload_bytes == 0 || max_depth == 0 { return Err(RuntimeError::new( RuntimeErrorCode::InvalidConfiguration, - "runtime::emit", + "stream::emit", "event payload and depth limits must be positive", )); } Ok(Self { max_payload_bytes, max_depth, - max_events, - max_total_bytes, }) } @@ -54,14 +40,6 @@ impl EventLimits { pub const fn max_depth(self) -> usize { self.max_depth } - - pub const fn max_events(self) -> u64 { - self.max_events - } - - pub const fn max_total_bytes(self) -> usize { - self.max_total_bytes - } } impl Default for EventLimits { @@ -69,13 +47,11 @@ impl Default for EventLimits { Self { max_payload_bytes: DEFAULT_MAX_EVENT_PAYLOAD_BYTES, max_depth: DEFAULT_MAX_EVENT_DEPTH, - max_events: DEFAULT_MAX_EVENTS, - max_total_bytes: DEFAULT_MAX_EVENT_BYTES, } } } -/// An event value whose size and nesting have already been checked. +/// An event value whose per-item bound has been validated. #[derive(Clone, Debug, PartialEq)] pub struct EventPayload { value: Value, @@ -88,10 +64,7 @@ impl EventPayload { Ok(Self { value, size_bytes }) } - pub fn value(&self) -> &Value { - &self.value - } - + #[allow(dead_code)] pub fn size_bytes(&self) -> usize { self.size_bytes } @@ -101,148 +74,7 @@ impl EventPayload { } } -/// Embedding-owned transport hook for bounded runtime events. -pub trait EventSink: Send { - fn emit(&mut self, payload: EventPayload) -> RuntimeResult<()>; -} - -impl EventSink for F -where - F: FnMut(EventPayload) -> RuntimeResult<()> + Send + 'static, -{ - fn emit(&mut self, payload: EventPayload) -> RuntimeResult<()> { - self(payload) - } -} - -/// Validates and forwards generic values without attaching agent or platform semantics. -pub struct EventEmitter { - limits: EventLimits, - sink: Option>, - emitted_events: u64, - emitted_bytes: usize, -} - -#[allow(dead_code)] -impl EventEmitter { - pub fn new(limits: EventLimits) -> Self { - Self { - limits, - sink: None, - emitted_events: 0, - emitted_bytes: 0, - } - } - - pub fn limits(&self) -> EventLimits { - self.limits - } - - pub fn set_sink(&mut self, sink: S) - where - S: EventSink + 'static, - { - self.sink = Some(Box::new(sink)); - } - - pub fn clear_sink(&mut self) { - self.sink = None; - } - - pub fn reset_for_reuse(&mut self) { - self.sink = None; - self.emitted_events = 0; - self.emitted_bytes = 0; - } - - pub fn emitted_events(&self) -> u64 { - self.emitted_events - } - - pub fn emit(&mut self, value: Value) -> RuntimeResult { - let payload = EventPayload::try_new(value, self.limits)?; - if self.emitted_events >= self.limits.max_events { - return Err(RuntimeError::new( - RuntimeErrorCode::EventSequenceExhausted, - "runtime::emit", - "event count exceeds the configured bound", - ) - .with_limit(self.limits.max_events.min(usize::MAX as u64) as usize) - .with_value(self.emitted_events)); - } - let total_bytes = self - .emitted_bytes - .checked_add(payload.size_bytes()) - .ok_or_else(|| { - RuntimeError::new( - RuntimeErrorCode::EventPayloadTooLarge, - "runtime::emit", - "cumulative event bytes overflowed", - ) - })?; - if total_bytes > self.limits.max_total_bytes { - return Err(RuntimeError::new( - RuntimeErrorCode::EventPayloadTooLarge, - "runtime::emit", - "cumulative event bytes exceed the configured bound", - ) - .with_limit(self.limits.max_total_bytes) - .with_value(total_bytes as u64)); - } - let sequence = self.emitted_events.checked_add(1).ok_or_else(|| { - RuntimeError::new( - RuntimeErrorCode::EventSequenceExhausted, - "runtime::emit", - "event sequence exhausted", - ) - })?; - let sink = self.sink.as_mut().ok_or_else(|| { - RuntimeError::new( - RuntimeErrorCode::EventSinkUnavailable, - "runtime::emit", - "an event sink has not been configured", - ) - })?; - sink.emit(payload.clone()).map_err(|error| { - RuntimeError::new( - RuntimeErrorCode::EventSinkRejected, - "runtime::emit", - error.to_string(), - ) - })?; - self.emitted_events = sequence; - self.emitted_bytes = total_bytes; - Ok(EventReceipt { - sequence, - payload_bytes: payload.size_bytes(), - }) - } -} - -impl Default for EventEmitter { - fn default() -> Self { - Self::new(EventLimits::default()) - } -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct EventReceipt { - sequence: u64, - payload_bytes: usize, -} - -#[allow(dead_code)] -impl EventReceipt { - pub const fn sequence(self) -> u64 { - self.sequence - } - - pub const fn payload_bytes(self) -> usize { - self.payload_bytes - } -} - -/// Estimates the bounded representation size used by [`EventPayload`]. +/// Estimates the bounded representation size of a value. /// /// The estimate is deliberately independent of serialization formats. It counts scalar tags, /// container headers, string/byte contents, and recursively contained values. The host transport @@ -255,7 +87,7 @@ fn measure_value(value: &Value, depth: usize, limits: EventLimits) -> RuntimeRes if depth > limits.max_depth { return Err(RuntimeError::new( RuntimeErrorCode::EventDepthExceeded, - "runtime::emit", + "stream::emit", "event payload nesting exceeds the configured bound", ) .with_limit(limits.max_depth) @@ -288,7 +120,7 @@ fn measure_value(value: &Value, depth: usize, limits: EventLimits) -> RuntimeRes if size > limits.max_payload_bytes { return Err(RuntimeError::new( RuntimeErrorCode::EventPayloadTooLarge, - "runtime::emit", + "stream::emit", "event payload exceeds the configured byte bound", ) .with_limit(limits.max_payload_bytes) @@ -305,7 +137,7 @@ fn checked_payload_add( let total = current.checked_add(additional).ok_or_else(|| { RuntimeError::new( RuntimeErrorCode::EventPayloadTooLarge, - "runtime::emit", + "stream::emit", "event payload size overflowed", ) .with_limit(limits.max_payload_bytes) @@ -313,7 +145,7 @@ fn checked_payload_add( if total > limits.max_payload_bytes { return Err(RuntimeError::new( RuntimeErrorCode::EventPayloadTooLarge, - "runtime::emit", + "stream::emit", "event payload exceeds the configured byte bound", ) .with_limit(limits.max_payload_bytes) @@ -324,22 +156,35 @@ fn checked_payload_add( #[cfg(test)] mod tests { - use super::{EventEmitter, EventLimits, EventPayload}; + use super::{EventLimits, EventPayload}; use crate::vm::Value; #[test] - fn payload_size_and_sequence_are_exposed_after_validation() { - let limits = EventLimits::new(128, 4).expect("limits should be valid"); + fn per_item_limits_validate_payload_and_depth() { + let limits = EventLimits::new(32, 4).expect("limits should be valid"); let payload = EventPayload::try_new(Value::string("event"), limits).expect("payload should fit"); assert!(payload.size_bytes() >= 5); + assert_eq!(payload.into_value(), Value::string("event")); + } - let mut emitter = EventEmitter::new(limits); - emitter.set_sink(|_| Ok(())); - let receipt = emitter - .emit(Value::string("event")) - .expect("event should be emitted"); - assert_eq!(receipt.sequence(), 1); - assert_eq!(emitter.emitted_events(), 1); + #[test] + fn oversized_or_too_deep_values_are_rejected_before_placement() { + let limits = EventLimits::new(8, 2).expect("limits should be valid"); + let too_large = EventPayload::try_new(Value::string("payload-too-large"), limits) + .expect_err("oversized event should be rejected"); + assert_eq!( + too_large.code(), + super::super::error::RuntimeErrorCode::EventPayloadTooLarge + ); + let too_deep = EventPayload::try_new( + Value::array(vec![Value::array(vec![Value::array(vec![Value::Int(1)])])]), + limits, + ) + .expect_err("too-deep event should be rejected"); + assert_eq!( + too_deep.code(), + super::super::error::RuntimeErrorCode::EventDepthExceeded + ); } } diff --git a/src/compiler/typing/context.rs b/src/compiler/typing/context.rs index efdfb045..a373ef20 100644 --- a/src/compiler/typing/context.rs +++ b/src/compiler/typing/context.rs @@ -1987,6 +1987,27 @@ impl<'a> TypeContext<'a> { } return Ok(()); } + // `stream::emit(value)` accepts any single value; the per-item event + // bound is validated at runtime by the invocation stream. The + // exemption is tied to the authoritative runtime builtin identity; a + // same-name function registered through another catalog does not + // inherit it. The identity constant lives in the `runtime`-featured + // builtins module, so in non-runtime builds the comparison is + // compiled out and the exemption does not apply. + #[cfg(feature = "runtime")] + if signature.runtime_builtin + && signature.name == crate::builtins::runtime::context::STREAM_EMIT_NAME + { + return validate_host_signature( + &signature.name, + &signature.params, + args, + state, + self, + line_context, + source_name, + ); + } if self.is_strict() && signature .params @@ -2609,3 +2630,67 @@ fn literal_int_index(key: &Expr) -> Option { }; usize::try_from(*index).ok() } + +#[cfg(test)] +mod tests { + use super::*; + use crate::builtins::{CallableParam, CallableParamType}; + + /// The authoritative `stream::emit` signature: one `any` payload. + fn emit_signature(runtime_builtin: bool) -> HostCallableSignature { + HostCallableSignature { + name: crate::builtins::runtime::context::STREAM_EMIT_NAME.to_string(), + params: vec![CallableParam { + name: "value", + ty: CallableParamType::Any, + optional: false, + }], + runtime_builtin, + } + } + + #[test] + fn stream_emit_any_payload_exemption_requires_authoritative_builtin_identity() { + let empty_impls: HashMap = HashMap::new(); + let empty_decls: HashMap = HashMap::new(); + let empty_structs: HashMap = HashMap::new(); + let empty_names: HashMap = HashMap::new(); + let empty_returns: HashMap = HashMap::new(); + let empty_signatures: HashMap = HashMap::new(); + let mut context = TypeContext::new( + &empty_impls, + &empty_decls, + &empty_structs, + &empty_names, + &empty_returns, + &empty_signatures, + TypingMode::StrictRustScript, + ); + let state = LocalTypeState::default(); + let args = [Expr::Int(1)]; + + assert!( + context + .validate_host_argument_types(&emit_signature(true), &args, &state, None, None,) + .is_ok(), + "the authoritative stream::emit builtin must accept any payload in strict mode" + ); + + // A same-name signature that is not the authoritative runtime builtin + // (for example one registered through another host catalog) must not + // inherit the strict-typing exemption. + assert!( + matches!( + context.validate_host_argument_types( + &emit_signature(false), + &args, + &state, + None, + None, + ), + Err(CompileError::StrictTypingRequired { .. }) + ), + "a same-name non-builtin signature must not inherit the stream::emit exemption" + ); + } +} diff --git a/src/compiler/typing/helpers.rs b/src/compiler/typing/helpers.rs index 28e76a8e..4c6d8a31 100644 --- a/src/compiler/typing/helpers.rs +++ b/src/compiler/typing/helpers.rs @@ -1234,6 +1234,7 @@ pub(super) fn known_host_signature(name: &str) -> Option return Some(HostCallableSignature { name: callable.name.to_string(), params: callable.signature.params.to_vec(), + runtime_builtin: true, }); } @@ -1253,6 +1254,7 @@ pub(super) fn known_host_signature(name: &str) -> Option optional: false, }) .collect(), + runtime_builtin: false, }) } diff --git a/src/compiler/typing/state.rs b/src/compiler/typing/state.rs index 5020bb27..5ce13470 100644 --- a/src/compiler/typing/state.rs +++ b/src/compiler/typing/state.rs @@ -435,4 +435,10 @@ pub(crate) struct TypeInferenceResult { pub(crate) struct HostCallableSignature { pub(crate) name: String, pub(crate) params: Vec, + /// True when this signature came from the authoritative runtime builtin + /// catalog (`default_host_callable`), false when it came from another + /// catalog such as edge ABI host functions. Strict-typing exemptions that + /// are tied to a builtin identity must check this marker so a same-name + /// function from another catalog cannot inherit them. + pub(crate) runtime_builtin: bool, } diff --git a/src/lib.rs b/src/lib.rs index 7fa63e1d..854a5edc 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -48,8 +48,6 @@ pub fn builtin_call_index(name: &str) -> Option { } #[cfg(feature = "runtime")] pub use builtins::runtime::error::{RuntimeError, RuntimeErrorCode, RuntimeResult}; -#[cfg(feature = "runtime")] -pub use builtins::runtime::event::{EventPayload, EventSink}; pub use compiler::diagnostics::{ render_compile_error, render_source_error, render_source_path_error, }; @@ -94,9 +92,10 @@ pub use vm::{ CapabilityProfileBuilder, DEFAULT_MAX_SCRIPT_CALL_DEPTH, EpochCheckpoint, EpochHandle, FuelCheckpoint, HostArgsFunction, HostAsyncBridge, HostBindingPlan, HostFunction, HostFunctionRegistry, HostFuture, HostFutureOutput, HostOpId, HostStackFunction, - IntoScriptValue, QueuedScriptInvocation, ScriptArgs, ScriptCallback, ScriptResult, - StaticHostArgsFunction, StaticHostFunction, StaticHostStackFunction, Store, Vm, VmError, - VmResult, VmStatus, VmYieldReason, + IntoScriptValue, Invocation, InvocationError, InvocationItem, InvocationPoll, + QueuedScriptInvocation, ScriptArgs, ScriptCallback, ScriptResult, StaticHostArgsFunction, + StaticHostFunction, StaticHostStackFunction, Store, Vm, VmError, VmResult, VmStatus, + VmYieldReason, }; #[cfg(feature = "runtime")] diff --git a/src/vm/host.rs b/src/vm/host.rs index 676997b4..775a3b2a 100644 --- a/src/vm/host.rs +++ b/src/vm/host.rs @@ -1092,60 +1092,6 @@ impl Vm { self.host.runtime_print_sink = None; } - pub fn set_runtime_input(&mut self, value: Value) -> VmResult<()> { - self.run_ctx - .runtime_context - .set_input(value) - .map_err(|error| VmError::HostError(error.to_string())) - } - - pub fn clear_runtime_input(&mut self) { - self.run_ctx.runtime_context.clear_input(); - } - - pub fn set_runtime_event_sink(&mut self, sink: S) -> VmResult<()> - where - S: crate::builtins::runtime::event::EventSink + 'static, - { - self.run_ctx - .runtime_context - .set_event_sink(sink) - .map_err(|error| VmError::HostError(error.to_string())) - } - - pub fn clear_runtime_event_sink(&mut self) { - self.run_ctx.runtime_context.clear_event_sink(); - } - - pub(crate) fn runtime_input_value(&self) -> VmResult { - crate::builtins::runtime::context::runtime_input(&self.run_ctx.runtime_context) - } - - pub(crate) fn emit_runtime_event(&mut self, value: Value) -> VmResult<()> { - crate::builtins::runtime::context::runtime_emit(&mut self.run_ctx.runtime_context, value) - } - - /// Configure a bounded event sink without exposing runtime implementation types. - pub fn set_runtime_value_event_sink(&mut self, mut sink: F) -> VmResult<()> - where - F: FnMut(Value) -> VmResult<()> + Send + 'static, - { - self.run_ctx - .runtime_context - .set_event_sink( - move |payload: crate::builtins::runtime::event::EventPayload| { - sink(payload.into_value()).map_err(|error| { - crate::builtins::runtime::error::RuntimeError::new( - crate::builtins::runtime::error::RuntimeErrorCode::EventSinkRejected, - "runtime::emit", - error.to_string(), - ) - }) - }, - ) - .map_err(|error| VmError::HostError(error.to_string())) - } - /// Enables or disables implicit binding of built-in host functions. /// /// Disabling this makes the VM use only explicitly registered host functions. The default diff --git a/src/vm/instance.rs b/src/vm/instance.rs index 475ff791..d74cbad0 100644 --- a/src/vm/instance.rs +++ b/src/vm/instance.rs @@ -19,6 +19,7 @@ use std::sync::{Arc, Weak}; use crate::bytecode::{CallableValue, Program, SharedCaptureCell, Value}; use crate::vm::async_host::WaitingHostOp; +use crate::vm::invocation::{InvocationPhase, InvocationState}; use crate::vm::map_iter::MapIteratorState; use crate::vm::{DEFAULT_MAX_SCRIPT_CALL_DEPTH, VmYieldReason}; @@ -84,6 +85,7 @@ pub(crate) struct Instance { pub(crate) shutdown: bool, pub(super) waiting_host_op: Option, pub(crate) last_yield_reason: Option, + pub(crate) invocation: Option, pub(crate) map_iterators: Vec>>, pub(crate) drop_contract_events_enabled: bool, pub(crate) drop_contract_events: u64, @@ -120,6 +122,7 @@ impl Instance { shutdown: false, waiting_host_op: None, last_yield_reason: None, + invocation: None, map_iterators: Vec::new(), drop_contract_events_enabled: false, drop_contract_events: 0, @@ -161,6 +164,8 @@ impl Instance { self.draining_queued_callables = false; self.shutdown = false; self.waiting_host_op = None; + self.drop_invocation_state(); + self.invocation = None; self.map_iterators.clear(); self.clear_interpreter_metrics(); } @@ -168,12 +173,32 @@ impl Instance { /// Releases interpreter-owned values with drop-contract accounting. Used by /// the facade's `Drop` (and by `shutdown`). pub(crate) fn drop_cleanup(&mut self) { + self.drop_invocation_state(); self.clear_stack_with_drop_contract(); self.capture_cells.clear(); self.shared_capture_slots.clear(); self.clear_locals_with_drop_contract(); } + /// Drops pending invocation stream values with drop-contract accounting and + /// rewinds the invocation state to a fresh, fused position. + pub(crate) fn drop_invocation_state(&mut self) { + let Some(state) = self.invocation.as_mut() else { + return; + }; + let value = match std::mem::replace(&mut state.phase, InvocationPhase::Fused) { + InvocationPhase::EventPending(value) | InvocationPhase::CompletePending(value) => { + Some(value) + } + _ => None, + }; + state.emit_yield_pending = false; + state.pending_error = None; + if let Some(value) = value { + self.drop_value_with_contract(value); + } + } + pub(crate) fn invalidate_callback_registries(&mut self) { for active in self .callback_registry_flags diff --git a/src/vm/invocation.rs b/src/vm/invocation.rs new file mode 100644 index 00000000..408fd095 --- /dev/null +++ b/src/vm/invocation.rs @@ -0,0 +1,525 @@ +//! Invocation item stream. +//! +//! One exported callable started with ordinary arguments behaves like +//! `Stream>`: zero or more +//! `Event` items produced by `stream::emit`, then exactly one `Complete` item +//! or one typed error, then a fused end of stream. Polling drives execution; +//! the VM does not produce items while the consumer is not polling, and at most +//! one event item is buffered between polls (natural backpressure). +//! +//! The invocation reuses the existing callable execution state +//! ([`Vm::start_callable`], [`Vm::run`], [`Vm::take_callable_result`]) and the +//! existing async host bridge; it does not duplicate interpreter or host loops, +//! and it does not add an executor, generator syntax, an event queue, or event +//! persistence policy. + +use std::fmt; +use std::task::{Context, Poll, Waker}; + +use crate::builtins::runtime::cancellation::{ + CancellationReason, CancellationToken, OperationId, OperationState, OperationStatus, +}; +use crate::builtins::runtime::error::RuntimeError; +use crate::vm::{CallOutcome, CallReturn, Value, Vm, VmError, VmResult, VmStatus, VmYieldReason}; + +/// One item yielded by an invocation stream. +#[derive(Clone, Debug, PartialEq)] +pub enum InvocationItem { + /// One bounded event produced by `stream::emit(value)`. + Event(Value), + /// The callable's return value; exactly one per invocation. + Complete(Value), +} + +/// Typed terminal failure of an invocation stream. +/// +/// The failure is machine-readable: cancellation keeps its reason, fuel and +/// deadline failures keep their numeric state, and runtime capability failures +/// keep their structured [`RuntimeError`] instead of being flattened to a +/// string. +#[derive(Debug)] +pub enum InvocationError { + /// The invocation was cancelled with this reason. + Cancelled(CancellationReason), + /// The configured fuel budget was exhausted. + OutOfFuel { needed: u64, remaining: u64 }, + /// The configured epoch deadline expired. + DeadlineReached { current: u64, deadline: u64 }, + /// A runtime capability failure with its machine-readable code. + Capability(RuntimeError), + /// An embedding host failure without a structured runtime code. + Host { message: String }, + /// A low-level VM failure (script error or invalid frame state). + Vm(VmError), +} + +/// Poll outcome of an invocation stream. +#[derive(Debug)] +pub enum InvocationPoll { + /// The VM is paused (waiting on a host operation or a host-driven yield); + /// drive the outstanding work and poll again. + Pending, + /// One stream item, or `None` after the fused end of stream. + Ready(Option>), +} + +/// Run-scoped state of the single active invocation on a VM. +#[derive(Debug)] +pub(crate) struct InvocationState { + pub(crate) phase: InvocationPhase, + /// True while the VM is yielded at a `stream::emit` call site whose event + /// has already been delivered. The resumed call site re-enters + /// `stream::emit` and consumes this marker instead of emitting a second + /// event for the same call. + pub(crate) emit_yield_pending: bool, + /// A structured runtime error produced by `stream::emit` validation, + /// preserved for the terminal error item without string flattening. + pub(crate) pending_error: Option, + /// Stack and frame position recorded when the invocation started, used to + /// release interpreter state on terminal failure. + pub(crate) stack_base: usize, + pub(crate) frame_count: usize, +} + +#[derive(Debug)] +pub(crate) enum InvocationPhase { + Running, + EventPending(Value), + CompletePending(Value), + ErrorPending(InvocationError), + Fused, +} + +/// One active invocation handle borrowing the VM. +/// +/// Polling drives execution; dropping the handle abandons the invocation but +/// keeps it active on the VM until it fuses (a new invocation is rejected while +/// one is active). +pub struct Invocation<'vm> { + vm: &'vm mut Vm, +} + +impl fmt::Debug for Invocation<'_> { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.debug_struct("Invocation").finish_non_exhaustive() + } +} + +impl Invocation<'_> { + /// Polls the invocation stream. + /// + /// Returns `Ready(Some(Ok(Event(value))))` for each emitted event, + /// `Ready(Some(Ok(Complete(value))))` exactly once for the callable return + /// value, `Ready(Some(Err(error)))` exactly once for a typed terminal + /// failure, and `Ready(None)` on every poll after the stream has fused. + /// `Pending` means the VM is paused on an outstanding host operation or + /// host-driven yield; drive it and poll again. + pub fn poll_next(&mut self) -> VmResult { + self.vm.poll_invocation() + } + + /// Cancels the active invocation with a typed reason. + /// + /// Outstanding owned host operations are cancelled with the same reason. + /// The next poll produces exactly one `Cancelled(reason)` error item, after + /// which the stream is fused. + pub fn cancel(&mut self, reason: CancellationReason) -> VmResult<()> { + let cancellation_result = self.vm.run_ctx.cancel(reason); + self.vm.cancel_waiting_host_op_with_reason(reason); + cancellation_result + } +} + +/// One poll step selected from the current invocation phase. +enum InvocationAction { + Cancelled, + Event, + Complete, + Error, + Fused, + Drive, +} + +impl Vm { + /// Starts one invocation of an exported callable with ordinary arguments. + /// + /// The VM must be halted (complete the root frame with [`Vm::run`] first), + /// and must not already have an active invocation. A second invocation on + /// the same VM is rejected while one is active. + pub fn start_invocation( + &mut self, + callable: Value, + args: Vec, + ) -> VmResult> { + if !matches!(callable, Value::Callable(_)) { + return Err(VmError::InvalidCallable); + } + if self + .instance + .invocation + .as_ref() + .is_some_and(|state| !matches!(state.phase, InvocationPhase::Fused)) + { + return Err(VmError::InvalidFrameState( + "an invocation is already active on this vm", + )); + } + let stack_base = self.instance.stack.len(); + let frame_count = self.instance.execution_frames.len(); + self.instance.invocation = Some(InvocationState { + phase: InvocationPhase::Running, + emit_yield_pending: false, + pending_error: None, + stack_base, + frame_count, + }); + + // A cancellation that predates the invocation terminates it + // immediately. No callable, frame, or host operation has started yet, + // so there is nothing to release here: the stream transitions + // directly to the typed error and normal error delivery releases the + // invocation exactly once when the item is consumed. + if let Some(reason) = self.run_ctx.cancellation.reason() { + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::ErrorPending(InvocationError::Cancelled(reason)); + return Ok(Invocation { vm: self }); + } + + match self.start_callable(callable, &args) { + Ok(VmStatus::Halted) => { + let result = self + .take_callable_result() + .ok_or(VmError::InvalidFrameState( + "invocation halted without a callable result", + ))?; + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::CompletePending(result); + } + Ok(VmStatus::Yielded) => { + // Either `stream::emit` placed one pending event, or the + // embedding must drive a host-owned yield; both are serviced by + // the next poll. + } + Ok(VmStatus::Waiting(_)) => {} + Err(error) => { + let error = self.map_invocation_error(error, None); + self.release_invocation(); + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::ErrorPending(error); + } + } + Ok(Invocation { vm: self }) + } + + fn poll_invocation(&mut self) -> VmResult { + loop { + let action = match self.instance.invocation.as_ref() { + Some(state) => { + // Authoritative cancellation supersedes a pending Event or + // Complete: the pending value is discarded (through the + // drop-contract path) and the stream transitions to one + // Cancelled item, then a fused end. + if self.run_ctx.cancellation.reason().is_some() + && matches!( + state.phase, + InvocationPhase::EventPending(_) | InvocationPhase::CompletePending(_) + ) + { + InvocationAction::Cancelled + } else { + match state.phase { + InvocationPhase::EventPending(_) => InvocationAction::Event, + InvocationPhase::CompletePending(_) => InvocationAction::Complete, + InvocationPhase::ErrorPending(_) => InvocationAction::Error, + InvocationPhase::Fused => InvocationAction::Fused, + InvocationPhase::Running => InvocationAction::Drive, + } + } + } + None => return Ok(InvocationPoll::Ready(None)), + }; + match action { + InvocationAction::Cancelled => { + let reason = self + .run_ctx + .cancellation + .reason() + .expect("a cancelled action requires a cancellation reason"); + let discarded = self.replace_invocation_phase(InvocationPhase::ErrorPending( + InvocationError::Cancelled(reason), + )); + match discarded { + InvocationPhase::EventPending(value) + | InvocationPhase::CompletePending(value) => { + self.drop_value_with_contract(value); + } + _ => unreachable!("the cancelled action matched a pending phase above"), + } + } + InvocationAction::Event => { + let value = match self.replace_invocation_phase(InvocationPhase::Running) { + InvocationPhase::EventPending(value) => value, + _ => unreachable!("phase matched above"), + }; + // `emit_yield_pending` stays set until the resumed call + // site re-enters `stream::emit`. + return Ok(InvocationPoll::Ready(Some(Ok(InvocationItem::Event( + value, + ))))); + } + InvocationAction::Complete => { + let value = match self.replace_invocation_phase(InvocationPhase::Fused) { + InvocationPhase::CompletePending(value) => value, + _ => unreachable!("phase matched above"), + }; + self.release_invocation(); + return Ok(InvocationPoll::Ready(Some(Ok(InvocationItem::Complete( + value, + ))))); + } + InvocationAction::Error => { + let error = match self.replace_invocation_phase(InvocationPhase::Fused) { + InvocationPhase::ErrorPending(error) => error, + _ => unreachable!("phase matched above"), + }; + self.release_invocation(); + return Ok(InvocationPoll::Ready(Some(Err(error)))); + } + InvocationAction::Fused => return Ok(InvocationPoll::Ready(None)), + InvocationAction::Drive => { + let result = self.drive_invocation(); + match result { + DriveOutcome::Continue => {} + DriveOutcome::Pending => return Ok(InvocationPoll::Pending), + DriveOutcome::Error(error) => { + self.release_invocation(); + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::ErrorPending(error); + } + } + } + } + } + } + + /// Runs the low-level pump once and folds the outcome into the invocation + /// phase. `Vm::run` itself is unchanged. + fn drive_invocation(&mut self) -> DriveOutcome { + if let Some(reason) = self.run_ctx.cancellation.reason() { + return DriveOutcome::Error(InvocationError::Cancelled(reason)); + } + match self.run() { + Ok(VmStatus::Halted) => { + let result = match self.take_callable_result() { + Some(result) => result, + None => { + return DriveOutcome::Error(InvocationError::Vm( + VmError::InvalidFrameState( + "invocation halted without a callable result", + ), + )); + } + }; + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::CompletePending(result); + DriveOutcome::Continue + } + Ok(VmStatus::Yielded) => match self.last_yield_reason() { + Some(VmYieldReason::Fuel) => DriveOutcome::Error(InvocationError::OutOfFuel { + needed: u64::from(self.run_ctx.fuel_check_interval), + remaining: self.run_ctx.fuel_remaining, + }), + Some(VmYieldReason::Epoch) => { + DriveOutcome::Error(InvocationError::DeadlineReached { + current: self.run_ctx.epoch_handle.current(), + deadline: self.run_ctx.epoch_deadline, + }) + } + _ => { + // A `stream::emit` yield leaves one pending event; any other + // host-driven yield is paused for the embedding. + let event_pending = matches!( + self.instance.invocation.as_ref().map(|state| &state.phase), + Some(InvocationPhase::EventPending(_)) + ); + if event_pending { + DriveOutcome::Continue + } else { + DriveOutcome::Pending + } + } + }, + Ok(VmStatus::Waiting(_)) => { + // Capture the waiting operation AFTER `run()`: the step may + // have registered a new host op. The operation state is + // retained before polling because failing the operation + // removes it from the registry; `map_invocation_error` must + // still be able to recover its typed `OperationStatus::Failed` + // error after the first poll clears the waiting state. + let waiting_operation = self.capture_waiting_operation(); + // Poll the outstanding host operation once with a noop waker. + // The embedding-owned driver completes it; re-polling observes + // readiness. + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + match self.poll_waiting_host_op(&mut cx) { + Poll::Ready(Ok(())) => DriveOutcome::Continue, + Poll::Ready(Err(error)) => { + DriveOutcome::Error(self.map_invocation_error(error, waiting_operation)) + } + Poll::Pending => DriveOutcome::Pending, + } + } + Err(error) => { + // `run()` may have registered a new host op before failing; + // retain its operation state for typed error mapping. + let waiting_operation = self.capture_waiting_operation(); + DriveOutcome::Error(self.map_invocation_error(error, waiting_operation)) + } + } + } + + /// Captures the state of the host operation the VM is waiting on, if any. + /// + /// The waiting state must be captured after `run()` (the step may have + /// registered a new host op) and before a poll that may fail and remove + /// the operation from the registry: `map_invocation_error` needs the + /// retained state to recover the typed `OperationStatus::Failed` error + /// once the waiting state has been cleared. + fn capture_waiting_operation(&self) -> Option { + self.instance + .waiting_host_op + .and_then(|op| OperationId::from_raw(op.op_id).ok()) + .and_then(|operation_id| self.host.runtime_operations.get(operation_id).ok()) + } + + /// Maps a low-level VM failure to the typed invocation error, preserving + /// structured runtime errors from `stream::emit` validation and from failed + /// host operations. The waiting operation state is captured by the caller + /// before the poll that may fail and remove it from the registry. + fn map_invocation_error( + &mut self, + error: VmError, + waiting_operation: Option, + ) -> InvocationError { + if let Some(state) = self.instance.invocation.as_mut() + && let Some(runtime_error) = state.pending_error.take() + { + return InvocationError::Capability(runtime_error); + } + if let Some(operation) = waiting_operation + && let OperationStatus::Failed(runtime_error) = operation.status() + { + return InvocationError::Capability(runtime_error); + } + match error { + VmError::OutOfFuel { needed, remaining } => { + InvocationError::OutOfFuel { needed, remaining } + } + VmError::EpochDeadlineReached { current, deadline } => { + InvocationError::DeadlineReached { current, deadline } + } + VmError::HostError(message) => InvocationError::Host { message }, + other => InvocationError::Vm(other), + } + } + + /// Replaces the active invocation phase, returning the previous one so the + /// caller can consume it or drop it (the pending-event drop contract stays + /// with the caller). + fn replace_invocation_phase(&mut self, phase: InvocationPhase) -> InvocationPhase { + std::mem::replace( + &mut self + .instance + .invocation + .as_mut() + .expect("invocation state") + .phase, + phase, + ) + } + + /// Releases the active invocation: cancels outstanding owned operations, + /// drops interpreter frames and stack entries introduced by the + /// invocation, and fuses the stream. + /// + /// Releasing is the invocation boundary for VM-level cancellation: the + /// run-context cancellation token is replaced with a fresh root so the + /// reason consumed by this invocation (or any stale pre-invocation + /// cancellation) cannot leak into a later invocation on the same VM. + /// Outstanding operations were cancelled above; operations registered by + /// a later invocation attach to the fresh token, preserving per-invocation + /// parent cancellation semantics. + fn release_invocation(&mut self) { + let (stack_base, frame_count) = self + .instance + .invocation + .as_ref() + .map(|state| (state.stack_base, state.frame_count)) + .unwrap_or((0, 0)); + self.abort_host_invocation(stack_base, frame_count); + if let Some(state) = self.instance.invocation.as_mut() { + state.phase = InvocationPhase::Fused; + state.emit_yield_pending = false; + state.pending_error = None; + } + self.run_ctx.cancellation = CancellationToken::root(); + } + + /// Implements the script-visible `stream::emit(value)` builtin: validates + /// the per-item bound, places one pending event, and yields control to the + /// invocation poller. `stream::emit` still evaluates to `()` inside RSS. + /// + /// When the poller has delivered the event and the VM resumes, the call + /// site re-executes; the second entry consumes the `emit_yield_pending` + /// marker and returns normally instead of emitting a second event. + pub(crate) fn emit_stream_item(&mut self, value: Value) -> VmResult { + let state = self.instance.invocation.as_mut().ok_or_else(|| { + VmError::HostError("stream::emit requires an active invocation".to_string()) + })?; + if !matches!(state.phase, InvocationPhase::Running) { + return Err(VmError::HostError( + "stream::emit is only valid while the invocation is running".to_string(), + )); + } + if state.emit_yield_pending { + state.emit_yield_pending = false; + return Ok(CallOutcome::Return(CallReturn::none())); + } + let limits = self.run_ctx.runtime_context.event_limits(); + match crate::builtins::runtime::event::EventPayload::try_new(value, limits) { + Ok(payload) => { + state.phase = InvocationPhase::EventPending(payload.into_value()); + state.emit_yield_pending = true; + Ok(CallOutcome::Yield) + } + Err(runtime_error) => { + let message = runtime_error.to_string(); + state.pending_error = Some(runtime_error); + Err(VmError::HostError(message)) + } + } + } +} + +/// Outcome of one low-level drive step. +enum DriveOutcome { + Continue, + Pending, + Error(InvocationError), +} diff --git a/src/vm/mod.rs b/src/vm/mod.rs index 96983ef3..54e8c873 100644 --- a/src/vm/mod.rs +++ b/src/vm/mod.rs @@ -13,6 +13,7 @@ mod fuel; mod host; mod host_runtime; mod instance; +pub mod invocation; pub(crate) mod jit; mod map_iter; pub(crate) mod native; @@ -39,6 +40,7 @@ pub use self::host::{ use self::host::{HostCallExecOutcome, VmHostFunction}; use self::host_runtime::HostRuntime; use self::instance::{ExecutionFrame, FrameContinuation, Instance, QueuedCallable}; +pub use self::invocation::{Invocation, InvocationError, InvocationItem, InvocationPoll}; use self::run_context::{InterruptMode, RunContext}; pub use crate::builtins::runtime::cancellation::CancellationReason; diff --git a/src/vm/run_context.rs b/src/vm/run_context.rs index 235a1c5a..a48495fa 100644 --- a/src/vm/run_context.rs +++ b/src/vm/run_context.rs @@ -1,10 +1,10 @@ //! Run-scoped execution context. //! //! [`RunContext`] owns everything that belongs to one execution of a program: -//! the generic runtime input/event context, fuel and epoch budgets, the -//! interrupt mode, and the epoch counter handle. A fresh logical run starts -//! from a reset context; nothing here survives a reset except the epoch handle -//! identity (which is intentionally process-lifetime). +//! the run-scoped invocation stream configuration (event limits), fuel and +//! epoch budgets, the interrupt mode, and the epoch counter handle. A fresh +//! logical run starts from a reset context; nothing here survives a reset +//! except the epoch handle identity (which is intentionally process-lifetime). //! //! The embedder-facing fuel/epoch APIs live on the VM facade (see //! `crate::vm::fuel` and `crate::vm::epoch`) and delegate here; cancellation @@ -35,11 +35,11 @@ impl InterruptMode { } } -/// Run-scoped input, events, budgets, deadlines, and interruption state. +/// Run-scoped configuration, budgets, deadlines, and interruption state. /// -/// Thread safety: `RunContext` is `!Sync` (event sink and counters are -/// mutable) and not shared; one facade owns one context. Clone semantics: -/// not `Clone` — a clone would duplicate event/input state across runs. +/// Thread safety: `RunContext` is `!Sync` (mutable counters) and not shared; +/// one facade owns one context. Clone semantics: not `Clone` — a clone would +/// duplicate run-scoped state across runs. pub(crate) struct RunContext { pub(crate) runtime_context: RuntimeContext, pub(crate) cancellation: CancellationToken, @@ -58,8 +58,8 @@ pub(crate) struct RunContext { } impl RunContext { - /// Creates a fresh run context with no input, no event sink, and no - /// budgets (interrupts disabled). + /// Creates a fresh run context with default event limits and no budgets + /// (interrupts disabled). pub(crate) fn new() -> Self { let epoch_handle = EpochHandle::default(); let epoch_counter_ptr = epoch_handle.as_ptr() as usize; @@ -78,15 +78,15 @@ impl RunContext { } } - /// Closes run-scoped state for reuse: input and events are cleared and - /// fuel/epoch budgets are dropped (metering disabled, no leftovers). + /// Closes run-scoped state for reuse: fuel/epoch budgets are dropped + /// (metering disabled, no leftovers). The invocation stream event limits + /// are configuration and intentionally survive a reset. pub(crate) fn reset_for_reuse(&mut self) { self.cancellation.cancel(CancellationReason::VmReset); self.cancellation = CancellationToken::root(); self.epoch_rearm_pending = false; self.clear_fuel_internal(); self.clear_epoch_deadline_internal(); - self.runtime_context.reset_for_reuse(); } pub(crate) fn cancel(&self, reason: CancellationReason) -> VmResult<()> { diff --git a/src/vm/tests.rs b/src/vm/tests.rs index f4790291..8777f528 100644 --- a/src/vm/tests.rs +++ b/src/vm/tests.rs @@ -2529,6 +2529,97 @@ fn run_yields_before_ret_in_call_ret_sequence_when_epoch_deadline_is_reached() { assert_eq!(vm.stack(), &[Value::Int(4)]); } +#[test] +fn pre_cancelled_invocation_stays_pending_until_error_delivery() { + // Regression: starting an invocation on an already-cancelled run context + // must not release early (no callable, frame, or host operation has + // started yet). The reason stays pending on the run context until normal + // error delivery consumes the typed error item, which releases exactly + // once. + let compiled = crate::compile_source( + r#" + pub fn run() -> int { + 42; + } + "#, + ) + .expect("invocation source should compile"); + let mut vm = Vm::new(compiled.program); + assert_eq!(vm.run().expect("root frame should halt"), VmStatus::Halted); + + vm.run_ctx + .cancel(CancellationReason::Requested) + .expect("pre-cancellation should be accepted"); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + let _invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + // Dropping the handle abandons the invocation but keeps it active on + // the VM; the pre-cancelled transition must not have released it. + } + assert!( + vm.run_ctx.cancellation.reason().is_some(), + "the pre-cancellation must remain pending until the error item is consumed" + ); +} + +#[test] +fn pre_cancelled_invocation_delivers_one_typed_error_then_fused_end() { + // Functional contract of the pre-cancelled path: exactly one typed + // Cancelled item, a fused end, and the cancellation consumed at the + // invocation boundary (a later invocation runs normally). + let compiled = crate::compile_source( + r#" + pub fn run() -> int { + 42; + } + "#, + ) + .expect("invocation source should compile"); + let mut vm = Vm::new(compiled.program); + assert_eq!(vm.run().expect("root frame should halt"), VmStatus::Halted); + + vm.run_ctx + .cancel(CancellationReason::Requested) + .expect("pre-cancellation should be accepted"); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + let mut invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + CancellationReason::Requested, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + } + + // A new invocation on the same VM must run to completion instead of + // being cancelled on arrival. + let mut second = vm + .start_invocation(callable, vec![]) + .expect("a new invocation may start after fusion"); + match second.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) => {} + other => panic!("the second invocation must complete normally, got {other:?}"), + } + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + #[test] fn call_ret_fusion_pattern_requires_immediate_ret() { let [call_lo, call_hi] = BuiltinFunction::Len.call_index().to_le_bytes(); diff --git a/tests/compiler/compiler_rustscript_tests.rs b/tests/compiler/compiler_rustscript_tests.rs index b7fa4afc..8641db26 100644 --- a/tests/compiler/compiler_rustscript_tests.rs +++ b/tests/compiler/compiler_rustscript_tests.rs @@ -3809,3 +3809,25 @@ fn rustscript_generic_schema_errors_are_reported() { run_source_error_cases(&cases); } + +#[test] +fn rustscript_strict_stream_emit_accepts_any_payload() { + // In strict RustScript, `stream::emit` is the one host function whose + // `any` payload is accepted at compile time; the per-item event bound is + // validated at runtime by the invocation stream. The exemption is tied to + // the authoritative runtime builtin identity (see the compiler unit test + // `stream_emit_any_payload_exemption_requires_authoritative_builtin_identity`), + // so a same-name function registered through another catalog cannot + // inherit it. + compile_source( + r#" + use stream; + pub fn run() -> int { + stream::emit({"a": 1, "b": 2}); + stream::emit("text"); + 42; + } + "#, + ) + .expect("strict stream::emit with any payloads must compile"); +} diff --git a/tests/invocation_stream_tests.rs b/tests/invocation_stream_tests.rs new file mode 100644 index 00000000..f5ea5a29 --- /dev/null +++ b/tests/invocation_stream_tests.rs @@ -0,0 +1,879 @@ +#![cfg(feature = "runtime")] + +//! Invocation item stream contract tests. +//! +//! An invocation behaves like `Stream>`: +//! zero or more `Event` items, then exactly one `Complete` item or one typed error, +//! then a fused end of stream. Input enters through ordinary callable arguments and +//! polling drives execution (backpressure). + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use vm::{ + CancellationReason, HostFunctionRegistry, InvocationError, InvocationItem, InvocationPoll, + Value, Vm, VmError, compile_source, +}; + +/// Compiles a source, binds the default runtime host registry, and completes the +/// root frame so exported callables can be started. +fn compiled_vm(source: &str) -> Vm { + let program = compile_source(source) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + HostFunctionRegistry::new() + .bind_vm_cached(&mut vm) + .expect("default runtime host registry should bind"); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + vm +} + +/// Drives one exported `run` callable to the end of its invocation stream. +fn collect_items(vm: &mut Vm, args: Vec) -> Vec> { + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, args) + .expect("invocation should start"); + let mut items = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + assert!( + Instant::now() < deadline, + "invocation drive loop must terminate" + ); + match invocation + .poll_next() + .expect("invocation poll should not fail") + { + InvocationPoll::Ready(Some(item)) => items.push(item), + InvocationPoll::Ready(None) => break, + InvocationPoll::Pending => std::thread::sleep(Duration::from_millis(1)), + } + } + items +} + +#[test] +fn invocation_input_arrives_as_ordinary_callable_arguments() { + let mut vm = compiled_vm( + r#" + pub fn run(input: map) -> map { + input; + } + "#, + ); + let input = Value::map(vec![(Value::string("kind"), Value::string("message"))]); + let items = collect_items(&mut vm, vec![input.clone()]); + assert_eq!(items.len(), 1, "expected exactly one stream item"); + assert!( + matches!(&items[0], Ok(InvocationItem::Complete(value)) if *value == input), + "the exact structured argument must be the callable input, got {:?}", + items + ); +} + +#[test] +fn invocation_without_events_yields_complete_then_fused_end() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + 42; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + assert!( + matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + ), + "the stream must stay fused after Complete" + ); + + // Once the first invocation has fused, a new invocation may start on the + // same VM. + let mut second = vm + .start_invocation(callable, vec![]) + .expect("a new invocation may start after fusion"); + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) + )); + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_starting_a_second_invocation_while_one_is_active_is_rejected() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + 42; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + // The handle is dropped without polling; the invocation stays active + // on the VM until it fuses. + let _invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("first invocation should start"); + } + // Dropping the handle keeps the invocation active on the VM; starting a + // second invocation must be rejected while the first one has not fused. + let rejected = vm + .start_invocation(callable, vec![]) + .expect_err("a second active invocation must be rejected"); + assert!( + matches!(rejected, VmError::InvalidFrameState(_)), + "unexpected rejection error: {rejected:?}" + ); +} + +#[test] +fn invocation_failures_are_typed_items_without_stack_or_string_inspection() { + let mut vm = compiled_vm( + r#" + pub fn run(input: int) -> int { + 100 / input; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![Value::Int(0)]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Vm(VmError::DivisionByZero)))) => {} + other => panic!("expected a typed division-by-zero item, got {other:?}"), + } + assert!( + matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + ), + "the stream must fuse after the error item" + ); +} + +/// Records one script-visible progress note per call. +struct ProgressNote(Arc>>); + +impl vm::HostArgsFunction for ProgressNote { + fn call(&mut self, args: &[Value]) -> vm::VmResult { + if let Some(value) = args.first() { + self.0 + .lock() + .expect("progress note lock should not be poisoned") + .push(value.clone()); + } + Ok(vm::CallOutcome::Return(vm::CallReturn::one( + args.first().cloned().unwrap_or(Value::Null), + ))) + } +} + +#[test] +fn invocation_emits_events_then_complete_in_order() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> string { + stream::emit("first"); + stream::emit("second"); + "done"; + } + "#, + ); + let items = collect_items(&mut vm, vec![]); + assert_eq!( + items.len(), + 3, + "expected event, event, complete; got {items:?}" + ); + assert!( + matches!(&items[0], Ok(InvocationItem::Event(value)) if *value == Value::string("first")) + ); + assert!( + matches!(&items[1], Ok(InvocationItem::Event(value)) if *value == Value::string("second")) + ); + assert!( + matches!(&items[2], Ok(InvocationItem::Complete(value)) if *value == Value::string("done")) + ); +} + +#[test] +fn invocation_event_values_never_replace_the_callable_return_value() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> int { + stream::emit("payload"); + 42; + } + "#, + ); + let items = collect_items(&mut vm, vec![]); + assert_eq!( + items.len(), + 2, + "expected event then complete; got {items:?}" + ); + assert!( + matches!(&items[0], Ok(InvocationItem::Event(value)) if *value == Value::string("payload")) + ); + assert!(matches!( + &items[1], + Ok(InvocationItem::Complete(Value::Int(42))) + )); +} + +#[test] +fn invocation_polling_pauses_execution_and_exposes_one_event_at_a_time() { + let program = compile_source( + r#" + use stream; + fn note_progress(value: string) -> string; + pub fn run() -> string { + stream::emit("a"); + note_progress("after-a"); + stream::emit("b"); + note_progress("after-b"); + "done"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + let notes = Arc::new(Mutex::new(Vec::::new())); + vm.bind_args_function("note_progress", Box::new(ProgressNote(Arc::clone(¬es)))); + // `stream::emit` binds lazily through the default host fallback; the custom + // host binding is not part of the registry plan. + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + // First poll: the script paused at the first emit; nothing after it ran. + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("a") + )); + assert!( + notes.lock().expect("notes lock").is_empty(), + "execution must not advance while polling is paused" + ); + + // Second poll: resume past emit(a), run note_progress("after-a"), pause at + // emit(b). Exactly one progress note may exist. + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("b") + )); + assert_eq!( + notes.lock().expect("notes lock").len(), + 1, + "exactly one progress note between polls" + ); + + // Third poll: resume past emit(b), run note_progress("after-b"), complete. + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(value)))) if value == Value::string("done") + )); + assert_eq!(notes.lock().expect("notes lock").len(), 2); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_cancellation_produces_one_typed_error_item_then_fused_end() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> string { + stream::emit("before"); + while true { + 1; + } + "unreachable"; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("before") + )); + + invocation + .cancel(CancellationReason::Requested) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + CancellationReason::Requested, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_fuel_exhaustion_produces_one_typed_error_item() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + while true { + 1; + } + 42; + } + "#, + ); + vm.set_fuel(8); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::OutOfFuel { + needed: _, + remaining: 0, + }))) => {} + other => panic!("expected a typed out-of-fuel item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_deadline_expiry_produces_one_typed_error_item() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + 42; + } + "#, + ); + vm.set_epoch_deadline(0) + .expect("epoch deadline should be configured"); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::DeadlineReached { + current: 0, + deadline: 0, + }))) => {} + other => panic!("expected a typed deadline item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_host_failure_produces_one_typed_error_item() { + let program = compile_source( + r#" + fn fail_host() -> int; + pub fn run() -> int { + fail_host(); + 42; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("fail_host", Box::new(FailingHost)); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Host { message }))) => { + assert_eq!(message, "boom"); + } + other => panic!("expected a typed host failure item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_event_bound_violations_are_typed_capability_errors() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run(input: string) -> int { + stream::emit(input); + 42; + } + "#, + ); + let oversized = "x".repeat(70 * 1024); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![Value::string(oversized)]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Capability(error)))) => { + assert_eq!(error.code(), vm::RuntimeErrorCode::EventPayloadTooLarge); + } + other => panic!("expected a typed capability error item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +/// Fails every host call with a plain embedding error. +struct FailingHost; + +impl vm::HostStackFunction for FailingHost { + fn call(&mut self, _vm: &mut Vm, _args: &[Value]) -> vm::VmResult { + Err(vm::VmError::HostError("boom".to_string())) + } +} + +#[cfg(feature = "async")] +#[path = "support/async_test_bridge.rs"] +mod async_test_bridge; + +/// Waits asynchronously through the embedding-owned host bridge. +#[cfg(feature = "async")] +struct AsyncWaitHost; + +#[cfg(feature = "async")] +impl vm::HostStackFunction for AsyncWaitHost { + fn call(&mut self, vm: &mut Vm, _args: &[Value]) -> vm::VmResult { + vm.submit_host_future(Box::pin(async move { + tokio::time::sleep(Duration::from_millis(20)).await; + Ok(vm::HostFutureOutput::returning(vm::CallReturn::one( + Value::Int(7), + ))) + })) + } +} + +#[cfg(feature = "async")] +#[test] +fn invocation_waiting_host_operation_returns_pending_and_preserves_item_order() { + let program = compile_source( + r#" + use stream; + fn wait_host() -> int; + pub fn run() -> string { + stream::emit("a"); + wait_host(); + stream::emit("b"); + "done"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("wait_host", Box::new(AsyncWaitHost)); + async_test_bridge::install(&mut vm); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("a") + )); + + // The outstanding host operation maps to Pending; drive it and poll again. + let deadline = Instant::now() + Duration::from_secs(10); + let mut polled_pending = false; + let next = loop { + assert!( + Instant::now() < deadline, + "waiting invocation must resume through the host driver" + ); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Pending => { + polled_pending = true; + std::thread::sleep(Duration::from_millis(1)); + } + ready => break ready, + } + }; + assert!( + polled_pending, + "the waiting host op must surface as Pending" + ); + assert!(matches!( + next, + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("b") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(value)))) if value == Value::string("done") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_cancellation_is_consumed_at_the_invocation_boundary() { + // Regression: after a cancelled invocation emits its typed error and + // fuses, the VM-level cancellation reason must not leak into a later + // invocation started on the same VM. + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> string { + stream::emit("before"); + while true { + 1; + } + "unreachable"; + } + pub fn plain() -> int { + 42; + } + "#, + ); + let cancellable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(cancellable, vec![]) + .expect("invocation should start"); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("before") + )); + + invocation + .cancel(CancellationReason::Requested) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + CancellationReason::Requested, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + + // A fresh invocation on the same VM must not inherit the old reason: it + // runs to completion instead of being cancelled on arrival. + let plain = vm + .resolve_exported_callable("plain") + .expect("exported plain callable should resolve"); + let mut second = vm + .start_invocation(plain, vec![]) + .expect("a new invocation may start after fusion"); + match second.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) => {} + other => panic!("the second invocation must complete normally, got {other:?}"), + } + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_cancel_during_event_pending_discards_the_pending_event() { + // Cancellation is authoritative: a pending event that was placed but not + // yet delivered must be discarded (through the drop-contract path) and + // the stream must produce exactly one Cancelled item, then a fused end. + let program = compile_source( + r#" + use stream; + pub fn run() -> string { + stream::emit({"a": 1, "b": 2}); + while true { + 1; + } + "unreachable"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.set_drop_contract_events_enabled(true); + HostFunctionRegistry::new() + .bind_vm_cached(&mut vm) + .expect("default runtime host registry should bind"); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let drops_before_cancel = vm.drop_contract_event_count(); + // `start_callable` runs to the first `stream::emit` yield, so the + // invocation is already in EventPending with the map payload. + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + invocation + .cancel(CancellationReason::Requested) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + CancellationReason::Requested, + )))) => {} + other => panic!("cancellation must supersede the pending event, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + + // The discarded event payload (map plus its two key/value pairs) must be + // dropped through the VM drop-contract path, not leaked. + assert!( + vm.drop_contract_event_count() >= drops_before_cancel + 5, + "the discarded pending event payload must be dropped through the drop contract path" + ); +} + +#[test] +fn invocation_cancel_during_complete_pending_discards_the_pending_complete() { + // Cancellation is authoritative over a not-yet-delivered Complete item: + // the callable result is discarded and the stream produces exactly one + // Cancelled item, then a fused end. + let mut vm = compiled_vm( + r#" + pub fn run() -> map { + {"a": 1, "b": 2}; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + // The callable completes during `start_callable`, so the invocation is + // already in CompletePending with the return map. + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + invocation + .cancel(CancellationReason::Deadline) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + CancellationReason::Deadline, + )))) => {} + other => panic!("cancellation must supersede the pending complete, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +/// Fails asynchronously on the first poll of its submitted host operation. +#[cfg(feature = "async")] +struct AsyncFailHost; + +#[cfg(feature = "async")] +impl vm::HostStackFunction for AsyncFailHost { + fn call(&mut self, vm: &mut Vm, _args: &[Value]) -> vm::VmResult { + vm.submit_host_future(Box::pin(async move { + Err(vm::VmError::HostError("bridge future failed".to_string())) + })) + } +} + +#[cfg(feature = "async")] +#[test] +fn invocation_host_op_first_poll_failure_keeps_typed_capability_error() { + // Regression: the waiting operation id must be captured after `run()` + // registers the host op. If the first poll fails and clears the waiting + // state, `map_invocation_error` must still recover the structured + // `OperationStatus::Failed` error instead of flattening it to a string. + let program = compile_source( + r#" + fn fail_host() -> int; + pub fn run() -> int { + fail_host(); + 42; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("fail_host", Box::new(AsyncFailHost)); + async_test_bridge::install(&mut vm); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Capability(error)))) => { + assert_eq!(error.code(), vm::RuntimeErrorCode::OperationFailed); + assert_eq!(error.operation(), "runtime::host_bridge"); + assert!( + error.value().is_some(), + "the typed failure must carry the operation id" + ); + } + other => panic!( + "expected a typed capability error for the first-poll host op failure, got {other:?}" + ), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[cfg(feature = "async")] +#[test] +fn invocation_cancellation_while_waiting_produces_one_typed_error_item() { + let program = compile_source( + r#" + use stream; + fn wait_host() -> int; + pub fn run() -> string { + stream::emit("a"); + wait_host(); + "unreachable"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("wait_host", Box::new(AsyncWaitHost)); + async_test_bridge::install(&mut vm); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("a") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Pending + )); + + invocation + .cancel(CancellationReason::Deadline) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + CancellationReason::Deadline, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} diff --git a/tests/runtime_context_tests.rs b/tests/runtime_context_tests.rs index 31dfb06b..2c2a05f3 100644 --- a/tests/runtime_context_tests.rs +++ b/tests/runtime_context_tests.rs @@ -1,5 +1,5 @@ mod vm { - pub use ::vm::{Value, VmError, VmResult}; + pub use ::vm::Value; } #[allow(dead_code)] @@ -30,55 +30,37 @@ use resource::{CloseStatus, ResourceArena, ResourceHandle, ResourceTypeId}; use vm::Value; #[test] -fn runtime_input_is_run_scoped_and_missing_input_is_typed() { - let mut context = RuntimeContext::default(); - let missing = context.input().expect_err("unset input should be rejected"); - assert_eq!(missing.code(), RuntimeErrorCode::InputUnavailable); +fn per_item_event_limits_are_run_scoped_configuration() { + let context = RuntimeContext::default(); + assert_eq!(context.event_limits(), EventLimits::default()); + assert_eq!(context.config().event_limits(), EventLimits::default()); - let input = Value::map(vec![(Value::string("kind"), Value::string("message"))]); - context - .set_input(input.clone()) - .expect("input should be accepted"); - assert_eq!(context.input().expect("input should be available"), input); -} - -#[test] -fn runtime_emit_validates_payload_before_calling_the_sink() { - let mut context = RuntimeContext::with_config(RuntimeContextConfig::new( + let configured = RuntimeContext::with_config(RuntimeContextConfig::new( EventLimits::new(8, 4).expect("test limits should be valid"), )) .expect("context should be constructible"); - let seen = Arc::new(Mutex::new(Vec::::new())); - let seen_by_sink = Arc::clone(&seen); - context - .set_event_sink(move |payload: EventPayload| { - seen_by_sink - .lock() - .expect("event sink lock should not be poisoned") - .push(payload.into_value()); - Ok(()) - }) - .expect("event sink should be installed"); + assert_eq!(configured.event_limits().max_payload_bytes(), 8); + assert_eq!(configured.event_limits().max_depth(), 4); +} + +#[test] +fn event_payload_validates_the_per_item_bound_before_placement() { + let limits = EventLimits::new(8, 4).expect("test limits should be valid"); - context - .emit(Value::string("ok")) - .expect("bounded event should reach the sink"); - assert_eq!(seen.lock().expect("event sink lock").len(), 1); + let payload = + EventPayload::try_new(Value::string("ok"), limits).expect("bounded event should validate"); + assert_eq!(payload.into_value(), Value::string("ok")); - let too_large = context - .emit(Value::string("payload-too-large")) + let too_large = EventPayload::try_new(Value::string("payload-too-large"), limits) .expect_err("oversized event should be rejected"); assert_eq!(too_large.code(), RuntimeErrorCode::EventPayloadTooLarge); - assert_eq!(seen.lock().expect("event sink lock").len(), 1); -} -#[test] -fn runtime_emit_reports_missing_sink_without_dropping_the_value_contract() { - let mut context = RuntimeContext::default(); - let error = context - .emit(Value::Bool(true)) - .expect_err("emit without a sink should fail"); - assert_eq!(error.code(), RuntimeErrorCode::EventSinkUnavailable); + let too_deep = EventPayload::try_new( + Value::array(vec![Value::array(vec![Value::array(vec![Value::Int(1)])])]), + EventLimits::new(1024, 2).expect("depth test limits should be valid"), + ) + .expect_err("too-deep event should be rejected"); + assert_eq!(too_deep.code(), RuntimeErrorCode::EventDepthExceeded); } #[test] diff --git a/tests/runtime_host_tests.rs b/tests/runtime_host_tests.rs index b81630f9..71772232 100644 --- a/tests/runtime_host_tests.rs +++ b/tests/runtime_host_tests.rs @@ -1,79 +1,115 @@ #![cfg(feature = "runtime")] -use std::sync::{Arc, Mutex}; - #[cfg(feature = "sqlite")] use vm::SqliteHostExt; use vm::{ - EventPayload, EventSink, HostFunctionRegistry, RuntimeResult, Value, Vm, VmStatus, - compile_source, + HostFunctionRegistry, InvocationError, InvocationItem, InvocationPoll, Value, Vm, VmError, + VmStatus, compile_source, }; -struct RecordingEventSink(Arc>>); - -impl EventSink for RecordingEventSink { - fn emit(&mut self, payload: EventPayload) -> RuntimeResult<()> { - self.0 - .lock() - .expect("event capture lock should not be poisoned") - .push(payload.into_value()); - Ok(()) - } -} - -#[test] -fn runtime_input_host_reads_embedding_run_value() { - let program = compile_source( - r#" - use runtime; - runtime::input(); - "#, - ) - .expect("runtime input source should compile") - .program; +/// Compiles a source, binds the default runtime host registry, and completes +/// the root frame so exported callables can be started. +fn prepared_vm(source: &str) -> Vm { + let program = compile_source(source) + .expect("runtime host source should compile") + .program; let mut vm = Vm::new(program); - vm.set_runtime_input(Value::string("run-input")) - .expect("runtime input should be configurable"); HostFunctionRegistry::new() .bind_vm_cached(&mut vm) .expect("default runtime host registry should bind"); + assert_eq!(vm.run().expect("root frame should halt"), VmStatus::Halted); + vm +} - assert_eq!( - vm.run().expect("runtime input should execute"), - VmStatus::Halted +#[test] +fn invocation_input_arrives_through_exported_callable_arguments() { + let mut vm = prepared_vm( + r#" + pub fn run(input: string) -> string { + input; + } + "#, ); - assert_eq!(vm.stack().last(), Some(&Value::string("run-input"))); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![Value::string("run-input")]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(value)))) => { + assert_eq!(value, Value::string("run-input")); + } + other => panic!("expected the callable input as the Complete value, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); } #[test] -fn runtime_input_host_reports_missing_embedding_value() { - let program = compile_source( +fn stream_emit_delivers_events_through_the_invocation_stream() { + let mut vm = prepared_vm( r#" - use runtime; - runtime::input(); + use stream; + pub fn run() -> string { + stream::emit("event-one"); + stream::emit("event-two"); + "done"; + } "#, - ) - .expect("runtime input source should compile") - .program; - let mut vm = Vm::new(program); - HostFunctionRegistry::new() - .bind_vm_cached(&mut vm) - .expect("default runtime host registry should bind"); + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); - let error = vm.run().expect_err("missing runtime input should fail"); - assert!(error.to_string().contains("input_unavailable")); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("event-one") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("event-two") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(value)))) if value == Value::string("done") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); } #[test] -fn public_runtime_event_contract_is_implementable_and_configurable() { - let program = compile_source("0;") - .expect("minimal runtime host program should compile") - .program; - let events = Arc::new(Mutex::new(Vec::new())); - let mut vm = Vm::new(program); - vm.set_runtime_event_sink(RecordingEventSink(Arc::clone(&events))) - .expect("public EventSink implementation should be configurable"); - vm.clear_runtime_event_sink(); +fn invocation_errors_are_typed_without_string_parsing() { + let mut vm = prepared_vm( + r#" + pub fn run(input: int) -> int { + 1 / input; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![Value::Int(0)]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Err(InvocationError::Vm(VmError::DivisionByZero)))) + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); } #[cfg(feature = "sqlite")] diff --git a/tests/vm/ownership_tests.rs b/tests/vm/ownership_tests.rs index ffad0c5c..95c2eefd 100644 --- a/tests/vm/ownership_tests.rs +++ b/tests/vm/ownership_tests.rs @@ -2,7 +2,7 @@ //! //! These tests pin the ownership contract through the public embedding API: //! - one immutable program can create multiple isolated instances; -//! - run input/events/budgets never leak between runs; +//! - invocation input/events/budgets never leak between runs; //! - backend caches may be shared without sharing stacks/resources; //! - reset closes run-scoped state and retains only documented reusable state. @@ -10,9 +10,9 @@ mod common; use common::*; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; -use vm::{HostFunctionRegistry, Value, VmStatus}; +use vm::{HostFunctionRegistry, InvocationError, InvocationItem, InvocationPoll, Value, VmStatus}; fn non_yielding_returns_zero(_: &[Value]) -> Result { Ok(vm::CallOutcome::Return(vm::CallReturn::one(Value::Int(0)))) @@ -30,6 +30,31 @@ fn non_yielding_returns_forty_two(_: &[Value]) -> Result, +) -> Vec> { + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, args) + .expect("invocation should start"); + let mut items = Vec::new(); + loop { + match invocation + .poll_next() + .expect("invocation poll should not fail") + { + InvocationPoll::Ready(Some(item)) => items.push(item), + InvocationPoll::Ready(None) => break, + InvocationPoll::Pending => std::thread::sleep(std::time::Duration::from_millis(1)), + } + } + items +} + struct PendingOneHost; impl vm::HostArgsFunction for PendingOneHost { @@ -38,16 +63,16 @@ impl vm::HostArgsFunction for PendingOneHost { } } -/// One immutable program produces independent instances: each run keeps its own -/// stack, locals, and input, and no instance observes another's execution. +/// One immutable program produces independent instances: each invocation keeps +/// its own stream items, and no instance observes another's execution. #[test] fn one_immutable_program_creates_multiple_isolated_instances() { let program = Arc::new( compile_source( r#" - use runtime; - let value: string = runtime::input_json(); - value; + pub fn run(input: string) -> string { + input; + } "#, ) .expect("source should compile") @@ -62,60 +87,65 @@ fn one_immutable_program_creates_multiple_isolated_instances() { HostFunctionRegistry::new() .bind_vm_cached(&mut second) .expect("runtime hosts should bind"); - - first - .set_runtime_input(Value::string("first")) - .expect("input should be accepted"); - second - .set_runtime_input(Value::string("second")) - .expect("input should be accepted"); - - assert_eq!(first.run().expect("first should run"), VmStatus::Halted); assert_eq!( - first.stack().last(), - Some(&Value::string("\"first\"")), - "first instance must observe its own input" + first.run().expect("first root should halt"), + VmStatus::Halted ); - assert_eq!(second.run().expect("second should run"), VmStatus::Halted); assert_eq!( - second.stack().last(), - Some(&Value::string("\"second\"")), - "second instance must observe its own input" + second.run().expect("second root should halt"), + VmStatus::Halted + ); + + let first_items = collect_invocation_items(&mut first, vec![Value::string("first")]); + let second_items = collect_invocation_items(&mut second, vec![Value::string("second")]); + + assert_eq!(first_items.len(), 1, "first invocation must complete once"); + assert!( + matches!(&first_items[0], Ok(InvocationItem::Complete(value)) if *value == Value::string("first")), + "first instance must observe its own input, got {first_items:?}" ); assert_eq!( - first.stack().last(), - Some(&Value::string("\"first\"")), - "second's run must not overwrite first's stack" + second_items.len(), + 1, + "second invocation must complete once" + ); + assert!( + matches!(&second_items[0], Ok(InvocationItem::Complete(value)) if *value == Value::string("second")), + "second instance must observe its own input, got {second_items:?}" ); - // Re-running one instance after reset must not disturb the other. + // Re-running one instance after reset must not disturb the other. Reset + // rewinds the root frame, so the root must halt again before callables can + // be started. first.reset_for_reuse(); - first - .set_runtime_input(Value::string("first-again")) - .expect("input should be accepted"); - assert_eq!(first.run().expect("first should rerun"), VmStatus::Halted); assert_eq!( - second.stack().last(), - Some(&Value::string("\"second\"")), - "first's rerun must not disturb second's stack" + first.run().expect("first root should halt again"), + VmStatus::Halted + ); + let first_again = collect_invocation_items(&mut first, vec![Value::string("first-again")]); + assert!( + matches!(&first_again[0], Ok(InvocationItem::Complete(value)) if *value == Value::string("first-again")), + "first rerun must observe its own fresh input, got {first_again:?}" ); assert_eq!( - first.stack().last(), - Some(&Value::string("\"first-again\"")) + second_items.len(), + 1, + "first's rerun must not disturb second" ); } -/// Run input and events are run-scoped: a reset closes them, and a later run -/// starts with a clean context. +/// Invocation events and results are run-scoped: a reset closes them, and a +/// later run starts with a clean stream. #[test] fn run_input_and_events_do_not_leak_between_runs() { let program = Arc::new( compile_source( r#" - use runtime; - let value: string = runtime::input_json(); - runtime::emit_json(value); - value; + use stream; + pub fn run(input: string) -> string { + stream::emit(input); + input; + } "#, ) .expect("source should compile") @@ -125,63 +155,33 @@ fn run_input_and_events_do_not_leak_between_runs() { HostFunctionRegistry::new() .bind_vm_cached(&mut vm) .expect("runtime hosts should bind"); + assert_eq!(vm.run().expect("root should halt"), VmStatus::Halted); - let seen = Arc::new(Mutex::new(Vec::::new())); - let sink_seen = Arc::clone(&seen); - vm.set_runtime_value_event_sink(move |value: Value| { - sink_seen.lock().expect("sink lock").push(value); - Ok(()) - }) - .expect("event sink should install"); - - vm.set_runtime_input(Value::string("run-one")) - .expect("input should be accepted"); - assert_eq!(vm.run().expect("first run should halt"), VmStatus::Halted); - assert_eq!(vm.stack().last(), Some(&Value::string("\"run-one\""))); - assert_eq!( - seen.lock().expect("sink lock").len(), - 1, - "first run must emit exactly one event" + let items = collect_invocation_items(&mut vm, vec![Value::string("run-one")]); + assert_eq!(items.len(), 2, "first run must emit one event and complete"); + assert!( + matches!(&items[0], Ok(InvocationItem::Event(value)) if *value == Value::string("run-one")) ); - assert_eq!( - seen.lock().expect("sink lock")[0], - Value::string("\"run-one\"") + assert!( + matches!(&items[1], Ok(InvocationItem::Complete(value)) if *value == Value::string("run-one")) ); - // A reset closes the run-scoped input: the next run must not see the - // previous run's input. + // A reset closes the run-scoped invocation state: the next run starts with + // a fresh stream and neither the old input nor the old events leak. vm.reset_for_reuse(); - let missing = vm.run().expect_err("reset must close run input"); + assert_eq!(vm.run().expect("root should halt again"), VmStatus::Halted); + let items_after_reset = collect_invocation_items(&mut vm, vec![Value::string("run-two")]); + assert_eq!( + items_after_reset.len(), + 2, + "reset must not leak prior events into the next run" + ); assert!( - missing.to_string().contains("input_unavailable"), - "unexpected error after reset: {missing:?}" + matches!(&items_after_reset[0], Ok(InvocationItem::Event(value)) if *value == Value::string("run-two")) ); - - // A fresh run (new instance from the same program) with fresh input sees - // neither the old input nor the old event stream. - let mut fresh = Vm::new_shared(Arc::clone(&program)); - HostFunctionRegistry::new() - .bind_vm_cached(&mut fresh) - .expect("runtime hosts should bind"); - let fresh_seen = Arc::new(Mutex::new(Vec::::new())); - let fresh_sink_seen = Arc::clone(&fresh_seen); - fresh - .set_runtime_value_event_sink(move |value: Value| { - fresh_sink_seen.lock().expect("sink lock").push(value); - Ok(()) - }) - .expect("event sink should install"); - fresh - .set_runtime_input(Value::string("run-two")) - .expect("input should be accepted"); - assert_eq!( - fresh.run().expect("fresh run should halt"), - VmStatus::Halted + assert!( + matches!(&items_after_reset[1], Ok(InvocationItem::Complete(value)) if *value == Value::string("run-two")) ); - assert_eq!(fresh.stack().last(), Some(&Value::string("\"run-two\""))); - let events = fresh_seen.lock().expect("sink lock"); - assert_eq!(events.len(), 1, "fresh run must emit exactly one event"); - assert_eq!(events[0], Value::string("\"run-two\"")); } /// Fuel budgets are run-scoped: a reset clears the budget, and a new run /// starts from its configured amount rather than inheriting leftovers. @@ -353,33 +353,55 @@ fn reset_closes_waiting_state_before_the_next_run() { #[test] #[ignore = "pre-existing JIT stale-trace replay after reset; tracked separately"] fn reset_after_host_error_reruns_cleanly_on_the_same_instance() { + use std::sync::OnceLock; + use std::sync::atomic::{AtomicBool, Ordering}; + + static FAIL_FIRST: OnceLock = OnceLock::new(); + let fail_first = FAIL_FIRST.get_or_init(|| AtomicBool::new(true)); + fail_first.store(true, Ordering::SeqCst); + + fn flaky_action(_: &[Value]) -> Result { + if FAIL_FIRST + .get_or_init(|| AtomicBool::new(true)) + .swap(false, Ordering::SeqCst) + { + Err(vm::VmError::HostError("first call fails".to_string())) + } else { + Ok(vm::CallOutcome::Return(vm::CallReturn::one(Value::Int(42)))) + } + } + let program = compile_source( r#" - use runtime; - let value: string = runtime::input_json(); - runtime::emit_json(value); - value; + fn action() -> int; + pub fn run() -> int { + action(); + } "#, ) .expect("source should compile") .program; - let mut vm = Vm::new(program); - HostFunctionRegistry::new() - .bind_vm_cached(&mut vm) - .expect("runtime hosts should bind"); - vm.set_runtime_value_event_sink(|_| Ok(())) - .expect("event sink should install"); - - vm.set_runtime_input(Value::string("run-one")) - .expect("input should be accepted"); - assert_eq!(vm.run().expect("first run should halt"), VmStatus::Halted); + let mut vm = vm::Vm::new(program); + vm.bind_static_non_yielding_args_function("action", flaky_action); + assert_eq!(vm.run().expect("root should halt"), VmStatus::Halted); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("first invocation should start"); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Err(InvocationError::Host { .. }))) + )); + } vm.reset_for_reuse(); - let missing = vm.run().expect_err("reset must close run input"); - assert!(missing.to_string().contains("input_unavailable")); - - vm.set_runtime_input(Value::string("run-two")) - .expect("input should be accepted"); - assert_eq!(vm.run().expect("rerun should halt"), VmStatus::Halted); - assert_eq!(vm.stack().last(), Some(&Value::string("\"run-two\""))); + let items = collect_invocation_items(&mut vm, vec![]); + assert!( + matches!(&items[0], Ok(InvocationItem::Complete(Value::Int(42)))), + "the rerun must execute cleanly after reset, got {items:?}" + ); }