feat(http): add bounded HTTP and callable SSE clients

Cargo.toml

@@ -29,6 +29,17 @@ name = "vm"
 default = ["runtime", "cli", "cranelift-jit"]
 runtime = []
 async = ["runtime", "dep:tokio"]
+http-client = [
+    "async",
+    "dep:futures-util",
+    "dep:http-body-util",
+    "dep:hyper",
+    "dep:hyper-util",
+    "dep:rustls",
+    "dep:tokio-rustls",
+    "dep:url",
+    "dep:webpki-roots",
+]
 sqlite = ["runtime", "dep:rusqlite"]
 edge-abi = [
     "dep:edge_abi",
@@ -79,7 +90,6 @@ 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" }
 rusqlite = { version = "0.32", default-features = false, features = ["bundled", "hooks", "limits"], optional = true }
-tokio = { version = "1", features = ["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"
@@ -90,6 +100,17 @@ rt-format = "0.3.1"
 self_cell = "1"
 rustyline = { version = "14", optional = true }
 
+[target.'cfg(not(target_family = "wasm"))'.dependencies]
+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 }
+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", "macros"], optional = true }
+webpki-roots = { version = "1", optional = true }
+
 [target.'cfg(windows)'.dependencies]
 windows-sys = { version = "0.59", features = ["Win32_System_Diagnostics_Debug", "Win32_System_Memory", "Win32_System_ProcessStatus", "Win32_System_Threading"] }
 
@@ -99,6 +120,11 @@ libc = "0.2"
 [dev-dependencies]
 pd-host-schema = { path = "./crates/pd-host-schema", version = "0.1.0" }
 syn = { version = "2", features = ["full"] }
+
+[target.'cfg(not(target_family = "wasm"))'.dev-dependencies]
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
+
+[target.'cfg(target_family = "wasm")'.dev-dependencies]
 tokio = { version = "1", features = ["macros", "rt", "time", "sync"] }
 
 [[test]]
@@ -126,6 +152,26 @@ name = "host_context_arch_tests"
 path = "tests/host_context_arch_tests.rs"
 required-features = ["runtime"]
 
+[[test]]
+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 = "io_http_coexistence_tests"
+path = "tests/vm/io_http_coexistence_tests.rs"
+required-features = ["runtime", "http-client"]
+
+[[test]]
+name = "http_feature_gating_tests"
+path = "tests/http_feature_gating_tests.rs"
+required-features = ["runtime"]
+
 [build-dependencies]
 pd-host-schema = { path = "./crates/pd-host-schema", version = "0.1.0" }
 syn = { version = "2", features = ["full"] }

build.rs

@@ -129,6 +129,17 @@ struct NamespaceDecl {
     runtime_supported_on_wasm: bool,
 }
 
+/// HTTP/SSE is a native transport extension. Keep this predicate identical to
+/// the `cfg` boundary used by the runtime and public exports: the Cargo
+/// feature remains selectable on wasm, but it must not publish transport
+/// sources or generated host/catalog entries there.
+pub(crate) fn http_transport_enabled(http_client_feature: bool, target_family: &str) -> bool {
+    http_client_feature
+        && !target_family
+            .split(',')
+            .any(|family| family.trim() == "wasm")
+}
+
 #[derive(Clone, Debug)]
 struct Group<'a> {
     key: String,
@@ -166,7 +177,8 @@ fn main() {
         catalog.retain(|entry| !entry.source_name.starts_with("sqlite::"));
     }
 
-    let host_sources = vec![
+    let target_family = env::var("CARGO_CFG_TARGET_FAMILY").expect("missing target family");
+    let mut host_sources = vec![
         SourceSpec {
             path: "src/builtins/runtime/host.rs".to_string(),
             module: "host".to_string(),
@@ -178,6 +190,21 @@ fn main() {
             category: SourceCategory::DefaultHost,
         },
     ];
+    if http_transport_enabled(
+        env::var_os("CARGO_FEATURE_HTTP_CLIENT").is_some(),
+        &target_family,
+    ) {
+        host_sources.push(SourceSpec {
+            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/sse.rs".to_string(),
+            module: "http::sse".to_string(),
+            category: SourceCategory::DefaultHost,
+        });
+    }
     let async_enabled = env::var_os("CARGO_FEATURE_ASYNC").is_some();
     let target_arch = env::var("CARGO_CFG_TARGET_ARCH").expect("missing target architecture");
     let builtin_sources = builtin_source_specs(&namespaces, async_enabled, &target_arch);
@@ -2216,11 +2243,21 @@ fn find_matching_paren(source: &str) -> usize {
 #[cfg(test)]
 mod tests {
     use super::{
-        HostExecutionKind, NamespaceDecl, SourceCategory, builtin_source_specs, parse_source_file,
-        select_io_source_path,
+        HostExecutionKind, NamespaceDecl, SourceCategory, builtin_source_specs,
+        http_transport_enabled, parse_source_file, select_io_source_path,
     };
     use std::path::Path;
 
+    #[test]
+    fn http_transport_predicate_matches_source_and_catalog_boundary() {
+        assert!(http_transport_enabled(true, "unix"));
+        assert!(http_transport_enabled(true, "windows"));
+        assert!(!http_transport_enabled(true, "wasm"));
+        assert!(!http_transport_enabled(true, "wasm,unix"));
+        assert!(!http_transport_enabled(false, "unix"));
+        assert!(!http_transport_enabled(false, "wasm"));
+    }
+
     fn io_namespace() -> NamespaceDecl {
         NamespaceDecl {
             namespace: "io".to_string(),

docs/http-client.md

@@ -0,0 +1,61 @@
+# Native HTTP client and SSE feature
+
+The `http-client` Cargo feature enables the buffered HTTP request and callable
+SSE host builtins on supported native targets. The feature name remains valid
+on every target so workspace feature selection stays uniform, but the native
+transport implementation is target-gated.
+
+## Target boundary
+
+The transport is compiled when both conditions hold:
+
+- the `http-client` feature is enabled; and
+- the target is not in Rust's `wasm` target family (`not(target_family = "wasm")`).
+
+On `wasm32-unknown-unknown` and other wasm-family targets, enabling
+`http-client` is intentionally a no-op for transport publication. Cargo does
+not build the native Tokio networking, Hyper, Rustls, URL, or HTTP-body
+transport dependencies. The HTTP module, `HttpConfig`/`HttpExtension`/
+`HttpHostExt` exports, HTTP builtins and callables, HTTP catalog functions, and
+LSP standard-catalog entries are absent. Browser or other wasm networking must
+be supplied by the embedding host instead.
+
+The build script uses the same target-family boundary when it selects host
+source files for generated dispatch and catalog metadata. This keeps the
+compiled runtime surface and generated metadata synchronized.
+
+## Native API
+
+On a supported native target, enabling `http-client` preserves the public API:
+
+- `HttpConfig` controls request and stream limits, redirects, timeouts, and
+  capability policy;
+- `HttpExtension` and `HttpHostExt` install the native HTTP host integration;
+- `register_http_builtin_module` and `http_host_catalog` expose the native
+  resource schema and callable metadata;
+- `http::client::request` returns a bounded response map; and
+- `http::client::sse` drives a bounded SSE stream through a script callback.
+
+The HTTP and SSE behavior, resource lifecycle, cancellation, and native async
+bridge contracts are unchanged by the wasm boundary. See
+[`callable-runtime.md`](callable-runtime.md) for the general callable and
+host-runtime contract.
+
+## Feature checks
+
+For a native HTTP build:
+
+```bash
+cargo test -p pd-vm --no-default-features --features runtime,http-client --test http_feature_gating_tests
+```
+
+For the wasm gating check, keep the feature enabled while selecting a wasm
+package or target:
+
+```bash
+cargo check -p pd-vm --no-default-features --features runtime,http-client \
+  --target wasm32-unknown-unknown
+```
+
+This verifies that feature selection is accepted without publishing the native
+transport surface.

src/builtins/mod.rs

@@ -5,7 +5,7 @@ mod metadata;
 #[cfg(feature = "runtime")]
 pub(crate) mod runtime;
 
-#[cfg(test)]
+#[allow(unused_imports)]
 pub use self::metadata::CallableType;
 pub use self::metadata::{
     CallableDef, CallableParam, CallableParamType, CallableSignature, HostExecution,

src/builtins/runtime/http/config.rs

@@ -0,0 +1,100 @@
+use std::time::Duration;
+
+use crate::vm::{VmError, VmResult};
+
+/// Bounded network policy for the built-in HTTP client and future streaming adapters.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct HttpConfig {
+    pub allowed_schemes: Vec<String>,
+    pub allowed_hosts: Vec<String>,
+    pub allowed_ports: Vec<u16>,
+    pub max_redirects: usize,
+    pub max_request_body_bytes: usize,
+    /// Maximum number of caller-supplied request header fields. Client-managed
+    /// fields such as `Host` are outside this extension surface.
+    pub max_request_header_count: usize,
+    /// Maximum serialized size of the caller-supplied request header block.
+    /// Every field contributes `name + ": " + value + "\\r\\n"`; the final
+    /// terminating `"\\r\\n"` is included as well.
+    pub max_request_header_bytes: usize,
+    pub max_response_body_bytes: usize,
+    pub connect_timeout: Duration,
+    pub request_timeout: Duration,
+    pub allow_private_ips: bool,
+    pub max_stream_item_bytes: usize,
+    pub max_stream_total_bytes: usize,
+    pub max_sse_line_bytes: usize,
+    pub max_stream_duration: Duration,
+    pub stream_idle_timeout: Duration,
+}
+
+impl HttpConfig {
+    /// Validates limits that must remain positive for every streaming adapter.
+    pub fn validate(&self) -> VmResult<()> {
+        let positive_limits = [
+            ("max_stream_item_bytes", self.max_stream_item_bytes),
+            ("max_stream_total_bytes", self.max_stream_total_bytes),
+            ("max_sse_line_bytes", self.max_sse_line_bytes),
+        ];
+        if let Some((name, _)) = positive_limits.iter().find(|(_, value)| *value == 0) {
+            return Err(VmError::HostError(format!(
+                "HTTP configuration field '{name}' must be positive"
+            )));
+        }
+        let positive_header_limits = [
+            ("max_request_header_count", self.max_request_header_count),
+            ("max_request_header_bytes", self.max_request_header_bytes),
+        ];
+        if let Some((name, _)) = positive_header_limits.iter().find(|(_, value)| *value == 0) {
+            return Err(VmError::HostError(format!(
+                "HTTP configuration field '{name}' must be positive"
+            )));
+        }
+        let positive_timeouts = [
+            ("connect_timeout", self.connect_timeout),
+            ("request_timeout", self.request_timeout),
+            ("max_stream_duration", self.max_stream_duration),
+            ("stream_idle_timeout", self.stream_idle_timeout),
+        ];
+        if let Some((name, _)) = positive_timeouts
+            .iter()
+            .find(|(_, timeout)| timeout.is_zero())
+        {
+            return Err(VmError::HostError(format!(
+                "HTTP configuration field '{name}' must be positive"
+            )));
+        }
+        if let Some((name, _)) = positive_timeouts
+            .iter()
+            .find(|(_, timeout)| std::time::Instant::now().checked_add(*timeout).is_none())
+        {
+            return Err(VmError::HostError(format!(
+                "HTTP configuration field '{name}' is too large"
+            )));
+        }
+        Ok(())
+    }
+}
+
+impl Default for HttpConfig {
+    fn default() -> Self {
+        Self {
+            allowed_schemes: vec!["https".to_string()],
+            allowed_hosts: Vec::new(),
+            allowed_ports: Vec::new(),
+            max_redirects: 5,
+            max_request_body_bytes: 1024 * 1024,
+            max_request_header_count: 100,
+            max_request_header_bytes: 64 * 1024,
+            max_response_body_bytes: 8 * 1024 * 1024,
+            connect_timeout: Duration::from_secs(10),
+            request_timeout: Duration::from_secs(30),
+            allow_private_ips: false,
+            max_stream_item_bytes: 1024 * 1024,
+            max_stream_total_bytes: 64 * 1024 * 1024,
+            max_sse_line_bytes: 64 * 1024,
+            max_stream_duration: Duration::from_secs(5 * 60),
+            stream_idle_timeout: Duration::from_secs(30),
+        }
+    }
+}