feat(http): add callable streaming clients

Cargo.toml

@@ -27,8 +27,18 @@ name = "vm"
 [features]
 default = ["runtime", "cli", "cranelift-jit"]
 runtime = []
-async = ["runtime", "dep:reqwest", "dep:url", "dep:tokio", "dep:futures-util"]
-http-client = ["async"]
+async = ["runtime", "dep:tokio", "dep:futures-util"]
+http-client = [
+    "async",
+    "dep:http-body-util",
+    "dep:hyper",
+    "dep:hyper-util",
+    "dep:rustls",
+    "dep:tokio-rustls",
+    "dep:tokio-tungstenite",
+    "dep:url",
+    "dep:webpki-roots",
+]
 sqlite = ["runtime", "dep:rusqlite"]
 edge-abi = [
     "dep:edge_abi",
@@ -63,11 +73,17 @@ cranelift-jit = { version = "0.129.1", optional = true }
 cranelift-module = { version = "0.129.1", optional = true }
 cranelift-native = { version = "0.129.1", optional = true }
 pd-host-function = { path = "./pd-host-function", version = "0.1.0" }
-reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"], optional = true }
+http-body-util = { version = "0.1", optional = true }
+hyper = { version = "1", default-features = false, features = ["client", "http1"], optional = true }
+hyper-util = { version = "0.1", default-features = false, features = ["tokio"], optional = true }
+rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"], optional = true }
+tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"], optional = true }
+tokio-tungstenite = { version = "0.30", default-features = false, features = ["handshake"], optional = true }
+webpki-roots = { version = "1", optional = true }
 rusqlite = { version = "0.32", default-features = false, features = ["bundled", "hooks", "limits"], optional = true }
 url = { version = "2", optional = true }
 futures-util = { version = "0.3", optional = true }
-tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "sync", "fs", "io-util", "process"], optional = true }
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "time", "sync", "fs", "io-util", "process"], optional = true }
 edge_abi = { package = "pd-edge-abi", version = "0.1.1", default-features = false, optional = true }
 futures-channel = "0.3"
 paste = "1"
@@ -86,6 +102,7 @@ libc = "0.2"
 
 [dev-dependencies]
 futures-util = "0.3"
+rcgen = "0.13"
 syn = { version = "2", features = ["full"] }
 tokio = { version = "1", features = ["macros", "rt", "time", "sync"] }
 
@@ -99,6 +116,21 @@ name = "http_host_tests"
 path = "tests/vm/http_host_tests.rs"
 required-features = ["runtime", "http-client"]
 
+[[test]]
+name = "http_sse_tests"
+path = "tests/vm/http_sse_tests.rs"
+required-features = ["runtime", "http-client"]
+
+[[test]]
+name = "host_stream_callback_tests"
+path = "tests/vm/host_stream_callback_tests.rs"
+required-features = ["runtime"]
+
+[[test]]
+name = "http_websocket_tests"
+path = "tests/vm/http_websocket_tests.rs"
+required-features = ["runtime", "http-client"]
+
 [[test]]
 name = "sqlite_host_tests"
 path = "tests/vm/sqlite_host_tests.rs"

README.md

@@ -14,6 +14,8 @@ The complete language, runtime, and implementation guides live on the [RustScrip
 - [RSS language](https://rustscript.org/docs/reference/rss/)
 - [Host functions](https://rustscript.org/docs/reference/host-functions/)
 - [Runtime controls and artifacts](https://rustscript.org/docs/reference/runtime-controls/)
+- [Callable-driven HTTP client contract](docs/http-client.md)
+- [Script call frames and callable values](docs/callable-runtime.md)
 - [Compiler frontend syntax and feature support](src/compiler/frontends/README.md)
 
 ## Crate usage

build.rs

@@ -162,12 +162,22 @@ fn main() {
             category: SourceCategory::DefaultHost,
         },
     ];
-    if env::var_os("CARGO_FEATURE_ASYNC").is_some() {
+    if env::var_os("CARGO_FEATURE_HTTP_CLIENT").is_some() {
         host_sources.push(SourceSpec {
-            path: "src/builtins/runtime/http.rs".to_string(),
+            path: "src/builtins/runtime/http/mod.rs".to_string(),
             module: "http".to_string(),
             category: SourceCategory::DefaultHost,
         });
+        host_sources.push(SourceSpec {
+            path: "src/builtins/runtime/http/websocket.rs".to_string(),
+            module: "http::websocket".to_string(),
+            category: SourceCategory::DefaultHost,
+        });
+        host_sources.push(SourceSpec {
+            path: "src/builtins/runtime/http/sse.rs".to_string(),
+            module: "http::sse".to_string(),
+            category: SourceCategory::DefaultHost,
+        });
     }
     if env::var_os("CARGO_FEATURE_SQLITE").is_some() {
         host_sources.push(SourceSpec {
@@ -1208,9 +1218,9 @@ fn render_callable_consts(callables: &[&CallableDecl]) -> String {
         for param in &callable.params {
             writeln!(
                 &mut out,
-                "    CallableParam {{ name: {:?}, ty: CallableParamType::{}, optional: {} }},",
+                "    CallableParam {{ name: {:?}, ty: {}, optional: {} }},",
                 param.name,
-                callable_param_variant(&param.ty_label),
+                callable_param_expr(&param.ty_label),
                 param.optional
             )
             .unwrap();
@@ -1633,18 +1643,38 @@ fn callable_const_base(callable: &CallableDecl) -> String {
     to_shouty_snake(&format!("{prefix}_{}", callable.rust_ident))
 }
 
-fn callable_param_variant(label: &str) -> &'static str {
+pub(crate) fn callable_param_expr(label: &str) -> String {
     match label {
-        "any" => "Any",
-        "null" => "Null",
-        "int" => "Int",
-        "float" => "Float",
-        "bool" => "Bool",
-        "string" => "String",
-        "bytes" => "Bytes",
-        "array" => "Array",
-        "map" => "Map",
-        "number" => "Number",
+        "any" => "CallableParamType::Any".to_string(),
+        "null" => "CallableParamType::Null".to_string(),
+        "int" => "CallableParamType::Int".to_string(),
+        "float" => "CallableParamType::Float".to_string(),
+        "bool" => "CallableParamType::Bool".to_string(),
+        "string" => "CallableParamType::String".to_string(),
+        "bytes" => "CallableParamType::Bytes".to_string(),
+        "array" => "CallableParamType::Array".to_string(),
+        "map" => "CallableParamType::Map".to_string(),
+        "number" => "CallableParamType::Number".to_string(),
+        other if other.starts_with("fn(") => {
+            let (params, result) = other
+                .strip_prefix("fn(")
+                .and_then(|value| value.split_once(") -> "))
+                .unwrap_or_else(|| panic!("invalid callable schema '{other}'"));
+            let params = if params.is_empty() {
+                Vec::new()
+            } else {
+                params
+                    .split(", ")
+                    .map(callable_param_expr)
+                    .collect::<Vec<_>>()
+            };
+            let result = callable_param_expr(result);
+            format!(
+                "CallableParamType::Callable(CallableType {{ params: &[{}], return_type: &{} }})",
+                params.join(", "),
+                result
+            )
+        }
         other => panic!("unsupported callable param type '{other}'"),
     }
 }
@@ -2070,7 +2100,7 @@ fn static_return_type_label(output: &ReturnType) -> String {
     value_type_from_label(&return_type_label(output)).to_string()
 }
 
-fn type_label(ty: &Type) -> String {
+pub(crate) fn type_label(ty: &Type) -> String {
     match ty {
         Type::Group(group) => type_label(&group.elem),
         Type::Paren(paren) => type_label(&paren.elem),
@@ -2108,6 +2138,7 @@ fn type_label(ty: &Type) -> String {
                 "Array" | "VmArray" | "VmArrayRef" | "VmArrayHandle" => "array".to_string(),
                 "Map" | "VmMap" | "VmMapRef" | "VmMapHandle" => "map".to_string(),
                 "Number" | "NumberValue" => "number".to_string(),
+                "VmCallable" => callable_type_label(segment),
                 "Unknown" | "UnknownValue" => "unknown".to_string(),
                 "CallOutcome" => "unknown".to_string(),
                 "Option" => {
@@ -2136,6 +2167,25 @@ fn type_label(ty: &Type) -> String {
     }
 }
 
+fn callable_type_label(segment: &syn::PathSegment) -> String {
+    let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
+        panic!("VmCallable requires a function signature");
+    };
+    let Some(syn::GenericArgument::Type(Type::BareFn(function))) = args.args.first() else {
+        panic!("VmCallable requires fn(...) -> ...");
+    };
+    let params = function
+        .inputs
+        .iter()
+        .map(|input| type_label(&input.ty))
+        .collect::<Vec<_>>();
+    let result = match &function.output {
+        ReturnType::Default => "null".to_string(),
+        ReturnType::Type(_, ty) => type_label(ty),
+    };
+    format!("fn({}) -> {result}", params.join(", "))
+}
+
 fn type_label_for_vec(segment: &syn::PathSegment) -> String {
     let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
         panic!("Vec<T> requires one generic argument");
@@ -2310,3 +2360,26 @@ fn find_matching_paren(source: &str) -> usize {
     }
     panic!("unterminated macro invocation");
 }
+
+#[cfg(test)]
+mod callable_schema_tests {
+    use super::*;
+    use syn::parse_quote;
+
+    #[test]
+    fn build_metadata_renders_typed_callable_parameters() {
+        let ty: Type = parse_quote!(VmCallable<fn(VmMap) -> VmMap>);
+        assert_eq!(type_label(&ty), "fn(map) -> map");
+        assert_eq!(
+            callable_param_expr("fn(map) -> map"),
+            "CallableParamType::Callable(CallableType { params: &[CallableParamType::Map], return_type: &CallableParamType::Map })"
+        );
+
+        let float_ty: Type = parse_quote!(VmCallable<fn(f64) -> f64>);
+        assert_eq!(type_label(&float_ty), "fn(float) -> float");
+        assert_eq!(
+            callable_param_expr("fn(float) -> float"),
+            "CallableParamType::Callable(CallableType { params: &[CallableParamType::Float], return_type: &CallableParamType::Float })"
+        );
+    }
+}

crates/rustscript/tests/alias_smoke.rs

@@ -53,6 +53,44 @@ fn alias_exports_public_invocation_stream_contract() {
     });
 }
 
+#[cfg(feature = "runtime")]
+#[test]
+fn alias_exports_public_host_callable_stream_embedding_api() {
+    use rustscript::{HostStreamAction, HostStreamDriver, HostStreamPoll, Vm};
+
+    struct CompileOnlyDriver;
+
+    impl HostStreamDriver for CompileOnlyDriver {
+        fn poll_next(
+            &mut self,
+            _cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<rustscript::VmResult<HostStreamPoll>> {
+            unreachable!("compile-only API smoke")
+        }
+
+        fn apply_action(
+            &mut self,
+            _action: rustscript::Value,
+        ) -> rustscript::VmResult<HostStreamAction> {
+            unreachable!("compile-only API smoke")
+        }
+    }
+
+    fn submit(
+        vm: &mut Vm,
+        callback: rustscript::Value,
+        driver: CompileOnlyDriver,
+    ) -> rustscript::VmResult<rustscript::CallOutcome> {
+        vm.submit_callable_stream(callback, driver)
+    }
+
+    let _submit: fn(
+        &mut Vm,
+        rustscript::Value,
+        CompileOnlyDriver,
+    ) -> rustscript::VmResult<rustscript::CallOutcome> = submit;
+}
+
 #[cfg(feature = "http-client")]
 #[test]
 fn alias_http_client_includes_runtime_contract() {

docs/callable-runtime.md

@@ -78,6 +78,14 @@ PDRC recordings preserve full execution-frame metadata. Callable environments us
 
 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.
 
+## Callable-driven HTTP streams
+
+With the `http-client` feature, `http::client::request(request)`, `http::client::sse(request, on_event)`, and `http::client::websocket(request, on_event)` are script-facing host imports. The two streaming calls are long-running ordinary host calls. Each handler has the schema `fn(map) -> map`. The host produces one protocol item, the VM runs one child callback frame, and the returned action controls continuation or a WebSocket write before another item can arrive at the VM boundary.
+
+The callback may yield or wait in an ordinary async host call. Existing frame machinery resumes the callback first and returns its final action to the suspended HTTP call. The network future does not own or enter the VM and is not polled while the callback is active, so at most one item remains unacknowledged and callback completion supplies backpressure.
+
+Each buffered, SSE, and WebSocket import is an independent capability. Streaming calls expose no script request IDs, handles, detached resources, `next`, `close`, or cancellation callables. Their complete event maps, action maps, terminal summaries, bounds, destination policy, and lifecycle contract are documented in [HTTP client callable contract](http-client.md).
+
 ## Optimized backends
 
 Whole-program AOT and Trace JIT use the same builtin call path (static catalog IDs) for environment binding, native frame dispatch for `callvalue`, and prototype-direct native dispatch for `callscript`. Script-frame entry and return preserve frame-relative locals and typed continuations.