Skip to main content

lex_runtime/handler/
http_serve.rs

1//! HTTP servers behind `net.serve*`: tiny_http / hyper / TLS / QUIC entry points, route matching, `ServeOpts`, and the request / response conversion between Lex records and the wire.
2
3use super::*;
4
5/// Blocks the calling thread, accepts incoming HTTP requests on
6/// `127.0.0.1:port`, and dispatches each through the named Lex
7/// stage. Each request gets a fresh `Vm`; the program and policy
8/// are shared.
9///
10/// Handler signature in Lex (by convention):
11///   fn <name>(req :: Record { method :: Str, path :: Str, body :: Str })
12///        -> Record { status :: Int, body :: Str }
13/// PEM-encoded certificate + private key, both as raw bytes.
14pub struct TlsConfig {
15    pub cert: Vec<u8>,
16    pub key: Vec<u8>,
17}
18
19pub(super) fn serve_http(
20    port: u16,
21    handler_name: String,
22    program: Arc<Program>,
23    policy: Policy,
24    tls: Option<TlsConfig>,
25    opts: ServeOpts,
26) -> Result<Value, String> {
27    match tls {
28        None => serve_http_plain(port, handler_name, program, policy, opts),
29        Some(cfg) => serve_http_tls_legacy(port, handler_name, program, policy, cfg),
30    }
31}
32
33/// Hyper 1.x + Tokio multi-thread HTTP/1.1 server for `net.serve`.
34/// Each connection is accepted in an async task; the synchronous Lex VM
35/// call runs inside `spawn_blocking` so it doesn't block the executor.
36///
37/// `LEX_NET_INLINE_VM=1` (or `=true`) skips the `spawn_blocking` hop and
38/// runs the VM directly on the tokio worker. Faster for handlers that
39/// return in tens of microseconds; pathological if handlers do real
40/// CPU/blocking work, since they stall the worker. Experimental — see
41/// lex-lang issue #431.
42pub(super) fn serve_http_plain(
43    port: u16,
44    handler_name: String,
45    program: Arc<Program>,
46    policy: Policy,
47    opts: ServeOpts,
48) -> Result<Value, String> {
49    use http_body_util::BodyExt as _;
50    use hyper::server::conn::http1;
51    use hyper::service::service_fn;
52    use hyper_util::rt::{TokioExecutor, TokioIo};
53    use hyper_util::server::conn::auto;
54    use tokio::net::TcpListener as TokioTcpListener;
55
56    let inline_vm = opts.inline_vm;
57    let http2 = opts.http2;
58    let host = opts.host.clone();
59    let rt = tokio::runtime::Builder::new_multi_thread()
60        .enable_all()
61        .build()
62        .map_err(|e| format!("net.serve: tokio runtime: {e}"))?;
63    rt.block_on(async move {
64        let listener = TokioTcpListener::bind((host.as_str(), port))
65            .await
66            .map_err(|e| format!("net.serve bind {host}:{port}: {e}"))?;
67        eprintln!(
68            "net.serve: listening on http://{host}:{port}{}{}",
69            if inline_vm { " (inline-vm)" } else { "" },
70            if http2 { " (http1+http2)" } else { "" }
71        );
72        loop {
73            let (stream, _) = listener
74                .accept()
75                .await
76                .map_err(|e| format!("net.serve accept: {e}"))?;
77            let io = TokioIo::new(stream);
78            let program = Arc::clone(&program);
79            let policy = policy.clone();
80            let handler_name = handler_name.clone();
81            tokio::spawn(async move {
82                let program2 = Arc::clone(&program);
83                let policy2 = policy.clone();
84                let handler_name2 = handler_name.clone();
85                let svc = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
86                    let program = Arc::clone(&program2);
87                    let policy = policy2.clone();
88                    let handler_name = handler_name2.clone();
89                    async move {
90                        let (parts, body) = req.into_parts();
91                        let body_bytes = body
92                            .collect()
93                            .await
94                            .map(|c| c.to_bytes())
95                            .unwrap_or_default();
96                        let result = if inline_vm {
97                            // Inline path — run the VM on this tokio worker.
98                            // Cheap when handlers return in microseconds; will
99                            // stall the worker on heavy handlers (caveat per #431).
100                            let lex_req = build_request_value_parts(&parts, &body_bytes);
101                            let handler = DefaultHandler::new(policy)
102                                .with_program(Arc::clone(&program));
103                            let mut vm = Vm::with_handler(&program, Box::new(handler));
104                            let r = vm.call(&handler_name, vec![lex_req]);
105                            // Unpack inline so the VM is still in
106                            // scope (#463 wire-up).
107                            Ok(r.map(|v| unpack_response(&mut vm, &v)))
108                        } else {
109                            tokio::task::spawn_blocking(move || {
110                                let lex_req = build_request_value_parts(&parts, &body_bytes);
111                                let handler = DefaultHandler::new(policy)
112                                    .with_program(Arc::clone(&program));
113                                let mut vm = Vm::with_handler(&program, Box::new(handler));
114                                let r = vm.call(&handler_name, vec![lex_req]);
115                                r.map(|v| unpack_response(&mut vm, &v))
116                            })
117                            .await
118                        };
119                        Ok::<_, std::convert::Infallible>(match result {
120                            Ok(Ok(unpacked)) => build_hyper_response(unpacked),
121                            Ok(Err(e)) => error_response(500, &format!("internal error: {e}")),
122                            Err(e) => error_response(500, &format!("task panicked: {e}")),
123                        })
124                    }
125                });
126                let result = if http2 {
127                    auto::Builder::new(TokioExecutor::new())
128                        .serve_connection(io, svc)
129                        .await
130                        .map_err(|e| e.to_string())
131                } else {
132                    http1::Builder::new()
133                        .serve_connection(io, svc)
134                        .await
135                        .map_err(|e| e.to_string())
136                };
137                if let Err(e) = result {
138                    eprintln!("net.serve: connection error: {e}");
139                }
140            });
141        }
142    })
143}
144
145/// TLS path: still uses tiny_http pending a tokio-rustls migration.
146pub(super) fn serve_http_tls_legacy(
147    port: u16,
148    handler_name: String,
149    program: Arc<Program>,
150    policy: Policy,
151    cfg: TlsConfig,
152) -> Result<Value, String> {
153    let ssl = tiny_http::SslConfig {
154        certificate: cfg.cert,
155        private_key: cfg.key,
156    };
157    let server = tiny_http::Server::https(("0.0.0.0", port), ssl)
158        .map_err(|e| format!("net.serve_tls bind {port}: {e}"))?;
159    eprintln!("net.serve: listening on https://0.0.0.0:{port}");
160    for req in server.incoming_requests() {
161        let program = Arc::clone(&program);
162        let policy = policy.clone();
163        let handler_name = handler_name.clone();
164        std::thread::spawn(move || handle_request_tls(req, program, policy, handler_name));
165    }
166    Ok(Value::Unit)
167}
168
169pub(super) fn handle_request_tls(
170    mut req: tiny_http::Request,
171    program: Arc<Program>,
172    policy: Policy,
173    handler_name: String,
174) {
175    let lex_req = build_request_value_tiny(&mut req);
176    let handler = DefaultHandler::new(policy).with_program(Arc::clone(&program));
177    let mut vm = Vm::with_handler(&program, Box::new(handler));
178    match vm.call(&handler_name, vec![lex_req]) {
179        Ok(resp) => {
180            // Drain lazy iters + read response fields straight out of
181            // any arena handles while the VM is still in scope — see
182            // #477 and `docs/design/arena-plumbing.md` § "Status
183            // update (2026-06-05)" for why this is a single fused
184            // step now.
185            let (status, body, headers) = unpack_response(&mut vm, &resp);
186            respond_with_body_tls(req, status, body, headers);
187        }
188        Err(e) => {
189            let response = tiny_http::Response::from_string(format!("internal error: {e}"))
190                .with_status_code(500);
191            let _ = req.respond(response);
192        }
193    }
194}
195
196/// Hyper 1.x + Tokio multi-thread HTTP/1.1 server for `net.serve_fn`.
197///
198/// `LEX_NET_INLINE_VM=1` skips `spawn_blocking` — see `serve_http_plain`'s
199/// doc-comment for the tradeoffs. Same env var gates both paths.
200pub(super) fn serve_http_fn(
201    port: u16,
202    closure: Value,
203    program: Arc<Program>,
204    policy: Policy,
205    opts: ServeOpts,
206) -> Result<Value, String> {
207    use http_body_util::BodyExt as _;
208    use hyper::server::conn::http1;
209    use hyper::service::service_fn;
210    use hyper_util::rt::{TokioExecutor, TokioIo};
211    use hyper_util::server::conn::auto;
212    use tokio::net::TcpListener as TokioTcpListener;
213
214    let inline_vm = opts.inline_vm;
215    let http2 = opts.http2;
216    let host = opts.host.clone();
217    let rt = tokio::runtime::Builder::new_multi_thread()
218        .enable_all()
219        .build()
220        .map_err(|e| format!("net.serve_fn: tokio runtime: {e}"))?;
221    rt.block_on(async move {
222        let listener = TokioTcpListener::bind((host.as_str(), port))
223            .await
224            .map_err(|e| format!("net.serve_fn bind {host}:{port}: {e}"))?;
225        eprintln!(
226            "net.serve_fn: listening on http://{host}:{port}{}{}",
227            if inline_vm { " (inline-vm)" } else { "" },
228            if http2 { " (http1+http2)" } else { "" }
229        );
230        loop {
231            let (stream, _) = listener
232                .accept()
233                .await
234                .map_err(|e| format!("net.serve_fn accept: {e}"))?;
235            let io = TokioIo::new(stream);
236            let program = Arc::clone(&program);
237            let policy = policy.clone();
238            let closure = closure.clone();
239            tokio::spawn(async move {
240                let program2 = Arc::clone(&program);
241                let policy2 = policy.clone();
242                let closure2 = closure.clone();
243                let svc = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
244                    let program = Arc::clone(&program2);
245                    let policy = policy2.clone();
246                    let closure = closure2.clone();
247                    async move {
248                        let (parts, body) = req.into_parts();
249                        let body_bytes = body
250                            .collect()
251                            .await
252                            .map(|c| c.to_bytes())
253                            .unwrap_or_default();
254                        let result = if inline_vm {
255                            let lex_req = build_request_value_parts(&parts, &body_bytes);
256                            let handler = DefaultHandler::new(policy)
257                                .with_program(Arc::clone(&program));
258                            let mut vm = Vm::with_handler(&program, Box::new(handler));
259                            // #463 scaffolding — bracket the user
260                            // handler with a request scope so the
261                            // arena lifecycle is exercised. The
262                            // arena itself is unused today; this
263                            // proves the lifecycle is sound for the
264                            // follow-on Value-rep slice.
265                            let scope = vm.enter_request_scope();
266                            let r = vm.invoke_closure_value(closure, vec![lex_req]);
267                            // Unpack inline so the VM is still in
268                            // scope for both lazy-iter draining and
269                            // slab-direct field reads (#463).
270                            let r = r.map(|v| unpack_response(&mut vm, &v));
271                            vm.exit_request_scope(scope);
272                            Ok(r)
273                        } else {
274                            tokio::task::spawn_blocking(move || {
275                                let lex_req = build_request_value_parts(&parts, &body_bytes);
276                                let handler = DefaultHandler::new(policy)
277                                    .with_program(Arc::clone(&program));
278                                let mut vm = Vm::with_handler(&program, Box::new(handler));
279                                let scope = vm.enter_request_scope();
280                                let r = vm.invoke_closure_value(closure, vec![lex_req]);
281                                let r = r.map(|v| unpack_response(&mut vm, &v));
282                                vm.exit_request_scope(scope);
283                                r
284                            })
285                            .await
286                        };
287                        Ok::<_, std::convert::Infallible>(match result {
288                            Ok(Ok(unpacked)) => build_hyper_response(unpacked),
289                            Ok(Err(e)) => error_response(500, &format!("internal error: {e}")),
290                            Err(e) => error_response(500, &format!("task panicked: {e}")),
291                        })
292                    }
293                });
294                let result = if http2 {
295                    auto::Builder::new(TokioExecutor::new())
296                        .serve_connection(io, svc)
297                        .await
298                        .map_err(|e| e.to_string())
299                } else {
300                    http1::Builder::new()
301                        .serve_connection(io, svc)
302                        .await
303                        .map_err(|e| e.to_string())
304                };
305                if let Err(e) = result {
306                    eprintln!("net.serve_fn: connection error: {e}");
307                }
308            });
309        }
310    })
311}
312
313/// Compiled segment of a route pattern. Patterns are split on `/`
314/// once at registration time so the per-request match loop is just a
315/// length check + segment-by-segment compare.
316#[derive(Clone, Debug)]
317pub(crate) enum RouteSeg {
318    Literal(String),
319    /// `:name` capture — binds the request segment under `name` in
320    /// `req.path_params`.
321    Param(String),
322}
323
324/// Compile a `:name`-style pattern (e.g. `"/users/:id/posts"`) into a
325/// segment list. Errors out at registration time so bad patterns
326/// surface before the server binds, not on the first matching request.
327pub(super) fn compile_path_pattern(pat: &str) -> Result<Vec<RouteSeg>, String> {
328    if pat.is_empty() {
329        return Err("path pattern must be non-empty (use \"/\" for the root)".into());
330    }
331    if !pat.starts_with('/') {
332        return Err(format!("path pattern must start with '/' (got {pat:?})"));
333    }
334    let mut segs = Vec::new();
335    for raw in pat.split('/') {
336        if let Some(name) = raw.strip_prefix(':') {
337            if name.is_empty() {
338                return Err(format!(
339                    ":-segment in pattern {pat:?} must have a name (e.g. :id)"
340                ));
341            }
342            segs.push(RouteSeg::Param(name.to_string()));
343        } else {
344            segs.push(RouteSeg::Literal(raw.to_string()));
345        }
346    }
347    Ok(segs)
348}
349
350/// Attempt to match a request `path` against a compiled pattern. On
351/// success returns the captured `:name` segments as a Lex-shaped map
352/// keyed by `MapKey::Str(name)`; on mismatch returns `None`. Strict
353/// segment-count match: trailing slashes matter (caller registers
354/// both forms if both should match).
355pub(super) fn match_path_pattern(
356    segs: &[RouteSeg],
357    path: &str,
358) -> Option<std::collections::BTreeMap<lex_bytecode::MapKey, Value>> {
359    let path_segs: Vec<&str> = path.split('/').collect();
360    if path_segs.len() != segs.len() {
361        return None;
362    }
363    let mut params = std::collections::BTreeMap::new();
364    for (pat, p) in segs.iter().zip(path_segs.iter()) {
365        match pat {
366            RouteSeg::Literal(lit) => {
367                if lit != p {
368                    return None;
369                }
370            }
371            RouteSeg::Param(name) => {
372                params.insert(
373                    lex_bytecode::MapKey::Str(name.clone()),
374                    Value::Str((*p).into()),
375                );
376            }
377        }
378    }
379    Some(params)
380}
381
382/// Decode the `routes` argument of `net.serve_routed` into a vector
383/// of `(uppercased-method-or-"*", compiled-pattern, handler-closure)`.
384/// Validates and pre-compiles up front so malformed routes fail before
385/// the server starts.
386pub(super) fn decode_routes_arg(
387    v: Value,
388) -> Result<Vec<(String, Vec<RouteSeg>, Value)>, String> {
389    let list = match v {
390        Value::List(xs) => xs,
391        _ => return Err("net.serve_routed: routes must be a List".into()),
392    };
393    let mut out = Vec::with_capacity(list.len());
394    for (i, item) in list.into_iter().enumerate() {
395        let tup = match item {
396            Value::Tuple(xs) if xs.len() == 3 => xs,
397            other => return Err(format!(
398                "net.serve_routed: route #{i} must be a (method, pattern, handler) 3-tuple, got {other:?}"
399            )),
400        };
401        let mut it = tup.into_iter();
402        let method_raw = match it.next() {
403            Some(Value::Str(s)) => s.to_string(),
404            _ => return Err(format!("net.serve_routed: route #{i} method must be Str")),
405        };
406        // Normalise method to uppercase for matching. "*" stays as-is.
407        let method = if method_raw == "*" { method_raw } else { method_raw.to_uppercase() };
408        let pattern = match it.next() {
409            Some(Value::Str(s)) => s.to_string(),
410            _ => return Err(format!("net.serve_routed: route #{i} path-pattern must be Str")),
411        };
412        let segs = compile_path_pattern(&pattern)
413            .map_err(|e| format!("net.serve_routed: route #{i} ({pattern:?}): {e}"))?;
414        let closure = match it.next() {
415            Some(c @ Value::Closure { .. }) => c,
416            _ => return Err(format!("net.serve_routed: route #{i} handler must be a closure")),
417        };
418        out.push((method, segs, closure));
419    }
420    Ok(out)
421}
422
423/// Pick the first matching route for `(method, path)` and return its
424/// handler closure plus captured path-params. Method match is
425/// case-insensitive vs the request (already uppercased at decode
426/// time); `"*"` in a route matches any method.
427pub(crate) fn dispatch_route<'a>(
428    routes: &'a [(String, Vec<RouteSeg>, Value)],
429    req_method: &str,
430    req_path: &str,
431) -> Option<(&'a Value, std::collections::BTreeMap<lex_bytecode::MapKey, Value>)> {
432    let req_method_upper = req_method.to_ascii_uppercase();
433    for (m, segs, closure) in routes {
434        if m != "*" && m != &req_method_upper {
435            continue;
436        }
437        if let Some(params) = match_path_pattern(segs, req_path) {
438            return Some((closure, params));
439        }
440    }
441    None
442}
443
444/// Overwrite the `path_params` field on a Request record with the
445/// captured map. Request records are always built with an empty
446/// `path_params` field, so this just updates the existing slot.
447pub(crate) fn stamp_path_params(
448    req: &mut Value,
449    params: std::collections::BTreeMap<lex_bytecode::MapKey, Value>,
450) {
451    if let Value::Record { fields: rec, .. } = req {
452        rec.insert("path_params".into(), Value::Map(params));
453    }
454}
455
456/// Hyper 1.x + Tokio multi-thread HTTP/1.1 server for `net.serve_routed`.
457/// Mirrors `serve_http_fn` (#431 inline-vm gate also applies); the only
458/// difference is that route dispatch picks the closure per-request from
459/// the precompiled `routes` table, falling back to the `fallback`
460/// closure when no route matches.
461pub(super) fn serve_http_routed(
462    port: u16,
463    routes: Vec<(String, Vec<RouteSeg>, Value)>,
464    fallback: Value,
465    program: Arc<Program>,
466    policy: Policy,
467    opts: ServeOpts,
468) -> Result<Value, String> {
469    use http_body_util::BodyExt as _;
470    use hyper::server::conn::http1;
471    use hyper::service::service_fn;
472    use hyper_util::rt::{TokioExecutor, TokioIo};
473    use hyper_util::server::conn::auto;
474    use tokio::net::TcpListener as TokioTcpListener;
475
476    let inline_vm = opts.inline_vm;
477    let http2 = opts.http2;
478    let host = opts.host.clone();
479    let routes = Arc::new(routes);
480    let rt = tokio::runtime::Builder::new_multi_thread()
481        .enable_all()
482        .build()
483        .map_err(|e| format!("net.serve_routed: tokio runtime: {e}"))?;
484    rt.block_on(async move {
485        let listener = TokioTcpListener::bind((host.as_str(), port))
486            .await
487            .map_err(|e| format!("net.serve_routed bind {host}:{port}: {e}"))?;
488        eprintln!(
489            "net.serve_routed: listening on http://{host}:{port} ({} routes{}{})",
490            routes.len(),
491            if inline_vm { ", inline-vm" } else { "" },
492            if http2 { ", http1+http2" } else { "" }
493        );
494        loop {
495            let (stream, _) = listener
496                .accept()
497                .await
498                .map_err(|e| format!("net.serve_routed accept: {e}"))?;
499            let io = TokioIo::new(stream);
500            let program = Arc::clone(&program);
501            let policy = policy.clone();
502            let routes = Arc::clone(&routes);
503            let fallback = fallback.clone();
504            tokio::spawn(async move {
505                let program2 = Arc::clone(&program);
506                let policy2 = policy.clone();
507                let routes2 = Arc::clone(&routes);
508                let fallback2 = fallback.clone();
509                let svc = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
510                    let program = Arc::clone(&program2);
511                    let policy = policy2.clone();
512                    let routes = Arc::clone(&routes2);
513                    let fallback = fallback2.clone();
514                    async move {
515                        let (parts, body) = req.into_parts();
516                        let body_bytes = body
517                            .collect()
518                            .await
519                            .map(|c| c.to_bytes())
520                            .unwrap_or_default();
521                        let method = parts.method.as_str().to_string();
522                        let path = match parts.uri.path() {
523                            "" => "/".to_string(),
524                            p => p.to_string(),
525                        };
526                        let result = if inline_vm {
527                            let mut lex_req = build_request_value_parts(&parts, &body_bytes);
528                            let (closure, params) = match dispatch_route(&routes, &method, &path) {
529                                Some((c, p)) => (c.clone(), p),
530                                None => (fallback.clone(), std::collections::BTreeMap::new()),
531                            };
532                            stamp_path_params(&mut lex_req, params);
533                            let handler = DefaultHandler::new(policy)
534                                .with_program(Arc::clone(&program));
535                            let mut vm = Vm::with_handler(&program, Box::new(handler));
536                            let r = vm.invoke_closure_value(closure, vec![lex_req]);
537                            // Unpack inline so the VM is still in
538                            // scope (#463 wire-up, see arena-plumbing.md).
539                            Ok(r.map(|v| unpack_response(&mut vm, &v)))
540                        } else {
541                            tokio::task::spawn_blocking(move || {
542                                let mut lex_req = build_request_value_parts(&parts, &body_bytes);
543                                let (closure, params) = match dispatch_route(&routes, &method, &path) {
544                                    Some((c, p)) => (c.clone(), p),
545                                    None => (fallback.clone(), std::collections::BTreeMap::new()),
546                                };
547                                stamp_path_params(&mut lex_req, params);
548                                let handler = DefaultHandler::new(policy)
549                                    .with_program(Arc::clone(&program));
550                                let mut vm = Vm::with_handler(&program, Box::new(handler));
551                                let r = vm.invoke_closure_value(closure, vec![lex_req]);
552                                r.map(|v| unpack_response(&mut vm, &v))
553                            })
554                            .await
555                        };
556                        Ok::<_, std::convert::Infallible>(match result {
557                            Ok(Ok(unpacked)) => build_hyper_response(unpacked),
558                            Ok(Err(e)) => error_response(500, &format!("internal error: {e}")),
559                            Err(e) => error_response(500, &format!("task panicked: {e}")),
560                        })
561                    }
562                });
563                let result = if http2 {
564                    auto::Builder::new(TokioExecutor::new())
565                        .serve_connection(io, svc)
566                        .await
567                        .map_err(|e| e.to_string())
568                } else {
569                    http1::Builder::new()
570                        .serve_connection(io, svc)
571                        .await
572                        .map_err(|e| e.to_string())
573                };
574                if let Err(e) = result {
575                    eprintln!("net.serve_routed: connection error: {e}");
576                }
577            });
578        }
579    })
580}
581
582/// Read `LEX_NET_INLINE_VM` and report whether the runtime should skip
583/// `spawn_blocking` on the per-request VM call. Accepts `1` / `true`
584/// (case-insensitive); anything else (including unset) keeps the
585/// default `spawn_blocking` behaviour. See issue #431.
586pub(super) fn env_inline_vm() -> bool {
587    match std::env::var("LEX_NET_INLINE_VM") {
588        Ok(v) => {
589            let s = v.trim().to_ascii_lowercase();
590            s == "1" || s == "true"
591        }
592        Err(_) => false,
593    }
594}
595
596/// Server-config record threaded through `serve_http_plain` / `_fn` /
597/// `_routed`. Built from env vars on the legacy `net.serve*` paths
598/// (`ServeOpts::from_env`) or decoded from a user-supplied Lex record
599/// literal on the new `net.serve*_with` paths (`decode_serve_opts`).
600/// See lex-lang#497 for the design rationale.
601#[derive(Debug, Clone)]
602pub(crate) struct ServeOpts {
603    pub(crate) http2: bool,
604    pub(crate) inline_vm: bool,
605    pub(crate) host: String,
606}
607
608impl ServeOpts {
609    /// Default values that match the legacy behaviour with env vars
610    /// honoured. Use this when entering via `net.serve`, `net.serve_fn`,
611    /// or `net.serve_routed` — preserves backwards compatibility.
612    pub(super) fn from_env() -> Self {
613        Self {
614            http2: env_http2(),
615            inline_vm: env_inline_vm(),
616            host: "0.0.0.0".to_string(),
617        }
618    }
619
620    /// Hard-coded defaults returned by `net.default_opts()`. Does NOT
621    /// consult env vars — the `*_with` paths read the opts record
622    /// literally, so the env-var escape hatch only applies to legacy
623    /// callers (`net.serve` et al).
624    pub(super) fn lex_defaults() -> Self {
625        Self {
626            http2: false,
627            inline_vm: false,
628            host: "0.0.0.0".to_string(),
629        }
630    }
631
632    /// Convert to a Lex `Value::Record` for return from `default_opts()`.
633    pub(super) fn to_value(&self) -> Value {
634        let mut rec = indexmap::IndexMap::new();
635        rec.insert("http2".to_string(),     Value::Bool(self.http2));
636        rec.insert("inline_vm".to_string(), Value::Bool(self.inline_vm));
637        rec.insert("host".to_string(),      Value::Str(self.host.clone().into()));
638        Value::record_dynamic(rec)
639    }
640}
641
642/// Decode a `ServeOpts` from a Lex record literal. Fields are
643/// required — the type-checker has already verified the shape, so
644/// here we just project them out. Any deviation from the expected
645/// shape is treated as an internal-consistency error.
646pub(super) fn decode_serve_opts(v: &Value) -> Result<ServeOpts, String> {
647    let rec = match v {
648        Value::Record { fields: r, .. } => r,
649        other => return Err(format!("opts must be a Record, got {other:?}")),
650    };
651    let http2 = match rec.get("http2") {
652        Some(Value::Bool(b)) => *b,
653        _ => return Err("opts.http2 must be Bool".into()),
654    };
655    let inline_vm = match rec.get("inline_vm") {
656        Some(Value::Bool(b)) => *b,
657        _ => return Err("opts.inline_vm must be Bool".into()),
658    };
659    let host = match rec.get("host") {
660        Some(Value::Str(s)) => s.to_string(),
661        _ => return Err("opts.host must be Str".into()),
662    };
663    Ok(ServeOpts { http2, inline_vm, host })
664}
665
666// ── tls.* and net.serve_quic* dispatch helpers (#496) ──────────────
667//
668// `TlsConfig` is opaque in the type system (a `Ty::Con("TlsConfig",…)`)
669// but at runtime it's a `Value::Record({cert :: Bytes, key :: Bytes})`
670// carrying the PEM-encoded chain + private key. The opacity matters
671// because we may switch the in-runtime representation to a Resource
672// handle later (e.g. to keep the private key out of GC-visible
673// memory) without breaking source code.
674
675pub(super) fn make_tls_config_value(cert_pem: Vec<u8>, key_pem: Vec<u8>) -> Value {
676    let mut rec = indexmap::IndexMap::new();
677    rec.insert("cert".into(), Value::Bytes(cert_pem));
678    rec.insert("key".into(),  Value::Bytes(key_pem));
679    Value::record_dynamic(rec)
680}
681
682#[cfg(feature = "quic")]
683pub(super) fn decode_tls_config(v: &Value) -> Result<crate::quic::QuicTls, String> {
684    let rec = match v {
685        Value::Record { fields: r, .. } => r,
686        other => return Err(format!("TlsConfig: expected Record, got {other:?}")),
687    };
688    let cert = match rec.get("cert") {
689        Some(Value::Bytes(b)) => b.to_vec(),
690        _ => return Err("TlsConfig.cert: must be Bytes".into()),
691    };
692    let key = match rec.get("key") {
693        Some(Value::Bytes(b)) => b.to_vec(),
694        _ => return Err("TlsConfig.key: must be Bytes".into()),
695    };
696    Ok(crate::quic::QuicTls { cert_pem: cert, key_pem: key })
697}
698
699pub(super) fn dispatch_tls_from_pem_files(
700    handler: &DefaultHandler,
701    args: Vec<Value>,
702) -> Result<Value, String> {
703    let cert_path = expect_str(args.first())?.to_string();
704    let key_path  = expect_str(args.get(1))?.to_string();
705    let cert_resolved = handler.resolve_read_path(&cert_path);
706    let key_resolved  = handler.resolve_read_path(&key_path);
707    if !handler.policy.allow_fs_read.is_empty() {
708        let allowed = |p: &std::path::Path| -> bool {
709            handler.policy.allow_fs_read.iter().any(|a| p.starts_with(a))
710        };
711        if !allowed(&cert_resolved) {
712            return Ok(err(Value::Str(
713                format!("tls.from_pem_files: cert `{cert_path}` outside --allow-fs-read").into(),
714            )));
715        }
716        if !allowed(&key_resolved) {
717            return Ok(err(Value::Str(
718                format!("tls.from_pem_files: key `{key_path}` outside --allow-fs-read").into(),
719            )));
720        }
721    }
722    let cert = match std::fs::read(&cert_resolved) {
723        Ok(b) => b,
724        Err(e) => return Ok(err(Value::Str(format!("read cert {cert_path}: {e}").into()))),
725    };
726    let key = match std::fs::read(&key_resolved) {
727        Ok(b) => b,
728        Err(e) => return Ok(err(Value::Str(format!("read key {key_path}: {e}").into()))),
729    };
730    Ok(ok(make_tls_config_value(cert, key)))
731}
732
733#[cfg(feature = "quic")]
734pub(super) fn dispatch_tls_self_signed(args: Vec<Value>) -> Result<Value, String> {
735    let hostname = expect_str(args.first())?.to_string();
736    match crate::quic::self_signed_pem(&hostname) {
737        Ok((cert, key)) => Ok(ok(make_tls_config_value(cert, key))),
738        Err(e) => Ok(err(Value::Str(format!("tls.self_signed: {e}").into()))),
739    }
740}
741
742#[cfg(not(feature = "quic"))]
743pub(super) fn dispatch_tls_self_signed(_args: Vec<Value>) -> Result<Value, String> {
744    Ok(err(Value::Str(
745        "tls.self_signed: lex-runtime was compiled without the `quic` feature (needed for rcgen)".into(),
746    )))
747}
748
749impl DefaultHandler {
750    #[cfg(feature = "quic")]
751    pub(super) fn dispatch_serve_quic_named(&self, args: Vec<Value>) -> Result<Value, String> {
752        let port = match args.first() {
753            Some(Value::Int(n)) if (0..=65535).contains(n) => *n as u16,
754            _ => return Err("net.serve_quic(port, tls, handler): port must be Int 0..=65535".into()),
755        };
756        let tls = decode_tls_config(args.get(1)
757            .ok_or_else(|| "net.serve_quic(port, tls, handler): missing tls".to_string())?)?;
758        let handler_name = expect_str(args.get(2))?.to_string();
759        let program = self.program.clone()
760            .ok_or_else(|| "net.serve_quic requires a Program reference; use DefaultHandler::with_program".to_string())?;
761        let policy = self.policy.clone();
762        crate::quic::serve_http3_named(port, handler_name, tls, program, policy, ServeOpts::from_env())
763    }
764
765    #[cfg(feature = "quic")]
766    pub(super) fn dispatch_serve_quic_fn(&self, args: Vec<Value>) -> Result<Value, String> {
767        let port = match args.first() {
768            Some(Value::Int(n)) if (0..=65535).contains(n) => *n as u16,
769            _ => return Err("net.serve_quic_fn(port, tls, handler): port must be Int 0..=65535".into()),
770        };
771        let tls = decode_tls_config(args.get(1)
772            .ok_or_else(|| "net.serve_quic_fn(port, tls, handler): missing tls".to_string())?)?;
773        let closure = match args.into_iter().nth(2) {
774            Some(c @ Value::Closure { .. }) => c,
775            _ => return Err("net.serve_quic_fn(port, tls, handler): handler must be a closure".into()),
776        };
777        let program = self.program.clone()
778            .ok_or_else(|| "net.serve_quic_fn requires a Program reference; use DefaultHandler::with_program".to_string())?;
779        let policy = self.policy.clone();
780        crate::quic::serve_http3_fn(port, closure, tls, program, policy, ServeOpts::from_env())
781    }
782
783    #[cfg(feature = "quic")]
784    pub(super) fn dispatch_serve_quic_routed(&self, args: Vec<Value>) -> Result<Value, String> {
785        let port = match args.first() {
786            Some(Value::Int(n)) if (0..=65535).contains(n) => *n as u16,
787            _ => return Err("net.serve_quic_routed(port, tls, routes, fallback): port must be Int 0..=65535".into()),
788        };
789        let tls = decode_tls_config(args.get(1)
790            .ok_or_else(|| "net.serve_quic_routed(port, tls, routes, fallback): missing tls".to_string())?)?;
791        let routes_val = args.get(2).cloned()
792            .ok_or_else(|| "net.serve_quic_routed(port, tls, routes, fallback): missing routes".to_string())?;
793        let fallback = match args.into_iter().nth(3) {
794            Some(c @ Value::Closure { .. }) => c,
795            _ => return Err("net.serve_quic_routed(port, tls, routes, fallback): fallback must be a closure".into()),
796        };
797        let routes = decode_routes_arg(routes_val)?;
798        let program = self.program.clone()
799            .ok_or_else(|| "net.serve_quic_routed requires a Program reference; use DefaultHandler::with_program".to_string())?;
800        let policy = self.policy.clone();
801        crate::quic::serve_http3_routed(port, routes, fallback, tls, program, policy, ServeOpts::from_env())
802    }
803
804    #[cfg(not(feature = "quic"))]
805    pub(super) fn dispatch_serve_quic_named(&self, _args: Vec<Value>) -> Result<Value, String> {
806        Err("net.serve_quic: lex-runtime was compiled without the `quic` feature (needed for quinn + h3)".into())
807    }
808    #[cfg(not(feature = "quic"))]
809    pub(super) fn dispatch_serve_quic_fn(&self, _args: Vec<Value>) -> Result<Value, String> {
810        Err("net.serve_quic_fn: lex-runtime was compiled without the `quic` feature (needed for quinn + h3)".into())
811    }
812    #[cfg(not(feature = "quic"))]
813    pub(super) fn dispatch_serve_quic_routed(&self, _args: Vec<Value>) -> Result<Value, String> {
814        Err("net.serve_quic_routed: lex-runtime was compiled without the `quic` feature (needed for quinn + h3)".into())
815    }
816}
817
818/// Read `LEX_NET_HTTP2` and report whether the runtime should accept
819/// HTTP/2 connections via hyper-util's auto builder (HTTP/1 ↔ HTTP/2
820/// preface detection). Accepts `1` / `true` (case-insensitive); anything
821/// else (including unset) keeps the HTTP/1-only default.
822///
823/// h2c (cleartext HTTP/2) needs prior-knowledge clients
824/// (`curl --http2-prior-knowledge`, wrk/h2load, gRPC). Browsers do not
825/// speak h2c — they require ALPN over TLS, which is a separate path.
826/// See lex-lang#488.
827pub(super) fn env_http2() -> bool {
828    match std::env::var("LEX_NET_HTTP2") {
829        Ok(v) => {
830            let s = v.trim().to_ascii_lowercase();
831            s == "1" || s == "true"
832        }
833        Err(_) => false,
834    }
835}
836
837/// Build a Lex request record from hyper request parts and pre-collected body bytes.
838pub(crate) fn build_request_value_parts(
839    parts: &hyper::http::request::Parts,
840    body: &bytes::Bytes,
841) -> Value {
842    let method = parts.method.as_str().to_string();
843    // `Uri::path()` returns just the origin-form path, regardless of
844    // whether the wire URI was relative (`/foo` — HTTP/1.1) or
845    // absolute (`https://host/foo` — HTTP/2 and HTTP/3 fold the
846    // `:scheme` + `:authority` pseudo-headers into the full URI).
847    // Reading `to_string()` would leak the scheme/authority into the
848    // Lex handler's `req.path`, which surprised handlers built for
849    // HTTP/1.1 (#496 surfaced this against `serve_quic`).
850    let path = parts.uri.path().to_string();
851    let query = parts.uri.query().map(str::to_string).unwrap_or_default();
852    let mut headers_map = std::collections::BTreeMap::new();
853    for (name, val) in &parts.headers {
854        if let Ok(v) = val.to_str() {
855            headers_map.insert(
856                lex_bytecode::MapKey::Str(name.as_str().to_ascii_lowercase()),
857                Value::Str(v.to_string().into()),
858            );
859        }
860    }
861    let body_str = String::from_utf8_lossy(body).into_owned();
862    let mut rec = indexmap::IndexMap::new();
863    rec.insert("method".into(), Value::Str(method.into()));
864    rec.insert("path".into(), Value::Str(path.into()));
865    rec.insert("query".into(), Value::Str(query.into()));
866    rec.insert("body".into(), Value::Str(body_str.into()));
867    rec.insert("headers".into(), Value::Map(headers_map));
868    rec.insert("path_params".into(), Value::Map(std::collections::BTreeMap::new()));
869    Value::record_dynamic(rec)
870}
871
872/// Build a Lex request record from a tiny_http request (used by the TLS path).
873pub(super) fn build_request_value_tiny(req: &mut tiny_http::Request) -> Value {
874    let method = format!("{:?}", req.method()).to_uppercase();
875    let url = req.url().to_string();
876    let (path, query) = match url.split_once('?') {
877        Some((p, q)) => (p.to_string(), q.to_string()),
878        None => (url, String::new()),
879    };
880    let mut headers_map = std::collections::BTreeMap::new();
881    for h in req.headers() {
882        headers_map.insert(
883            lex_bytecode::MapKey::Str(h.field.as_str().as_str().to_ascii_lowercase()),
884            Value::Str(h.value.as_str().to_string().into()),
885        );
886    }
887    let mut body = String::new();
888    let _ = req.as_reader().read_to_string(&mut body);
889    let mut rec = indexmap::IndexMap::new();
890    rec.insert("method".into(), Value::Str(method.into()));
891    rec.insert("path".into(), Value::Str(path.into()));
892    rec.insert("query".into(), Value::Str(query.into()));
893    rec.insert("body".into(), Value::Str(body.into()));
894    rec.insert("headers".into(), Value::Map(headers_map));
895    rec.insert("path_params".into(), Value::Map(std::collections::BTreeMap::new()));
896    Value::record_dynamic(rec)
897}
898
899pub(crate) fn unpack_response(vm: &mut Vm, v: &Value) -> UnpackedResponse {
900    // Accept both heap `Record` and arena `ArenaRecord` — the new
901    // slab-direct accessors below read each uniformly without
902    // requiring a tree-wide materialize first. See
903    // `docs/design/arena-plumbing.md` § "Status update (2026-06-05)"
904    // for the wire-up rationale.
905    if !matches!(v, Value::Record { .. } | Value::ArenaRecord { .. }) {
906        return (
907            500,
908            ResponseBodyOut::Str(format!("handler returned non-record: {v:?}")),
909            vec![],
910        );
911    }
912
913    let status = vm.get_record_field(v, "status").and_then(|s| match s {
914        Value::Int(n) => Some(n as u16),
915        _ => None,
916    }).unwrap_or(200);
917
918    // Body — read once, drain lazy iters inline so the VM is still
919    // in scope when `materialize_lazy_iter` runs. Replaces the
920    // previously-separate `materialize_response_body` pass.
921    let body = match vm.get_record_field(v, "body") {
922        Some(Value::Variant { name, mut args }) if args.len() == 1 => {
923            let inner = args.pop().unwrap();
924            match (name.as_str(), inner) {
925                // Tagged ResponseBody (#375): BodyStr | BodyStream | BodyBytes.
926                ("BodyStr", Value::Str(s)) => ResponseBodyOut::Str(s.to_string()),
927                ("BodyStream", iter_v) => {
928                    let drained = materialize_lazy_iter(vm, iter_v);
929                    ResponseBodyOut::TextChunks(drain_iter_str(&drained))
930                }
931                ("BodyBytes", iter_v) => {
932                    let drained = materialize_lazy_iter(vm, iter_v);
933                    ResponseBodyOut::BytesChunks(drain_iter_bytes(&drained))
934                }
935                _ => ResponseBodyOut::Str(String::new()),
936            }
937        }
938        // Escape hatch for handlers that don't use the nominal
939        // `Response` alias and just return a structural record with
940        // `body :: Str` (the pre-#375 contract). Lets internal
941        // test handlers and one-liners keep working without
942        // wrapping in `BodyStr(...)`.
943        Some(Value::Str(s)) => ResponseBodyOut::Str(s.to_string()),
944        _ => ResponseBodyOut::Str(String::new()),
945    };
946
947    let headers: Vec<(String, String)> = match vm.get_record_field(v, "headers") {
948        Some(Value::Map(hmap)) => hmap.iter().filter_map(|(k, val)| {
949            if let (lex_bytecode::MapKey::Str(name), Value::Str(s)) = (k, val) {
950                Some((name.clone(), s.to_string()))
951            } else {
952                None
953            }
954        }).collect(),
955        _ => vec![],
956    };
957
958    (status, body, headers)
959}
960
961pub(super) type HyperRespBody =
962    http_body_util::combinators::BoxBody<bytes::Bytes, std::convert::Infallible>;
963
964/// Build a hyper response from the unpacked handler tuple
965/// `(status, body, headers)`. The `unpack_response` step runs inside
966/// the spawn_blocking closure (where `vm` is still alive) so this
967/// function doesn't need `&Vm` — arena handles, lazy iters, and the
968/// like are already resolved by the time we get here. Streaming
969/// bodies (`BodyStream`, `BodyBytes`) use `ChunkedBody` which has no
970/// known `size_hint`, so hyper emits `Transfer-Encoding: chunked` on
971/// the wire. Plain string bodies use `Full<Bytes>` which carries
972/// `Content-Length`.
973pub(super) fn build_hyper_response(
974    (status, body, headers): UnpackedResponse,
975) -> hyper::Response<HyperRespBody> {
976    use http_body_util::BodyExt as _;
977    let boxed_body: HyperRespBody = match body {
978        ResponseBodyOut::Str(s) => {
979            http_body_util::Full::new(bytes::Bytes::from(s.into_bytes())).boxed()
980        }
981        ResponseBodyOut::TextChunks(chunks) | ResponseBodyOut::BytesChunks(chunks) => {
982            HyperChunkedBody::from(chunks).boxed()
983        }
984    };
985    let mut builder = hyper::Response::builder().status(status);
986    for (name, val) in headers {
987        builder = builder.header(name, val);
988    }
989    builder
990        .body(boxed_body)
991        .unwrap_or_else(|_| error_response(500, "response build error"))
992}
993
994pub(super) fn error_response(status: u16, msg: &str) -> hyper::Response<HyperRespBody> {
995    use http_body_util::BodyExt as _;
996    hyper::Response::builder()
997        .status(status)
998        .body(
999            http_body_util::Full::new(bytes::Bytes::from(msg.to_owned()))
1000                .boxed(),
1001        )
1002        .unwrap_or_else(|_| {
1003            use http_body_util::BodyExt as _;
1004            hyper::Response::new(http_body_util::Empty::new().map_err(|e| match e {}).boxed())
1005        })
1006}
1007
1008/// Async body that emits pre-collected chunks as separate HTTP frames, causing
1009/// hyper to use `Transfer-Encoding: chunked` (no `size_hint` exact count).
1010pub(super) struct HyperChunkedBody {
1011    pub(super) chunks: std::collections::VecDeque<Vec<u8>>,
1012}
1013
1014impl From<Vec<Vec<u8>>> for HyperChunkedBody {
1015    fn from(chunks: Vec<Vec<u8>>) -> Self {
1016        Self {
1017            chunks: chunks.into_iter().filter(|c| !c.is_empty()).collect(),
1018        }
1019    }
1020}
1021
1022impl hyper::body::Body for HyperChunkedBody {
1023    type Data = bytes::Bytes;
1024    type Error = std::convert::Infallible;
1025
1026    fn poll_frame(
1027        mut self: std::pin::Pin<&mut Self>,
1028        _cx: &mut std::task::Context<'_>,
1029    ) -> std::task::Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
1030        match self.chunks.pop_front() {
1031            Some(chunk) => std::task::Poll::Ready(Some(Ok(hyper::body::Frame::data(
1032                bytes::Bytes::from(chunk),
1033            )))),
1034            None => std::task::Poll::Ready(None),
1035        }
1036    }
1037}
1038
1039/// Send `body` back on a TLS `tiny_http` request. Used only by the
1040/// `net.serve_tls` path which still runs on tiny_http pending a
1041/// tokio-rustls migration.
1042pub(super) fn respond_with_body_tls(
1043    req: tiny_http::Request,
1044    status: u16,
1045    body: ResponseBodyOut,
1046    headers: Vec<(String, String)>,
1047) {
1048    let tiny_headers: Vec<tiny_http::Header> = headers
1049        .into_iter()
1050        .filter_map(|(name, val)| format!("{name}: {val}").parse::<tiny_http::Header>().ok())
1051        .collect();
1052    match body {
1053        ResponseBodyOut::Str(s) => {
1054            let mut response = tiny_http::Response::from_string(s).with_status_code(status);
1055            for h in tiny_headers {
1056                response.add_header(h);
1057            }
1058            let _ = req.respond(response);
1059        }
1060        ResponseBodyOut::TextChunks(chunks) | ResponseBodyOut::BytesChunks(chunks) => {
1061            let reader = ChunkReader::new(chunks);
1062            let response = tiny_http::Response::new(
1063                tiny_http::StatusCode(status),
1064                tiny_headers,
1065                reader,
1066                None,
1067                None,
1068            );
1069            let _ = req.respond(response);
1070        }
1071    }
1072}
1073
1074/// Decoded `Response.body` (#375). The runtime emits each variant via a
1075/// different `tiny_http` path: a single `Response::from_string` for
1076/// `Str`, and a chunked-encoding `Response::new` with a `Read`-backed
1077/// chunk list for the streaming variants.
1078///
1079/// The shape `unpack_response` returns: `(status_code, body, headers)`.
1080/// Factored out as a `type` alias so call sites that store it (the
1081/// spawn_blocking closures' `Result<UnpackedResponse, ...>`) don't
1082/// trip clippy's `type_complexity` lint.
1083pub(crate) type UnpackedResponse = (u16, ResponseBodyOut, Vec<(String, String)>);
1084
1085pub(crate) enum ResponseBodyOut {
1086    Str(String),
1087    /// Pre-drained text chunks. v1 ships eager-iter only; lazy producers
1088    /// (#376 follow-up) will replace this with a Read adapter that pulls
1089    /// chunks on demand from the VM.
1090    TextChunks(Vec<Vec<u8>>),
1091    /// Pre-drained binary chunks. Each inner `Vec<u8>` is one Lex
1092    /// `List[Int]` collapsed down to a byte vector.
1093    BytesChunks(Vec<Vec<u8>>),
1094}
1095
1096/// Walk a Lex `Iter[Str]` (eager (List, Int) representation) and produce
1097/// a chunk list. The chunks are byte vectors so the chunked-Read adapter
1098/// is uniform across text and binary streams.
1099///
1100/// Iter[T] representation shifted in #376: from `Tuple([list, idx])` to
1101/// `Variant("__IterEager", [list, idx])` for the eager form. Lazy iters
1102/// produced by `iter.unfold` (`Variant("__IterLazy", [seed, step])`) and
1103/// cursor-backed iters (`Variant("__IterCursor", [handle])` from #379)
1104/// are not drained eagerly here — the v1 streaming path covers only the
1105/// eager form. Lazy/cursor producers will be wired through the
1106/// `ChunkReader` in a follow-up so each `read()` calls `iter.next` via
1107/// the VM, preserving wall-clock chunk boundaries on the wire.
1108pub(super) fn drain_iter_str(v: &Value) -> Vec<Vec<u8>> {
1109    match v {
1110        Value::Variant { name, args }
1111            if name == "__IterEager" && args.len() == 2 =>
1112        {
1113            if let (Value::List(items), Value::Int(idx)) = (&args[0], &args[1]) {
1114                items.iter().skip(*idx as usize).filter_map(|item| {
1115                    if let Value::Str(s) = item { Some(s.as_bytes().to_vec()) } else { None }
1116                }).collect()
1117            } else {
1118                Vec::new()
1119            }
1120        }
1121        _ => Vec::new(),
1122    }
1123}
1124
1125/// Walk a Lex `Iter[List[Int]]` and produce a chunk list. Each `List[Int]`
1126/// element is collapsed by truncating each Int to u8 (0..=255). See
1127/// `drain_iter_str` for the lazy/cursor-iter limitation.
1128pub(super) fn drain_iter_bytes(v: &Value) -> Vec<Vec<u8>> {
1129    match v {
1130        Value::Variant { name, args }
1131            if name == "__IterEager" && args.len() == 2 =>
1132        {
1133            if let (Value::List(items), Value::Int(idx)) = (&args[0], &args[1]) {
1134                items.iter().skip(*idx as usize).filter_map(|item| {
1135                    if let Value::List(ints) = item {
1136                        Some(ints.iter().filter_map(|i| match i {
1137                            Value::Int(n) => Some((*n & 0xff) as u8),
1138                            _ => None,
1139                        }).collect::<Vec<u8>>())
1140                    } else {
1141                        None
1142                    }
1143                }).collect()
1144            } else {
1145                Vec::new()
1146            }
1147        }
1148        _ => Vec::new(),
1149    }
1150}
1151
1152/// Drive an `__IterLazy(seed, step)` to exhaustion by invoking the step
1153/// closure via `vm`, then return an equivalent `__IterEager(list, 0)` so
1154/// the existing `drain_iter_*` paths can consume it.
1155///
1156/// Without this pre-pass, `BodyStream(iter.unfold(...))` produces empty
1157/// response bodies because the drain helpers match only on the eager
1158/// variant (#477). The step closure can carry effects; we ignore that
1159/// here — the handler is already running on a tokio task with the same
1160/// effect bindings, so any `[net]` / `[time]` calls inside the step
1161/// re-enter the same handler context.
1162///
1163/// `__IterEager` is returned untouched. Unknown variants pass through.
1164pub(super) fn materialize_lazy_iter(vm: &mut Vm, v: Value) -> Value {
1165    let mut current = v;
1166    let mut items: Vec<Value> = Vec::new();
1167    loop {
1168        match current {
1169            Value::Variant { name, args } if name == "__IterLazy" && args.len() == 2 => {
1170                let seed = args[0].clone();
1171                let step = args[1].clone();
1172                match vm.invoke_closure_value(step.clone(), vec![seed]) {
1173                    Ok(Value::Variant { name: opt, args: opt_args })
1174                        if opt == "None" =>
1175                    {
1176                        let _ = opt_args;
1177                        break;
1178                    }
1179                    Ok(Value::Variant { name: opt, args: opt_args })
1180                        if opt == "Some" && opt_args.len() == 1 =>
1181                    {
1182                        if let Value::Tuple(pair) = &opt_args[0] {
1183                            if pair.len() == 2 {
1184                                items.push(pair[0].clone());
1185                                current = Value::Variant {
1186                                    name: "__IterLazy".to_string(),
1187                                    args: vec![pair[1].clone(), step],
1188                                };
1189                                continue;
1190                            }
1191                        }
1192                        // Malformed pair — bail to avoid infinite loop.
1193                        break;
1194                    }
1195                    _ => break,
1196                }
1197            }
1198            // Already eager (or unknown) — return as-is, possibly with
1199            // any items we collected from a partial drain.
1200            other => {
1201                if items.is_empty() {
1202                    return other;
1203                }
1204                // Mixed shape shouldn't happen in practice; fall through
1205                // to the eager builder below with the items we have.
1206                let _ = other;
1207                break;
1208            }
1209        }
1210    }
1211    Value::Variant {
1212        name: "__IterEager".to_string(),
1213        args: vec![
1214            Value::List(items.into_iter().collect()),
1215            Value::Int(0),
1216        ],
1217    }
1218}
1219
1220
1221/// `Read` adapter that returns one Lex chunk per `read()` call so
1222/// `tiny_http`'s chunked transfer-encoding emits each Lex chunk as a
1223/// distinct HTTP chunk on the wire. When the requested buffer is smaller
1224/// than the current chunk we serve a slice and keep the remainder for
1225/// the next call.
1226pub(super) struct ChunkReader {
1227    pub(super) chunks: std::collections::VecDeque<Vec<u8>>,
1228    pub(super) cursor: usize,
1229}
1230
1231impl ChunkReader {
1232    pub(super) fn new(chunks: Vec<Vec<u8>>) -> Self {
1233        Self {
1234            chunks: chunks.into_iter().filter(|c| !c.is_empty()).collect(),
1235            cursor: 0,
1236        }
1237    }
1238}
1239
1240impl std::io::Read for ChunkReader {
1241    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
1242        loop {
1243            let Some(front) = self.chunks.front() else {
1244                return Ok(0);
1245            };
1246            let remaining = &front[self.cursor..];
1247            if remaining.is_empty() {
1248                self.chunks.pop_front();
1249                self.cursor = 0;
1250                continue;
1251            }
1252            let n = remaining.len().min(buf.len());
1253            buf[..n].copy_from_slice(&remaining[..n]);
1254            self.cursor += n;
1255            if self.cursor >= front.len() {
1256                self.chunks.pop_front();
1257                self.cursor = 0;
1258            }
1259            return Ok(n);
1260        }
1261    }
1262}
1263
1264/// #463 slab-direct wire-up — locally-runnable coverage for
1265/// `unpack_response`'s arena path. The lex-runtime integration tests
1266/// (`tests/std_http.rs` etc.) overflow the dev-container disk per
1267/// `arena-plumbing.md`, so CI is the only place they run end-to-end;
1268/// these focused tests give us a local regression gate on the
1269/// boundary code itself.
1270#[cfg(test)]
1271mod unpack_response_tests {
1272    use super::*;
1273    use std::sync::Arc;
1274    use indexmap::IndexMap;
1275    use lex_bytecode::{Const, Op, Program, Value};
1276    use lex_bytecode::program::{Function, ZERO_BODY_HASH};
1277    use lex_bytecode::vm::Vm;
1278
1279    /// Build a single-fn `Program` whose body produces an
1280    /// `AllocArenaRecord`-backed `Response { status, body }`. The
1281    /// constants table holds the field names, the body variant name,
1282    /// the response text, and the status code.
1283    fn build_arena_response_program() -> Arc<Program> {
1284        let constants = vec![
1285            Const::FieldName("status".into()), // 0
1286            Const::FieldName("body".into()),   // 1
1287            Const::Int(200),                   // 2
1288            Const::VariantName("BodyStr".into()), // 3
1289            Const::Str("hello".into()),        // 4
1290        ];
1291        let mut function_names = IndexMap::new();
1292        function_names.insert("handler".to_string(), 0);
1293        Arc::new(Program {
1294            constants,
1295            functions: vec![Function {
1296                name: "handler".into(),
1297                arity: 0,
1298                locals_count: 0,
1299                code: vec![
1300                    Op::PushConst(2),                                       // 200
1301                    Op::PushConst(4),                                       // "hello"
1302                    Op::MakeVariant { name_idx: 3, arity: 1 },              // BodyStr("hello")
1303                    Op::AllocArenaRecord { shape_idx: 0, field_count: 2 },  // { status, body }
1304                    Op::Return,
1305                ],
1306                effects: vec![],
1307                body_hash: ZERO_BODY_HASH,
1308                refinements: vec![],
1309                field_ic_sites: 0,
1310            }],
1311            function_names,
1312            module_aliases: IndexMap::new(),
1313            entry: Some(0),
1314            record_shapes: vec![vec![0, 1]], // {status, body}
1315        })
1316    }
1317
1318    /// The happy path: arena handle goes in, the unpacked tuple comes
1319    /// out, no `materialize_arena_handles` walk in between. The
1320    /// boundary call site no longer holds a heap `Value::Record` —
1321    /// `unpack_response` reads straight out of the slab via
1322    /// `Vm::get_record_field`.
1323    #[test]
1324    fn unpack_response_reads_arena_record_via_slab() {
1325        let p = build_arena_response_program();
1326        let mut vm = Vm::new(&p);
1327        let scope = vm.enter_request_scope();
1328
1329        let resp = vm.invoke(p.function_names["handler"], vec![]).unwrap();
1330        // Test precondition — without this the slab-direct path isn't
1331        // being exercised at all.
1332        assert!(matches!(resp, Value::ArenaRecord { .. }),
1333            "expected ArenaRecord (slab path), got {resp:?}");
1334
1335        let (status, body, headers) = unpack_response(&mut vm, &resp);
1336        vm.exit_request_scope(scope);
1337
1338        assert_eq!(status, 200);
1339        assert!(headers.is_empty());
1340        match body {
1341            ResponseBodyOut::Str(s) => assert_eq!(s, "hello"),
1342            _ => panic!("expected BodyStr"),
1343        }
1344    }
1345
1346    /// Heap path uniformity: a handler that returns a plain
1347    /// `Value::Record` (no arena scope, or a non-arena-lowered site)
1348    /// produces the same tuple. The same `unpack_response` is the
1349    /// single chokepoint.
1350    #[test]
1351    fn unpack_response_reads_heap_record() {
1352        let p = build_arena_response_program();
1353        let mut vm = Vm::new(&p);
1354
1355        // No scope — `AllocArenaRecord` falls back to heap `MakeRecord`.
1356        let resp = vm.invoke(p.function_names["handler"], vec![]).unwrap();
1357        assert!(matches!(resp, Value::Record { .. }),
1358            "expected heap Record (fallback path), got {resp:?}");
1359
1360        let (status, body, headers) = unpack_response(&mut vm, &resp);
1361        assert_eq!(status, 200);
1362        assert!(headers.is_empty());
1363        match body {
1364            ResponseBodyOut::Str(s) => assert_eq!(s, "hello"),
1365            _ => panic!("expected BodyStr"),
1366        }
1367    }
1368
1369    /// Defaults: handler returns a non-record. The error path produces
1370    /// a 500 with a diagnostic. Unchanged from pre-wire-up behavior.
1371    #[test]
1372    fn unpack_response_falls_back_to_500_on_non_record() {
1373        let p = build_arena_response_program();
1374        let mut vm = Vm::new(&p);
1375        let v = Value::Int(7);
1376        let (status, _body, _headers) = unpack_response(&mut vm, &v);
1377        assert_eq!(status, 500);
1378    }
1379}