diff --git a/bin/el/src/main.rs b/bin/el/src/main.rs index 304afb4..b1b5395 100644 --- a/bin/el/src/main.rs +++ b/bin/el/src/main.rs @@ -27,8 +27,335 @@ //! el seal seal an existing release artifact //! el unseal unseal a sealed artifact +mod cache; +mod net; +mod telemetry; + use std::path::PathBuf; +// ── App block context (parsed from `app "name" { ... }` at startup) ─────────── + +thread_local! { + /// Resolved environment: "dev" | "stage" | "prod" (from NEURON_ENV, default "dev") + static APP_ENV: std::cell::RefCell = std::cell::RefCell::new(String::new()); + /// App name from `app "name" { ... }` + static APP_SERVICE: std::cell::RefCell = std::cell::RefCell::new(String::new()); + /// App version from `version "x.y.z"` + static APP_VERSION: std::cell::RefCell = std::cell::RefCell::new(String::new()); + /// Stable UUID for this process lifetime (generated once) + static APP_INSTANCE: std::cell::RefCell = std::cell::RefCell::new(String::new()); + /// Resolved config key → value (after env overlay) + static APP_CONFIG: std::cell::RefCell> = + std::cell::RefCell::new(std::collections::HashMap::new()); + /// Resolved secrets key → value + static APP_SECRETS: std::cell::RefCell> = + std::cell::RefCell::new(std::collections::HashMap::new()); + /// Feature flags key → Bool value + static APP_FLAGS: std::cell::RefCell> = + std::cell::RefCell::new(std::collections::HashMap::new()); + /// Set of secret values (for redaction in traces/logs) + static SECRET_VALUES: std::cell::RefCell> = + std::cell::RefCell::new(std::collections::HashSet::new()); +} + +// ── App block parser ────────────────────────────────────────────────────────── + +/// Parsed representation of an `app` block found in El source. +#[derive(Default, Debug)] +struct AppBlock { + name: String, + version: String, + /// Base config values: key → (value_str, type_hint) + config_base: std::collections::HashMap, + /// Env overlay values: env_name → (key → value_str) + config_overlays: std::collections::HashMap>, + /// Required secrets: key → type_hint + secrets: Vec<(String, String)>, + /// Feature flags: key → default_value + flags: std::collections::HashMap, +} + +/// Minimal tokenizer that splits El source into tokens, ignoring comments. +fn tokenize_simple(src: &str) -> Vec { + let mut tokens = Vec::new(); + let mut chars = src.chars().peekable(); + while let Some(&c) = chars.peek() { + if c.is_whitespace() { chars.next(); continue; } + // Line comment + if c == '/' { + chars.next(); + if chars.peek() == Some(&'/') { + // consume to end of line + while let Some(&nc) = chars.peek() { chars.next(); if nc == '\n' { break; } } + continue; + } + tokens.push("/".to_string()); + continue; + } + // String literal + if c == '"' { + chars.next(); + let mut s = String::new(); + while let Some(&nc) = chars.peek() { + chars.next(); + if nc == '"' { break; } + if nc == '\\' { if let Some(&esc) = chars.peek() { chars.next(); match esc { 'n'=>s.push('\n'), 't'=>s.push('\t'), _=>s.push(esc) } } } + else { s.push(nc); } + } + tokens.push(format!("\"{}\"", s)); + continue; + } + // Braces, colons, equals, semicolons + if "{}:;=".contains(c) { chars.next(); tokens.push(c.to_string()); continue; } + // Identifier or number + if c.is_alphanumeric() || c == '_' || c == '-' || c == '.' { + let mut w = String::new(); + while let Some(&nc) = chars.peek() { + if nc.is_alphanumeric() || nc == '_' || nc == '-' || nc == '.' { w.push(nc); chars.next(); } + else { break; } + } + tokens.push(w); + continue; + } + chars.next(); + } + tokens +} + +/// Extract and parse the first `app "name" { ... }` block from El source. +/// Returns None if no app block is present. +fn parse_app_block(source: &str) -> Option { + let tokens = tokenize_simple(source); + let mut i = 0usize; + // Find `app` + while i < tokens.len() { + if tokens[i] == "app" && i + 1 < tokens.len() && tokens[i+1].starts_with('"') { + break; + } + i += 1; + } + if i >= tokens.len() { return None; } + // tokens[i] = "app", tokens[i+1] = "\"name\"", tokens[i+2] = "{" + let raw_name = &tokens[i + 1]; + let name = raw_name.trim_matches('"').to_string(); + i += 2; + if i >= tokens.len() || tokens[i] != "{" { return None; } + i += 1; + + let mut block = AppBlock { name, ..Default::default() }; + let mut depth = 1usize; + + while i < tokens.len() && depth > 0 { + let tok = tokens[i].clone(); + match tok.as_str() { + "{" => { depth += 1; i += 1; } + "}" => { depth -= 1; i += 1; } + "version" if depth == 1 => { + i += 1; + if i < tokens.len() && tokens[i].starts_with('"') { + block.version = tokens[i].trim_matches('"').to_string(); + i += 1; + } + } + "config" if depth == 1 => { + i += 1; + if i < tokens.len() && tokens[i] == "{" { i += 1; } + // Parse config entries until matching } + let mut cfg_depth = 1usize; + while i < tokens.len() && cfg_depth > 0 { + let ct = tokens[i].clone(); + match ct.as_str() { + "{" => { cfg_depth += 1; i += 1; } + "}" => { cfg_depth -= 1; i += 1; } + _ => { + // Could be: KEY: Type = "value" or env_name { ... } + // Peek ahead to determine + if i + 1 < tokens.len() && tokens[i+1] == "{" { + // env overlay block e.g. `prod { KEY = "val" ... }` + let env_name = ct.clone(); + i += 2; // skip name and { + let mut env_entries: std::collections::HashMap = Default::default(); + while i < tokens.len() && tokens[i] != "}" { + // KEY = "value" + let key = tokens[i].clone(); + i += 1; + if i < tokens.len() && tokens[i] == "=" { i += 1; } + let val = if i < tokens.len() { + tokens[i].trim_matches('"').to_string() + } else { String::new() }; + i += 1; + env_entries.insert(key, val); + } + if i < tokens.len() { i += 1; } // skip } + block.config_overlays.insert(env_name, env_entries); + } else { + // KEY: Type = "default" OR KEY: Type (no default) + let key = ct.clone(); + i += 1; + let mut type_hint = "String".to_string(); + let mut default_val = String::new(); + if i < tokens.len() && tokens[i] == ":" { + i += 1; + if i < tokens.len() { type_hint = tokens[i].clone(); i += 1; } + } + if i < tokens.len() && tokens[i] == "=" { + i += 1; + if i < tokens.len() { default_val = tokens[i].trim_matches('"').to_string(); i += 1; } + } + block.config_base.insert(key, (default_val, type_hint)); + } + } + } + } + } + "secrets" if depth == 1 => { + i += 1; + if i < tokens.len() && tokens[i] == "{" { i += 1; } + while i < tokens.len() && tokens[i] != "}" { + let key = tokens[i].clone(); + i += 1; + let mut type_hint = "String".to_string(); + if i < tokens.len() && tokens[i] == ":" { + i += 1; + if i < tokens.len() { type_hint = tokens[i].clone(); i += 1; } + } + block.secrets.push((key, type_hint)); + } + if i < tokens.len() { i += 1; } // skip } + } + "flags" if depth == 1 => { + i += 1; + if i < tokens.len() && tokens[i] == "{" { i += 1; } + while i < tokens.len() && tokens[i] != "}" { + let key = tokens[i].clone(); + i += 1; + if i < tokens.len() && tokens[i] == ":" { + i += 1; + if i < tokens.len() { i += 1; } // skip type hint (Bool) + } + let mut default_val = false; + if i < tokens.len() && tokens[i] == "=" { + i += 1; + if i < tokens.len() { + default_val = tokens[i] == "true"; + i += 1; + } + } + block.flags.insert(key, default_val); + } + if i < tokens.len() { i += 1; } // skip } + } + _ => { i += 1; } + } + } + Some(block) +} + +/// Load secrets.json from ~/.neuron/secrets.json, return key→value map. +fn load_neuron_secrets_json() -> std::collections::HashMap { + let home = std::env::var("HOME").unwrap_or_default(); + let path = format!("{home}/.neuron/secrets.json"); + if let Ok(content) = std::fs::read_to_string(&path) { + if let Ok(serde_json::Value::Object(map)) = serde_json::from_str::(&content) { + return map.into_iter() + .filter_map(|(k, v)| v.as_str().map(|s| (k, s.to_string()))) + .collect(); + } + } + std::collections::HashMap::new() +} + +/// Resolve an app block against the current environment and load everything into thread-locals. +/// Returns Err with a descriptive message if a required secret is missing. +fn apply_app_block(block: AppBlock) -> Result<(), String> { + use el_compiler::Value; + + let env = std::env::var("NEURON_ENV").unwrap_or_else(|_| "dev".to_string()); + let instance = uuid::Uuid::new_v4().to_string(); + + // Resolve config: env var > env overlay > base default + let overlay = block.config_overlays.get(&env).cloned().unwrap_or_default(); + let mut config_map: std::collections::HashMap = std::collections::HashMap::new(); + for (key, (base_default, type_hint)) in &block.config_base { + // Check env var override first + let raw = std::env::var(key) + .ok() + .or_else(|| overlay.get(key).cloned()) + .unwrap_or_else(|| base_default.clone()); + let value = match type_hint.as_str() { + "Int" => raw.trim().parse::().map(Value::Int).unwrap_or(Value::Str(raw.clone())), + "Bool" => Value::Bool(raw.trim() == "true" || raw.trim() == "1"), + _ => Value::Str(raw.clone()), + }; + config_map.insert(key.clone(), value); + } + + // Resolve secrets + let secrets_file = load_neuron_secrets_json(); + let mut secrets_map: std::collections::HashMap = std::collections::HashMap::new(); + let mut secret_vals: std::collections::HashSet = std::collections::HashSet::new(); + for (key, _type_hint) in &block.secrets { + let val = std::env::var(key) + .ok() + .or_else(|| secrets_file.get(key).cloned()); + match val { + Some(v) => { + if !v.is_empty() { secret_vals.insert(v.clone()); } + secrets_map.insert(key.clone(), v); + } + None => { + return Err(format!( + "Error: required secret '{}' not found. Set environment variable or add to ~/.neuron/secrets.json", + key + )); + } + } + } + + // Store everything in thread-locals + APP_ENV.with(|e| *e.borrow_mut() = env); + APP_SERVICE.with(|s| *s.borrow_mut() = block.name); + APP_VERSION.with(|v| *v.borrow_mut() = block.version); + APP_INSTANCE.with(|id| *id.borrow_mut() = instance); + APP_CONFIG.with(|c| *c.borrow_mut() = config_map); + APP_SECRETS.with(|s| *s.borrow_mut() = secrets_map); + APP_FLAGS.with(|f| *f.borrow_mut() = block.flags); + SECRET_VALUES.with(|sv| *sv.borrow_mut() = secret_vals); + + Ok(()) +} + +/// Redact any known secret values from a string (best-effort). +#[allow(dead_code)] +fn redact_secrets(s: &str) -> String { + SECRET_VALUES.with(|sv| { + let sv = sv.borrow(); + let mut result = s.to_string(); + for secret in sv.iter() { + if !secret.is_empty() && result.contains(secret.as_str()) { + result = result.replace(secret.as_str(), "[REDACTED]"); + } + } + result + }) +} + +/// Try to parse and apply app block from source. If parsing succeeds, apply context. +/// Prints error and exits on secret resolution failure. +fn maybe_apply_app_block(source: &str) { + if let Some(block) = parse_app_block(source) { + if !block.name.is_empty() { + // Update telemetry service name from app block + telemetry::set_service_name(&block.name.clone()); + telemetry::init(&block.name); + if let Err(e) = apply_app_block(block) { + eprintln!("{e}"); + std::process::exit(1); + } + } + } +} + use clap::{Parser, Subcommand}; use el_build::BuildSystem; use el_compiler::{Compiler, CompilerOptions, Target}; @@ -120,6 +447,21 @@ fn next_net_id(prefix: &str) -> String { }) } +// ── Telemetry — per-interpreter function span stack ─────────────────────────── + +thread_local! { + /// Active user-function spans. Each entry: (fn_name, start_ns, trace_id, span_id, parent_span_id) + static FN_SPAN_STACK: std::cell::RefCell)>> = + std::cell::RefCell::new(Vec::new()); + + /// Explicit span registry for trace_start / trace_end / trace_tag builtins. + /// Key: span_id → (trace_id, span_id, start_ns, name, attrs) + static EXPLICIT_SPANS: std::cell::RefCell) + >> = std::cell::RefCell::new(std::collections::HashMap::new()); +} + // ── CLI definition ──────────────────────────────────────────────────────────── #[derive(Parser, Debug)] @@ -325,6 +667,41 @@ enum Command { #[arg(long, short = 'o')] output: Option, }, + + // ── El VM (bytecode compile + execute) ──────────────────────────────────── + + /// Compile a .el source file to an El bytecode (.elc) container. + /// + /// Produces a binary .elc file with the ELVM magic header, suitable for + /// distribution and execution via `el exec` or the `elvm` standalone binary. + /// + /// Examples: + /// el compile main.el + /// el compile main.el -o dist/ + Compile { + /// Source file to compile (*.el). + file: PathBuf, + /// Output path: a directory (places .elc inside) or a full file path. + /// Defaults to .elc in the same directory as the source file. + #[arg(long, short = 'o')] + output: Option, + }, + + /// Execute a compiled El bytecode file (.elc) on the El VM. + /// + /// The file must have been produced by `el compile` or `el build-file`. + /// Both ELVM-container and legacy JSON-only .elc files are accepted. + /// + /// Examples: + /// el exec main.elc + /// el exec dist/myapp.elc + Exec { + /// Compiled bytecode file (*.elc). + file: PathBuf, + /// Arguments to forward to the program (available via `args()`). + #[arg(trailing_var_arg = true)] + args: Vec, + }, } #[derive(Subcommand, Debug)] @@ -410,10 +787,32 @@ async fn run(cli: Cli) -> Result<(), Box> { Command::Run { manifest } => { let manifest_path = resolve_manifest(manifest.as_deref())?; let bs = BuildSystem::from_manifest_file(&manifest_path)?; - let out = bs.build(Some(BuildTarget::Debug)).await?; - let artifact = std::fs::read(&out.artifact_path)?; - let instructions = el_compiler::Bytecode::deserialize_all(&artifact) + // Initialise telemetry with the package name as service name. + let svc_name = std::env::var("OTEL_SERVICE_NAME") + .unwrap_or_else(|_| bs.manifest.package.name.clone()); + telemetry::set_service_name(&svc_name); + telemetry::init(&svc_name); + + // Resolve entry source (with imports). + let entry_path = manifest_path + .parent() + .unwrap_or(std::path::Path::new(".")) + .join(&bs.manifest.build.entry); + let entry_source = resolve_imports(&entry_path) + .map_err(|e| format!("cannot resolve imports for {}: {e}", entry_path.display()))?; + + // Parse and apply `app` block before running. + maybe_apply_app_block(&entry_source); + + // Compile via Rust compiler. + let opts = CompilerOptions { + target: Target::Debug, + ..Default::default() + }; + let compiled = Compiler::compile(&entry_source, opts) + .map_err(|e| format!("compile error: {e}"))?; + let instructions = el_compiler::Bytecode::deserialize_all(&compiled.artifact) .unwrap_or_default(); // block_in_place lets the interpreter use reqwest::blocking from within tokio::main tokio::task::block_in_place(|| run_interpreter(&instructions)); @@ -582,9 +981,21 @@ async fn run(cli: Cli) -> Result<(), Box> { // ── Low-level / single-file ─────────────────────────────────────────── Command::RunFile { file, args } => { + // Initialise telemetry — service name from file stem or OTEL_SERVICE_NAME. + let svc_name = std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| { + file.file_stem() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|| "el-program".to_string()) + }); + telemetry::set_service_name(&svc_name); + telemetry::init(&svc_name); + let source = resolve_imports(&file) .map_err(|e| format!("cannot resolve imports for {}: {e}", file.display()))?; + // Parse and apply `app` block before compilation so builtins are ready. + maybe_apply_app_block(&source); + let opts = CompilerOptions { target: Target::Debug, source_path: file.clone(), @@ -679,10 +1090,216 @@ async fn run(cli: Cli) -> Result<(), Box> { println!("unsealed {} -> {} ({} bytes)", artifact.display(), out_path.display(), plaintext.len()); } + + // ── El VM: compile ──────────────────────────────────────────────────── + + Command::Compile { file, output } => { + // Resolve imports into a single pre-resolved source string. + let source = resolve_imports(&file) + .map_err(|e| format!("cannot resolve imports for {}: {e}", file.display()))?; + + // Compile via the El self-compiler running on elvm. + let elvm_bytes = compile_via_el_self_compiler(&source, &file)?; + + // Determine output path. + let out_path = resolve_compile_output(&file, output.as_deref())?; + + std::fs::write(&out_path, &elvm_bytes) + .map_err(|e| format!("cannot write {}: {e}", out_path.display()))?; + + println!( + "compile: {} -> {} ({} bytes, ELVM v{})", + file.display(), + out_path.display(), + elvm_bytes.len(), + el_compiler::ELVM_VERSION, + ); + } + + // ── El VM: exec ─────────────────────────────────────────────────────── + + Command::Exec { file, args } => { + // Initialise telemetry from the file stem. + let svc_name = std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| { + file.file_stem() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|| "el-program".to_string()) + }); + telemetry::set_service_name(&svc_name); + telemetry::init(&svc_name); + + let bytes = std::fs::read(&file) + .map_err(|e| format!("cannot read {}: {e}", file.display()))?; + + let instructions = el_compiler::Bytecode::deserialize_all(&bytes) + .map_err(|e| format!("cannot load bytecode from {}: {e}", file.display()))?; + + // block_in_place lets the interpreter use reqwest::blocking from within tokio::main + tokio::task::block_in_place(|| run_interpreter_with_args(&instructions, &args)); + } } Ok(()) } +/// Resolve the output path for `el compile`. +/// +/// - If `output` is a directory → place `.elc` inside it. +/// - If `output` is a full file path → use it directly. +/// - If `output` is None → use `.elc` next to the source file. +fn resolve_compile_output( + source: &std::path::Path, + output: Option<&std::path::Path>, +) -> Result> { + let stem = source.file_stem().unwrap_or_default().to_string_lossy(); + let elc_name = format!("{stem}.elc"); + + match output { + None => { + // Same directory as source. + let dir = source.parent().unwrap_or(std::path::Path::new(".")); + Ok(dir.join(elc_name)) + } + Some(p) => { + if p.is_dir() || p.to_string_lossy().ends_with('/') { + Ok(p.join(elc_name)) + } else { + Ok(p.to_path_buf()) + } + } + } +} + +// ── El self-compiler (elvm-based) ───────────────────────────────────────────── + +/// Find the `elvm` binary. +/// +/// Search order: +/// 1. EL_ELVM_PATH env var (override for development) +/// 2. Same directory as the running `el` binary +/// 3. PATH +fn find_elvm_binary() -> Option { + // 1. Explicit override + if let Ok(p) = std::env::var("EL_ELVM_PATH") { + let path = PathBuf::from(p); + if path.exists() { return Some(path); } + } + + // 2. Sibling to current executable + if let Ok(exe) = std::env::current_exe() { + if let Some(dir) = exe.parent() { + let sibling = dir.join("elvm"); + if sibling.exists() { return Some(sibling); } + } + } + + // 3. PATH + which_binary("elvm") +} + +/// Find the `el-compiler.elc` bootstrap bytecode. +/// +/// Search order: +/// 1. EL_COMPILER_ELC env var (override for development) +/// 2. Same directory as the running `el` binary +/// 3. `/../lib/el/el-compiler.elc` (installed layout) +fn find_el_compiler_elc() -> Option { + // 1. Explicit override + if let Ok(p) = std::env::var("EL_COMPILER_ELC") { + let path = PathBuf::from(p); + if path.exists() { return Some(path); } + } + + // 2. Sibling to current executable + if let Ok(exe) = std::env::current_exe() { + if let Some(dir) = exe.parent() { + let sibling = dir.join("el-compiler.elc"); + if sibling.exists() { return Some(sibling); } + } + } + + // 3. Installed layout: /lib/el/el-compiler.elc + if let Ok(exe) = std::env::current_exe() { + if let Some(dir) = exe.parent() { + if let Some(prefix) = dir.parent() { + let installed = prefix.join("lib").join("el").join("el-compiler.elc"); + if installed.exists() { return Some(installed); } + } + } + } + + None +} + +/// Find a binary on PATH. +fn which_binary(name: &str) -> Option { + let path_var = std::env::var("PATH").unwrap_or_default(); + for dir in path_var.split(':') { + let candidate = PathBuf::from(dir).join(name); + if candidate.exists() { + return Some(candidate); + } + } + None +} + +/// Compile El source via the El self-compiler running on elvm. +/// +/// Writes pre-resolved source to a temp file, invokes: +/// `elvm ` +/// then reads the JSON bytecode output and wraps it in an ELVM binary container. +/// +/// Returns the raw `.elc` bytes (ELVM header + JSON payload). +fn compile_via_el_self_compiler( + source: &str, + source_path: &std::path::Path, +) -> Result, Box> { + let elvm = find_elvm_binary() + .ok_or("cannot find elvm binary (set EL_ELVM_PATH or place elvm next to el)")?; + let compiler_elc = find_el_compiler_elc() + .ok_or("cannot find el-compiler.elc (set EL_COMPILER_ELC or place el-compiler.elc next to el)")?; + + // Write pre-resolved source to a temp file. + let tmp_dir = std::env::temp_dir(); + let stem = source_path.file_stem().unwrap_or_default().to_string_lossy(); + let tmp_src = tmp_dir.join(format!("el-compile-{stem}-{}.el", std::process::id())); + let tmp_out = tmp_dir.join(format!("el-compile-{stem}-{}.json", std::process::id())); + + std::fs::write(&tmp_src, source) + .map_err(|e| format!("cannot write temp source: {e}"))?; + + // Invoke: elvm + let status = std::process::Command::new(&elvm) + .arg(&compiler_elc) + .arg(&tmp_src) + .arg(&tmp_out) + .status() + .map_err(|e| format!("cannot spawn elvm: {e}"))?; + + // Clean up temp source regardless of outcome. + let _ = std::fs::remove_file(&tmp_src); + + if !status.success() { + let _ = std::fs::remove_file(&tmp_out); + return Err(format!( + "el-compiler exited with status {}", + status.code().unwrap_or(-1) + ).into()); + } + + // Read JSON bytecode output. + let json_bytes = std::fs::read(&tmp_out) + .map_err(|e| format!("cannot read compiler output {}: {e}", tmp_out.display()))?; + let _ = std::fs::remove_file(&tmp_out); + + // Parse JSON and wrap in ELVM binary container. + let instructions = el_compiler::Bytecode::deserialize_all(&json_bytes) + .map_err(|e| format!("invalid bytecode from El compiler: {e}"))?; + let elvm_bytes = el_compiler::wrap_elvm(&instructions) + .map_err(|e| format!("ELVM wrap failed: {e}"))?; + + Ok(elvm_bytes) +} + // ── Command implementations ─────────────────────────────────────────────────── fn cmd_new(name: &str, dir: Option<&std::path::Path>) -> Result<(), Box> { @@ -696,28 +1313,24 @@ fn cmd_new(name: &str, dir: Option<&std::path::Path>) -> Result<(), Box Void {{ std::fs::write(project_dir.join(".gitignore"), "dist/\n.el/\n")?; println!("created project '{name}' in {}/", project_dir.display()); - println!(" el.toml"); + println!(" manifest.el"); println!(" src/main.el"); Ok(()) } @@ -859,8 +1472,9 @@ fn resolve_imports_inner( .map_err(|e| format!("cannot read {}: {e}", file.display()))?; let mut out = String::new(); + let mut lines_iter = source.lines().peekable(); - for line in source.lines() { + while let Some(line) = lines_iter.next() { let trimmed = line.trim(); if let Some(rest) = trimmed.strip_prefix("import ") { // import "filename.el" @@ -875,6 +1489,42 @@ fn resolve_imports_inner( out.push_str(line); out.push('\n'); } + } else if let Some(rest) = trimmed.strip_prefix("from ") { + // from ModuleName import { ... } — resolve ModuleName.el in the same dir + if let Some(module_part) = rest.split(" import").next() { + let module_name = module_part.trim(); + if !module_name.contains('"') && !module_name.contains('/') && !module_name.is_empty() { + let module_file = format!("{}.el", module_name); + let import_path = dir.join(&module_file); + + // Consume any continuation lines of a multi-line import: + // `from X import {\n Y,\n Z,\n}` + let import_rest = rest.splitn(2, " import").nth(1).unwrap_or("").trim(); + let is_multiline = import_rest.contains('{') && !import_rest.contains('}'); + if is_multiline { + for cont in lines_iter.by_ref() { + if cont.trim().contains('}') { + break; + } + } + } + + if import_path.exists() { + let imported = resolve_imports_inner(&import_path, visited)?; + out.push_str(&imported); + out.push('\n'); + } else { + out.push_str(line); + out.push('\n'); + } + } else { + out.push_str(line); + out.push('\n'); + } + } else { + out.push_str(line); + out.push('\n'); + } } else { out.push_str(line); out.push('\n'); @@ -884,7 +1534,7 @@ fn resolve_imports_inner( Ok(out) } -/// Find the nearest `el.toml` starting from the current directory. +/// Find the nearest `manifest.el` starting from the current directory. fn resolve_manifest(path: Option<&std::path::Path>) -> Result> { if let Some(p) = path { return Ok(p.to_path_buf()); @@ -1235,18 +1885,42 @@ fn run_sub_interpreter_with_stack( stack.push(v); } Bytecode::Call { name, arity } => { + eprintln!("[DEBUG sub] Call {name} arity={arity} stack_top={:?}", stack.last()); let result = dispatch_builtin(name, *arity, &mut stack, &program_args); + eprintln!("[DEBUG sub] dispatch_builtin({name}) -> {:?} stack_top={:?}", matches!(result, BuiltinResult::Handled), stack.last()); match result { BuiltinResult::Handled | BuiltinResult::HttpServe => {} BuiltinResult::Exit(code) => std::process::exit(code), BuiltinResult::NotBuiltin => { - if let Some(&entry) = fn_table.get(name.as_str()) { - let saved = locals.clone(); - call_stack.push((ip + 1, saved)); - ip = entry; - continue; + eprintln!("[DEBUG sub] NotBuiltin for {name}, fn_table has: {:?}", fn_table.keys().collect::>()); + // HOF dispatch in sub-interpreter + match name.as_str() { + "map" | "list_map" => { + let fn_ref = stack.pop().unwrap_or(Value::Nil); + let list = match stack.pop().unwrap_or(Value::List(vec![])) { + Value::List(l) => l, + _ => vec![], + }; + let fn_name = match &fn_ref { Value::Str(s) => s.clone(), _ => String::new() }; + let mut result = Vec::new(); + if let Some(&entry) = fn_table.get(&fn_name) { + for item in list { + let mapped = run_sub_interpreter_with_stack(instructions, fn_table, entry, vec![item]); + result.push(mapped); + } + } + stack.push(Value::List(result)); + } + _ => { + if let Some(&entry) = fn_table.get(name.as_str()) { + let saved = locals.clone(); + call_stack.push((ip + 1, saved)); + ip = entry; + continue; + } + stack.push(Value::Nil); + } } - stack.push(Value::Nil); } } } @@ -1631,21 +2305,26 @@ fn run_interpreter_with_args(instructions: &[el_compiler::Bytecode], program_arg BuiltinResult::NotBuiltin => { // Handle HOFs that need fn_table access match name.as_str() { - "list_map" => { - let fn_name = match stack.pop().unwrap_or(Value::Nil) { - Value::Str(s) => s, - _ => String::new(), - }; + // `list_map` or `.map(fn)` method on List + "list_map" | "map" => { + let fn_ref = stack.pop().unwrap_or(Value::Nil); let list = match stack.pop().unwrap_or(Value::List(vec![])) { Value::List(l) => l, _ => vec![], }; + let fn_name = match &fn_ref { + Value::Str(s) => s.clone(), + _ => String::new(), + }; let mut result = Vec::new(); if let Some(&entry) = fn_table.get(&fn_name) { for item in list { let mapped = run_sub_interpreter_with_stack(instructions, &fn_table, entry, vec![item]); result.push(mapped); } + } else { + // fn_ref not found — just return empty + let _ = fn_ref; } stack.push(Value::List(result)); } @@ -1697,6 +2376,20 @@ fn run_interpreter_with_args(instructions: &[el_compiler::Bytecode], program_arg _ => { // Try user-defined function if let Some(&entry) = fn_table.get(name.as_str()) { + // Open a function span for automatic tracing. + let (_fn_trace_id, _fn_parent_id) = + telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + let _fn_span_id = telemetry::new_span_id(); + let _fn_start_ns = telemetry::now_ns(); + FN_SPAN_STACK.with(|s| s.borrow_mut().push(( + name.to_string(), + _fn_start_ns, + _fn_trace_id, + _fn_span_id, + _fn_parent_id, + ))); // Save current locals and return address let saved = locals.clone(); call_stack.push((ip + 1, saved)); @@ -1807,6 +2500,27 @@ fn run_interpreter_with_args(instructions: &[el_compiler::Bytecode], program_arg } } Bytecode::Return => { + // Close automatic function span (if any) before restoring frame. + FN_SPAN_STACK.with(|s| { + if let Some((fn_name, start_ns, trace_id, span_id, parent_id)) = s.borrow_mut().pop() { + let end_ns = telemetry::now_ns(); + let latency_ms = ((end_ns.saturating_sub(start_ns)) / 1_000_000) as i64; + telemetry::emit_span(telemetry::Span { + trace_id, + span_id, + parent_id, + name: fn_name, + start_ns, + end_ns, + attrs: vec![ + ("fn.latency_ms".to_string(), telemetry::AttrValue::Int(latency_ms)), + ], + events: vec![], + status: telemetry::SpanStatus::Ok, + service: telemetry::service_name().to_string(), + }); + } + }); // Return from a user function call if let Some((ret_ip, saved_locals)) = call_stack.pop() { locals = saved_locals; @@ -1895,6 +2609,68 @@ fn run_interpreter_with_args(instructions: &[el_compiler::Bytecode], program_arg } ip += 1; } + + // ── Component app bootstrap ─────────────────────────────────────────────── + // If the program defined an `App` component (registered as __fn_App in fn_table), + // call it and render the result as the app UI. + if let Some(&app_entry) = fn_table.get("App").or_else(|| fn_table.get("__component_App")) { + eprintln!("[DEBUG] Found App at entry={app_entry}, fn_table keys: {:?}", fn_table.keys().collect::>()); + // Dump instructions around app_entry for debugging + for i in 0..std::cmp::min(instructions.len(), app_entry + 30) { + eprintln!("[DEBUG bytecode] [{i}] {:?}", instructions[i]); + } + let html = run_sub_interpreter(instructions, &fn_table, app_entry); + eprintln!("[DEBUG] run_sub_interpreter returned: {:?}", html); + render_component_to_terminal(&html); + } else { + eprintln!("[DEBUG] No App found. fn_table keys: {:?}", fn_table.keys().collect::>()); + } +} + +/// Render a component value to the terminal. +/// If the value is an HTML string (from __jsx__), pretty-print it. +/// Otherwise just print it. +fn render_component_to_terminal(val: &el_compiler::Value) { + use el_compiler::Value; + match val { + Value::Str(html) => { + if html.starts_with('<') { + // Strip HTML tags and print text content for terminal output + let text = strip_html_tags(html); + println!("{text}"); + } else if !html.is_empty() { + println!("{html}"); + } + } + Value::Nil => {} + other => println!("{other}"), + } +} + +/// Strip HTML tags from a string, preserving text content. +fn strip_html_tags(html: &str) -> String { + let mut result = String::new(); + let mut in_tag = false; + let mut depth = 0; + + for ch in html.chars() { + match ch { + '<' => { in_tag = true; depth += 1; } + '>' => { + depth -= 1; + if depth == 0 { in_tag = false; } + } + _ if !in_tag => result.push(ch), + _ => {} + } + } + + // Clean up whitespace + result.lines() + .map(|l| l.trim()) + .filter(|l| !l.is_empty()) + .collect::>() + .join("\n") } /// Call soma AI inference endpoint and return response text. @@ -2025,14 +2801,39 @@ fn dispatch_builtin( Value::Str(s) => s, _ => String::new(), }; - let result = reqwest::blocking::Client::new() - .post(&url) - .header("Content-Type", "application/json") - .header("X-NC-CLI", "true") - .body(body) - .send() - .and_then(|r| r.text()) - .unwrap_or_else(|e| format!("{{\"error\":\"{e}\"}}")); + let t0 = telemetry::now_ms(); + let body_bytes = body.len() as i64; + // Automatic single retry on transient connection/timeout errors. + let result = net::with_single_retry(500, || { + reqwest::blocking::Client::new() + .post(&url) + .header("Content-Type", "application/json") + .header("X-NC-CLI", "true") + .body(body.clone()) + .send() + .and_then(|r| r.text()) + }); + // Emit http.client POST span. + let latency = (telemetry::now_ms() - t0) as i64; + let (trace_id, parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + telemetry::emit_span(telemetry::Span { + trace_id, span_id: telemetry::new_span_id(), parent_id, + name: "http.client POST".to_string(), + start_ns: (t0 as u64).saturating_sub(latency as u64) * 1_000_000, + end_ns: t0 as u64 * 1_000_000, + status: if result.starts_with("{\"error\"") { + telemetry::SpanStatus::Error("http error".to_string()) + } else { telemetry::SpanStatus::Ok }, + attrs: vec![ + ("http.method".to_string(), telemetry::AttrValue::Str("POST".to_string())), + ("http.url".to_string(), telemetry::AttrValue::Str(url.clone())), + ("http.latency_ms".to_string(), telemetry::AttrValue::Int(latency)), + ("http.request_body_bytes".to_string(), telemetry::AttrValue::Int(body_bytes)), + ], + events: Vec::new(), service: telemetry::service_name().to_string(), + }); stack.push(Value::Str(result)); BuiltinResult::Handled } @@ -2041,9 +2842,40 @@ fn dispatch_builtin( Value::Str(s) => s, _ => String::new(), }; - let result = reqwest::blocking::get(&url) - .and_then(|r| r.text()) - .unwrap_or_else(|e| format!("{{\"error\":\"{e}\"}}")); + // Check in-memory response cache first (default TTL: 60 s). + if let Some(cached) = cache::cache_get(&url) { + stack.push(Value::Str(cached)); + return BuiltinResult::Handled; + } + let t0 = telemetry::now_ms(); + // Automatic single retry on transient connection/timeout errors. + let result = net::with_single_retry(500, || { + reqwest::blocking::get(&url).and_then(|r| r.text()) + }); + // Emit http.client GET span. + let latency = (telemetry::now_ms() - t0) as i64; + let (trace_id, parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + telemetry::emit_span(telemetry::Span { + trace_id, span_id: telemetry::new_span_id(), parent_id, + name: "http.client GET".to_string(), + start_ns: (t0 as u64).saturating_sub(latency as u64) * 1_000_000, + end_ns: t0 as u64 * 1_000_000, + status: if result.starts_with("{\"error\"") { + telemetry::SpanStatus::Error("http error".to_string()) + } else { telemetry::SpanStatus::Ok }, + attrs: vec![ + ("http.method".to_string(), telemetry::AttrValue::Str("GET".to_string())), + ("http.url".to_string(), telemetry::AttrValue::Str(url.clone())), + ("http.latency_ms".to_string(), telemetry::AttrValue::Int(latency)), + ], + events: Vec::new(), service: telemetry::service_name().to_string(), + }); + // Cache only successful (non-error) responses. + if !result.starts_with("{\"error\"") { + cache::cache_set(url, result.clone(), 60); + } stack.push(Value::Str(result)); BuiltinResult::Handled } @@ -2054,7 +2886,7 @@ fn dispatch_builtin( }; BuiltinResult::Exit(code) } - "str_contains" => { + "str_contains" | "native_string_contains" => { let needle = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new(), @@ -2093,7 +2925,7 @@ fn dispatch_builtin( stack.push(Value::List(parts)); BuiltinResult::Handled } - "list_len" => { + "list_len" | "native_list_len" => { let list = match stack.pop().unwrap_or(Value::Nil) { Value::List(l) => l, _ => vec![], @@ -2101,7 +2933,7 @@ fn dispatch_builtin( stack.push(Value::Int(list.len() as i64)); BuiltinResult::Handled } - "list_get" => { + "list_get" | "native_list_get" => { let idx = match stack.pop().unwrap_or(Value::Nil) { Value::Int(n) => n, _ => -1, @@ -2118,7 +2950,17 @@ fn dispatch_builtin( stack.push(v); BuiltinResult::Handled } - "int_to_str" => { + "native_string_chars" => { + // Split a string into a list of single-character strings. + let s = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let chars: Vec = s.chars().map(|c| Value::Str(c.to_string())).collect(); + stack.push(Value::List(chars)); + BuiltinResult::Handled + } + "int_to_str" | "native_int_to_str" => { let n = match stack.pop().unwrap_or(Value::Nil) { Value::Int(n) => n, _ => 0, @@ -2190,10 +3032,31 @@ fn dispatch_builtin( Value::Str(s) => s, _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } }; - let result = std::fs::read_to_string(&path) - .map(Value::Str) - .unwrap_or(Value::Nil); - stack.push(result); + let t0 = telemetry::now_ms(); + let read_result = std::fs::read_to_string(&path); + let latency = (telemetry::now_ms() - t0) as i64; + let (trace_id, parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + let bytes_read = read_result.as_ref().map(|s| s.len() as i64).unwrap_or(-1); + let (status, val) = match read_result { + Ok(s) => (telemetry::SpanStatus::Ok, Value::Str(s)), + Err(_) => (telemetry::SpanStatus::Error(format!("fs_read failed: {path}")), Value::Nil), + }; + telemetry::emit_span(telemetry::Span { + trace_id, span_id: telemetry::new_span_id(), parent_id, + name: "fs.read".to_string(), + start_ns: (t0 as u64).saturating_sub(latency as u64) * 1_000_000, + end_ns: t0 as u64 * 1_000_000, + status, + attrs: vec![ + ("fs.path".to_string(), telemetry::AttrValue::Str(path)), + ("fs.bytes_read".to_string(), telemetry::AttrValue::Int(bytes_read)), + ("fs.latency_ms".to_string(), telemetry::AttrValue::Int(latency)), + ], + events: Vec::new(), service: telemetry::service_name().to_string(), + }); + stack.push(val); BuiltinResult::Handled } "fs_write" => { @@ -2205,8 +3068,29 @@ fn dispatch_builtin( Value::Str(s) => s, _ => String::new(), }; - let ok = std::fs::write(&path, &content).is_ok(); - stack.push(Value::Bool(ok)); + let t0 = telemetry::now_ms(); + let bytes_written = content.len() as i64; + let write_ok = std::fs::write(&path, &content).is_ok(); + let latency = (telemetry::now_ms() - t0) as i64; + let (trace_id, parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + telemetry::emit_span(telemetry::Span { + trace_id, span_id: telemetry::new_span_id(), parent_id, + name: "fs.write".to_string(), + start_ns: (t0 as u64).saturating_sub(latency as u64) * 1_000_000, + end_ns: t0 as u64 * 1_000_000, + status: if write_ok { telemetry::SpanStatus::Ok } else { + telemetry::SpanStatus::Error(format!("fs_write failed: {path}")) + }, + attrs: vec![ + ("fs.path".to_string(), telemetry::AttrValue::Str(path)), + ("fs.bytes_written".to_string(), telemetry::AttrValue::Int(bytes_written)), + ("fs.latency_ms".to_string(), telemetry::AttrValue::Int(latency)), + ], + events: Vec::new(), service: telemetry::service_name().to_string(), + }); + stack.push(Value::Bool(write_ok)); BuiltinResult::Handled } "fs_append" => { @@ -2389,6 +3273,9 @@ fn dispatch_builtin( Value::Str(s) => s, _ => String::new(), }; + if !key.starts_with("__") { + telemetry::log_debug(&format!("state_set: key={}", key)); + } GLOBAL_STATE.with(|gs| gs.borrow_mut().insert(key, value)); stack.push(Value::Bool(true)); BuiltinResult::Handled @@ -2497,6 +3384,29 @@ fn dispatch_builtin( continue; } + // GET /about — serve __about_html_file__ if set + if method == "GET" && path == "/about" { + let about_path = GLOBAL_STATE.with(|gs| gs.borrow().get("__about_html_file__").cloned()); + if let Some(p) = about_path { + let html = std::fs::read_to_string(&p) + .unwrap_or_else(|_| r#"{"error":"about page not found"}"#.to_string()); + let is_html = html.trim_start().starts_with("__ keys. GLOBAL_STATE.with(|gs| { @@ -2937,6 +3854,7 @@ fn dispatch_builtin( s.insert("__method__".to_string(), method.clone()); s.insert("__path__".to_string(), path.clone()); s.insert("__request__".to_string(), body.clone()); + s.insert("__request_id__".to_string(), request_id.clone()); s.remove("__response__"); // Store each request header for access via state_get() for header in request.headers() { @@ -2946,6 +3864,9 @@ fn dispatch_builtin( } }); + // Push server span so nested builtins (llm_call, http_post, etc.) parent to it. + telemetry::context::push_span(_req_trace_id.clone(), _req_span_id.clone()); + // Call handle_request via the thread-local fn executor HTTP_SERVE_CALL.with(|f| { if let Some(ref call_fn) = *f.borrow() { @@ -2953,11 +3874,47 @@ fn dispatch_builtin( } }); + // Close the server span and record latency. + telemetry::context::pop_span(); + let _req_end_ns = telemetry::now_ns(); + let _req_latency_ms = (telemetry::now_ms() - _req_t0_ms) as i64; + let response_body = GLOBAL_STATE.with(|gs| { gs.borrow().get("__response__").cloned() .unwrap_or_else(|| r#"{"error":"not found"}"#.to_string()) }); + // Infer HTTP status from response body (best-effort). + let _http_status: i64 = if response_body.contains("\"error\"") { 500 } else { 200 }; + + // Emit http.server span to OTLP. + telemetry::emit_span(telemetry::Span { + trace_id: _req_trace_id.clone(), + span_id: _req_span_id.clone(), + parent_id: None, + name: format!("http.server {} {}", method, path), + start_ns: _req_start_ns, + end_ns: _req_end_ns, + attrs: vec![ + ("http.method".to_string(), telemetry::AttrValue::Str(method.clone())), + ("http.target".to_string(), telemetry::AttrValue::Str(path.clone())), + ("http.status_code".to_string(), telemetry::AttrValue::Int(_http_status)), + ("http.latency_ms".to_string(), telemetry::AttrValue::Int(_req_latency_ms)), + ("request_id".to_string(), telemetry::AttrValue::Str(request_id.clone())), + ], + events: vec![], + status: if _http_status >= 500 { + telemetry::SpanStatus::Error("handler error".to_string()) + } else { telemetry::SpanStatus::Ok }, + service: telemetry::service_name().to_string(), + }); + + // Structured access log: METHOD path → status latency_ms [request_id] + telemetry::log_info(&format!( + "{} {} \u{2192} {} {}ms [{}]", + method, path, _http_status, _req_latency_ms, request_id + )); + let _ = request.respond( tiny_http::Response::from_string(response_body) .with_header(content_json) @@ -3550,7 +4507,7 @@ fn dispatch_builtin( // ── String aliases and extras ───────────────────────────────────────── - "string_len" | "str_len" => { + "string_len" | "str_len" | "native_string_len" => { let s = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new(), @@ -3686,7 +4643,7 @@ fn dispatch_builtin( stack.push(Value::Str(v.to_string())); BuiltinResult::Handled } - "str_to_int" | "parse_int" => { + "str_to_int" | "parse_int" | "native_str_to_int" => { let s = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new(), @@ -4313,6 +5270,36 @@ fn dispatch_builtin( BuiltinResult::Handled } + // ── Form-encoded POST with Basic auth (Stripe, etc.) ───────────────── + + // http_post_form_auth(url, username, body) -> String + // Posts form-encoded body with HTTP Basic auth (username, empty password). + // Used for Stripe API: username = sk_test_... or sk_live_... + "http_post_form_auth" => { + let body = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let username = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let url = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let result = reqwest::blocking::Client::new() + .post(&url) + .basic_auth(&username, Option::<&str>::None) + .header("Content-Type", "application/x-www-form-urlencoded") + .body(body) + .send() + .and_then(|r| r.text()) + .unwrap_or_default(); + stack.push(Value::Str(result)); + BuiltinResult::Handled + } + // ── Engram HTTP helpers (X-Engram-Key auth) ────────────────────────── // http_post_engram(url, api_key, body) -> String @@ -5738,7 +6725,7 @@ fn dispatch_builtin( } // ── List additions ──────────────────────────────────────────────────── - "list_push" => { + "list_push" | "native_list_append" => { let item = stack.pop().unwrap_or(Value::Nil); let list = match stack.pop().unwrap_or(Value::List(vec![])) { Value::List(l) => l, _ => vec![] }; let mut new_list = list; @@ -5767,7 +6754,7 @@ fn dispatch_builtin( let list: Vec = (start..end).map(Value::Int).collect(); stack.push(Value::List(list)); BuiltinResult::Handled } - "list_new" => { stack.push(Value::List(vec![])); BuiltinResult::Handled } + "list_new" | "native_list_empty" => { stack.push(Value::List(vec![])); BuiltinResult::Handled } "list_empty" => { let list = match stack.pop().unwrap_or(Value::List(vec![])) { Value::List(l) => l, _ => vec![] }; stack.push(Value::Bool(list.is_empty())); BuiltinResult::Handled @@ -6150,7 +7137,54 @@ fn dispatch_builtin( Value::Str(s) => s, _ => "claude".to_string(), }; + // Automatic LLM response caching (TTL: 300 s). + // Disabled when NEURON_LLM_CACHE=0. + let llm_cache_enabled = std::env::var("NEURON_LLM_CACHE") + .map(|v| v != "0") + .unwrap_or(true); + let cache_key = format!("llm:{}:{}", model, prompt); + if llm_cache_enabled { + if let Some(cached) = cache::cache_get(&cache_key) { + stack.push(Value::Str(cached)); + return BuiltinResult::Handled; + } + } + let _llm_t0 = telemetry::now_ns(); let result = call_llm_blocking(&model, None, &prompt); + let _llm_end_ns = telemetry::now_ns(); + let _llm_latency_ms = ((_llm_end_ns - _llm_t0) / 1_000_000) as i64; + let _llm_is_error = result.starts_with("{\"error\""); + { + let (trace_id, parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + let mut attrs = vec![ + ("llm.model".to_string(), telemetry::AttrValue::Str(model.clone())), + ("llm.prompt_length".to_string(), telemetry::AttrValue::Int(prompt.len() as i64)), + ("llm.latency_ms".to_string(), telemetry::AttrValue::Int(_llm_latency_ms)), + ]; + if _llm_is_error { attrs.push(("error".to_string(), telemetry::AttrValue::Bool(true))); } + let events = if _llm_is_error { + vec![telemetry::SpanEvent { + name: "exception".to_string(), time_ns: _llm_end_ns, + attrs: vec![("message".to_string(), telemetry::AttrValue::Str(result.clone()))], + }] + } else { vec![] }; + telemetry::emit_span(telemetry::Span { + trace_id, span_id: telemetry::new_span_id(), parent_id, + name: format!("llm.call {}", model), + start_ns: _llm_t0, end_ns: _llm_end_ns, + attrs, events, + status: if _llm_is_error { + telemetry::SpanStatus::Error(result.clone()) + } else { telemetry::SpanStatus::Ok }, + service: telemetry::service_name().to_string(), + }); + } + telemetry::log_info(&format!("llm_call: model={} latency={}ms", model, _llm_latency_ms)); + if llm_cache_enabled && !_llm_is_error { + cache::cache_set(cache_key, result.clone(), 300); + } stack.push(Value::Str(result)); BuiltinResult::Handled } @@ -6169,7 +7203,21 @@ fn dispatch_builtin( Value::Str(s) => s, _ => "claude".to_string(), }; + // Automatic LLM response caching (TTL: 300 s). + let llm_cache_enabled = std::env::var("NEURON_LLM_CACHE") + .map(|v| v != "0") + .unwrap_or(true); + let cache_key = format!("llm:{}:{}:{}", model, system, prompt); + if llm_cache_enabled { + if let Some(cached) = cache::cache_get(&cache_key) { + stack.push(Value::Str(cached)); + return BuiltinResult::Handled; + } + } let result = call_llm_blocking(&model, Some(&system), &prompt); + if llm_cache_enabled && !result.starts_with("{\"error\"") { + cache::cache_set(cache_key, result.clone(), 300); + } stack.push(Value::Str(result)); BuiltinResult::Handled } @@ -6184,7 +7232,30 @@ fn dispatch_builtin( Value::Str(s) => s, _ => "claude".to_string(), }; + let _llms_t0 = telemetry::now_ns(); let result = call_llm_blocking(&model, None, &prompt); + let _llms_end = telemetry::now_ns(); + { + let is_err = result.starts_with("{\"error\""); + let (trace_id, parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + let mut attrs = vec![ + ("llm.model".to_string(), telemetry::AttrValue::Str(model.clone())), + ("llm.prompt_length".to_string(), telemetry::AttrValue::Int(prompt.len() as i64)), + ("llm.latency_ms".to_string(), telemetry::AttrValue::Int(((_llms_end - _llms_t0) / 1_000_000) as i64)), + ]; + if is_err { attrs.push(("error".to_string(), telemetry::AttrValue::Bool(true))); } + telemetry::emit_span(telemetry::Span { + trace_id, span_id: telemetry::new_span_id(), parent_id, + name: format!("llm.stream {}", model), + start_ns: _llms_t0, end_ns: _llms_end, + attrs, events: vec![], + status: if is_err { telemetry::SpanStatus::Error(result.clone()) } else { telemetry::SpanStatus::Ok }, + service: telemetry::service_name().to_string(), + }); + } + telemetry::log_info(&format!("llm_stream: model={} latency={}ms", model, ((_llms_end - _llms_t0) / 1_000_000))); stack.push(Value::Str(result)); BuiltinResult::Handled } @@ -6242,10 +7313,1409 @@ fn dispatch_builtin( BuiltinResult::Handled } + // ── GC / memory management builtins ────────────────────────────────── + + // gc_collect() — force a GC cycle. + // Since the El runtime uses clone-based value semantics (no heap graph), + // this resets the alloc counter and records the collect timestamp. + "gc_collect" => { + GC_ALLOC_COUNT.store(0, std::sync::atomic::Ordering::Relaxed); + GC_LAST_COLLECT_MS.store(now_ms_u64(), std::sync::atomic::Ordering::Relaxed); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // gc_stats() -> String — JSON: {heap_used, heap_total, objects_alive, last_collect_ms} + "gc_stats" => { + let heap_used = estimate_mem_usage_bytes(); + let alloc_count = GC_ALLOC_COUNT.load(std::sync::atomic::Ordering::Relaxed); + let last_collect = GC_LAST_COLLECT_MS.load(std::sync::atomic::Ordering::Relaxed); + let json = serde_json::json!({ + "heap_used": heap_used, + "heap_total": heap_used, // no separate total tracking + "objects_alive": alloc_count, + "last_collect_ms": last_collect + }); + stack.push(Value::Str(json.to_string())); + BuiltinResult::Handled + } + + // gc_set_threshold(bytes: Int) — trigger gc_collect when heap exceeds this. + "gc_set_threshold" => { + let bytes = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n.max(0) as u64, + _ => 0, + }; + GC_THRESHOLD_BYTES.store(bytes, std::sync::atomic::Ordering::Relaxed); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // mem_limit_set(bytes: Int) — set a soft memory limit. + "mem_limit_set" => { + let bytes = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n.max(0) as u64, + _ => 0, + }; + MEM_LIMIT_BYTES.store(bytes, std::sync::atomic::Ordering::Relaxed); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // mem_usage() -> Int — current estimated memory usage in bytes. + "mem_usage" => { + let usage = estimate_mem_usage_bytes(); + // Check threshold and GC if needed + let threshold = GC_THRESHOLD_BYTES.load(std::sync::atomic::Ordering::Relaxed); + if threshold > 0 && usage >= threshold { + GC_ALLOC_COUNT.store(0, std::sync::atomic::Ordering::Relaxed); + GC_LAST_COLLECT_MS.store(now_ms_u64(), std::sync::atomic::Ordering::Relaxed); + } + stack.push(Value::Int(usage as i64)); + BuiltinResult::Handled + } + + // ── Rate limiting builtins ──────────────────────────────────────────── + + // rate_limit_create(name, max_calls, window_ms) + "rate_limit_create" => { + let window_ms = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n.max(1) as u64, + _ => 1000, + }; + let max_calls = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n.max(0) as u64, + _ => 10, + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + { + let mut cfg = rate_limiter_configs().lock().unwrap_or_else(|e| e.into_inner()); + cfg.insert(name.clone(), (max_calls, window_ms)); + } + { + let mut rl = rate_limiters().lock().unwrap_or_else(|e| e.into_inner()); + rl.entry(name).or_insert_with(std::collections::VecDeque::new); + } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // rate_limit_check(name) -> Bool — true if call is allowed. + "rate_limit_check" => { + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let now = now_ms_u64(); + let allowed = { + let cfg = rate_limiter_configs().lock().unwrap_or_else(|e| e.into_inner()); + let (max_calls, window_ms) = cfg.get(&name).cloned().unwrap_or((10, 1000)); + let mut rl = rate_limiters().lock().unwrap_or_else(|e| e.into_inner()); + let timestamps = rl.entry(name).or_insert_with(std::collections::VecDeque::new); + // Evict expired timestamps + timestamps.retain(|&ts| now.saturating_sub(ts) < window_ms); + if (timestamps.len() as u64) < max_calls { + timestamps.push_back(now); + true + } else { + false + } + }; + stack.push(Value::Bool(allowed)); + BuiltinResult::Handled + } + + // rate_limit_wait(name) — block until a slot is available. + "rate_limit_wait" => { + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + loop { + let now = now_ms_u64(); + let (allowed, sleep_ms) = { + let cfg = rate_limiter_configs().lock().unwrap_or_else(|e| e.into_inner()); + let (max_calls, window_ms) = cfg.get(&name).cloned().unwrap_or((10, 1000)); + let mut rl = rate_limiters().lock().unwrap_or_else(|e| e.into_inner()); + let timestamps = rl.entry(name.clone()).or_insert_with(std::collections::VecDeque::new); + timestamps.retain(|&ts| now.saturating_sub(ts) < window_ms); + if (timestamps.len() as u64) < max_calls { + timestamps.push_back(now); + (true, 0u64) + } else { + // Wait until oldest timestamp expires + let oldest = timestamps.front().cloned().unwrap_or(now); + let wait = window_ms.saturating_sub(now.saturating_sub(oldest)).max(1); + (false, wait) + } + }; + if allowed { break; } + std::thread::sleep(std::time::Duration::from_millis(sleep_ms)); + } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ── Backpressure builtins ───────────────────────────────────────────── + + // backpressure_create(name, high_watermark, low_watermark) + "backpressure_create" => { + let low = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n, + _ => 10, + }; + let high = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n, + _ => 100, + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let ctrl = BackpressureController { + high_watermark: high, + low_watermark: low, + state: "ok".to_string(), + }; + backpressure_map().lock().unwrap_or_else(|e| e.into_inner()) + .insert(name, ctrl); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // backpressure_signal(name, current_depth) + // Producer reports current queue depth; updates the state. + "backpressure_signal" => { + let depth = match stack.pop().unwrap_or(Value::Nil) { + Value::Int(n) => n, + _ => 0, + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + { + let mut map = backpressure_map().lock().unwrap_or_else(|e| e.into_inner()); + if let Some(ctrl) = map.get_mut(&name) { + ctrl.state = if depth >= ctrl.high_watermark { + "stop".to_string() + } else if depth > ctrl.low_watermark { + "slow".to_string() + } else { + "ok".to_string() + }; + } + } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // backpressure_check(name) -> String — "ok" | "slow" | "stop" + "backpressure_check" => { + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let state = { + let map = backpressure_map().lock().unwrap_or_else(|e| e.into_inner()); + map.get(&name).map(|c| c.state.clone()).unwrap_or_else(|| "ok".to_string()) + }; + stack.push(Value::Str(state)); + BuiltinResult::Handled + } + + // ── Program lifecycle hooks ─────────────────────────────────────────── + + // on_shutdown(handler: String) — register handler for SIGTERM/SIGINT / exit_clean(). + "on_shutdown" => { + let handler = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + if !handler.is_empty() { + shutdown_hooks().lock().unwrap_or_else(|e| e.into_inner()).push(handler); + } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // on_error(handler: String) — register handler(error_message: String) for unhandled errors. + "on_error" => { + let handler = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + if !handler.is_empty() { + error_hooks().lock().unwrap_or_else(|e| e.into_inner()).push(handler); + } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // exit_clean() — run on_shutdown handlers then exit(0). + "exit_clean" => { + run_shutdown_hooks(); + BuiltinResult::Exit(0) + } + + // ── Cache builtins ──────────────────────────────────────────────────── + + // cache_set(key: String, value: String, ttl_secs: Int) + "cache_set" => { + let ttl = match stack.pop().unwrap_or(Value::Int(60)) { + Value::Int(n) => n.max(0) as u64, + _ => 60, + }; + let value = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + other => other.to_string(), + }; + let key = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + cache::cache_set(key, value, ttl); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // cache_get(key: String) -> String ("" if missing/expired) + "cache_get" => { + let key = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let value = cache::cache_get(&key).unwrap_or_default(); + stack.push(Value::Str(value)); + BuiltinResult::Handled + } + + // cache_invalidate(key: String) + "cache_invalidate" => { + let key = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + cache::cache_invalidate(&key); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // cache_clear() + "cache_clear" => { + cache::cache_clear(); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ── HTTP retry builtins ─────────────────────────────────────────────── + + // http_get_retry(url: String, max_attempts: Int, backoff_ms: Int) -> String + "http_get_retry" => { + let backoff_ms = match stack.pop().unwrap_or(Value::Int(200)) { + Value::Int(n) => n.max(0) as u64, + _ => 200, + }; + let max_attempts = match stack.pop().unwrap_or(Value::Int(3)) { + Value::Int(n) => n.max(1) as u32, + _ => 3, + }; + let url = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let result = net::http_get_retry(&url, max_attempts, backoff_ms); + stack.push(Value::Str(result)); + BuiltinResult::Handled + } + + // http_post_retry(url: String, body: String, max_attempts: Int, backoff_ms: Int) -> String + "http_post_retry" => { + let backoff_ms = match stack.pop().unwrap_or(Value::Int(200)) { + Value::Int(n) => n.max(0) as u64, + _ => 200, + }; + let max_attempts = match stack.pop().unwrap_or(Value::Int(3)) { + Value::Int(n) => n.max(1) as u32, + _ => 3, + }; + let body = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let url = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let result = net::http_post_retry(&url, &body, max_attempts, backoff_ms); + stack.push(Value::Str(result)); + BuiltinResult::Handled + } + + // ── Circuit breaker builtins ────────────────────────────────────────── + + // circuit_open(name: String, failure_threshold: Int, reset_secs: Int) + "circuit_open" => { + let reset_secs = match stack.pop().unwrap_or(Value::Int(30)) { + Value::Int(n) => n.max(1) as u64, + _ => 30, + }; + let failure_threshold = match stack.pop().unwrap_or(Value::Int(5)) { + Value::Int(n) => n.max(1) as u32, + _ => 5, + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + net::circuit_open(name, failure_threshold, reset_secs); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // circuit_call(name: String, url: String, body: String) -> String + "circuit_call" => { + let body = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let url = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let result = net::circuit_call(&name, &url, &body); + stack.push(Value::Str(result)); + BuiltinResult::Handled + } + + // ── WebSocket server builtins ───────────────────────────────────────── + + // ws_serve(port: Int, handler: String) + // Starts a WebSocket server in the background. Incoming messages are + // queued; call ws_serve_poll(port) in a loop to drain them. + "ws_serve" => { + // handler name is accepted for API compatibility; messages are + // delivered via ws_serve_poll, not by invoking the handler directly. + let _handler = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let port = match stack.pop().unwrap_or(Value::Int(8765)) { + Value::Int(n) => n as u16, + _ => 8765, + }; + net::ws_serve_start(port); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ws_serve_poll(port: Int) -> List of {client_id, message} maps + // Drain all pending incoming messages on the server for one iteration. + "ws_serve_poll" => { + let port = match stack.pop().unwrap_or(Value::Int(8765)) { + Value::Int(n) => n as u16, + _ => 8765, + }; + let msgs = net::ws_server_poll(port, 0); + let list: Vec = msgs.into_iter().map(|m| { + Value::Map(vec![ + ("client_id".to_string(), Value::Str(m.client_id)), + ("message".to_string(), Value::Str(m.message)), + ]) + }).collect(); + stack.push(Value::List(list)); + BuiltinResult::Handled + } + + // ws_server_send(port: Int, client_id: String, message: String) -> Bool + "ws_server_send" => { + let message = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let client_id = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let port = match stack.pop().unwrap_or(Value::Int(8765)) { + Value::Int(n) => n as u16, + _ => 8765, + }; + let ok = net::ws_server_send(port, &client_id, message); + stack.push(Value::Bool(ok)); + BuiltinResult::Handled + } + + // ws_broadcast(port: Int, message: String) + "ws_broadcast" => { + let message = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let port = match stack.pop().unwrap_or(Value::Int(8765)) { + Value::Int(n) => n as u16, + _ => 8765, + }; + net::ws_server_broadcast(port, message); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ws_server_close(port: Int, client_id: String) + "ws_server_close" => { + let client_id = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let port = match stack.pop().unwrap_or(Value::Int(8765)) { + Value::Int(n) => n as u16, + _ => 8765, + }; + net::ws_server_close(port, &client_id); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ── WebSocket client (background / poll-based) builtins ─────────────── + + // ws_connect_async(url: String) -> String (conn_id) + // Unlike the synchronous ws_connect, this runs on a background thread + // and queues incoming messages for ws_client_poll(). + "ws_connect_async" => { + let url = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let conn_id = net::ws_client_connect(&url); + stack.push(Value::Str(conn_id)); + BuiltinResult::Handled + } + + // ws_client_send(conn_id: String, message: String) + "ws_client_send" => { + let message = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let conn_id = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + net::ws_client_send(&conn_id, message); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ws_client_close(conn_id: String) + "ws_client_close" => { + let conn_id = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + net::ws_client_close(&conn_id); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ws_client_poll(conn_id: String) -> List of {conn_id, message} maps + // Pass "" to get messages from all connections. + "ws_client_poll" => { + let conn_id_filter = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => String::new(), + }; + let filter = if conn_id_filter.is_empty() { None } else { Some(conn_id_filter.as_str()) }; + let msgs = net::ws_client_poll(filter); + let list: Vec = msgs.into_iter().map(|m| { + Value::Map(vec![ + ("conn_id".to_string(), Value::Str(m.conn_id)), + ("message".to_string(), Value::Str(m.message)), + ]) + }).collect(); + stack.push(Value::List(list)); + BuiltinResult::Handled + } + + // ── Subprocess / shell execution ────────────────────────────────────── + + // sh(command: String) -> String + "sh" => { + let cmd = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Str(String::new())); return BuiltinResult::Handled; } + }; + let output = std::process::Command::new("sh").arg("-c").arg(&cmd).output(); + let result = match output { + Ok(o) => { + if o.status.success() { + String::from_utf8_lossy(&o.stdout).to_string() + } else { + let out = String::from_utf8_lossy(&o.stdout); + let err = String::from_utf8_lossy(&o.stderr); + if out.is_empty() { err.to_string() } + else if err.is_empty() { out.to_string() } + else { format!("{}{}", out, err) } + } + } + Err(e) => format!("sh error: {}", e), + }; + stack.push(Value::Str(result)); + BuiltinResult::Handled + } + + // sh_async(command: String, callback: String) + "sh_async" => { + let callback = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let cmd = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let instr_arc = SERVE_INSTRUCTIONS.with(|si| si.borrow().clone()); + let fn_arc = SERVE_FN_TABLE.with(|sf| sf.borrow().clone()); + std::thread::spawn(move || { + let output = std::process::Command::new("sh").arg("-c").arg(&cmd).output(); + let result = match output { + Ok(o) => { + if o.status.success() { + String::from_utf8_lossy(&o.stdout).to_string() + } else { + let out = String::from_utf8_lossy(&o.stdout); + let err = String::from_utf8_lossy(&o.stderr); + if out.is_empty() { err.to_string() } else { out.to_string() } + } + } + Err(e) => format!("sh error: {}", e), + }; + if let (Some(instr), Some(fn_tbl)) = (instr_arc, fn_arc) { + if let Some(&entry) = fn_tbl.get(&callback) { + run_sub_interpreter_with_stack( + &instr, &fn_tbl, entry, + vec![el_compiler::Value::Str(result)], + ); + } + } + }); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ── Cryptographic primitives ────────────────────────────────────────── + + // crypto_keygen_ed25519() -> String "private: public:" + "crypto_keygen_ed25519" => { + use ed25519_dalek::SigningKey; + let sk = SigningKey::generate(&mut rand::rngs::OsRng); + let vk = sk.verifying_key(); + stack.push(Value::Str(format!( + "private:{} public:{}", + hex::encode(sk.to_bytes()), + hex::encode(vk.to_bytes()) + ))); + BuiltinResult::Handled + } + + // crypto_sign_ed25519(private_key_hex, message) -> String (hex sig) + "crypto_sign_ed25519" => { + use ed25519_dalek::{SigningKey, Signer}; + let message = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let priv_hex = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let result = (|| -> Result> { + let kb = hex::decode(&priv_hex)?; + let ka: [u8; 32] = kb.try_into().map_err(|_| "key must be 32 bytes")?; + Ok(hex::encode(SigningKey::from_bytes(&ka).sign(message.as_bytes()).to_bytes())) + })(); + stack.push(Value::Str(result.unwrap_or_else(|e| format!("error: {e}")))); + BuiltinResult::Handled + } + + // crypto_verify_ed25519(public_key_hex, message, sig_hex) -> Bool + "crypto_verify_ed25519" => { + use ed25519_dalek::{VerifyingKey, Verifier, Signature}; + let sig_hex = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let message = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let pub_hex = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let ok = (|| -> Result> { + let pb = hex::decode(&pub_hex)?; + let pa: [u8; 32] = pb.try_into().map_err(|_| "pub key 32 bytes")?; + let vk = VerifyingKey::from_bytes(&pa)?; + let sb = hex::decode(&sig_hex)?; + let sa: [u8; 64] = sb.try_into().map_err(|_| "sig 64 bytes")?; + vk.verify(message.as_bytes(), &Signature::from_bytes(&sa))?; + Ok(true) + })(); + stack.push(Value::Bool(ok.unwrap_or(false))); + BuiltinResult::Handled + } + + // crypto_encrypt_aes(key, plaintext) -> String (base64 nonce+ciphertext) + "crypto_encrypt_aes" => { + use aes_gcm::{Aes256Gcm, KeyInit, aead::{Aead, AeadCore, OsRng as AesOsRng}}; + let plaintext = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let key_in = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let result = (|| -> Result> { + let key_bytes = if key_in.len() == 64 { + hex::decode(&key_in).unwrap_or_else(|_| derive_aes_key(&key_in)) + } else { derive_aes_key(&key_in) }; + let ka: [u8; 32] = key_bytes.try_into().map_err(|_| "key error")?; + let cipher = Aes256Gcm::new_from_slice(&ka)?; + let nonce = Aes256Gcm::generate_nonce(&mut AesOsRng); + let ct = cipher.encrypt(&nonce, plaintext.as_bytes()).map_err(|_| "enc failed")?; + let mut out = nonce.to_vec(); out.extend_from_slice(&ct); + Ok(base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &out)) + })(); + stack.push(Value::Str(result.unwrap_or_else(|e| format!("error: {e}")))); + BuiltinResult::Handled + } + + // crypto_decrypt_aes(key, ciphertext_b64) -> String + "crypto_decrypt_aes" => { + use aes_gcm::{Aes256Gcm, KeyInit, aead::{Aead, generic_array::GenericArray}}; + let ct_b64 = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let key_in = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let result = (|| -> Result> { + let key_bytes = if key_in.len() == 64 { + hex::decode(&key_in).unwrap_or_else(|_| derive_aes_key(&key_in)) + } else { derive_aes_key(&key_in) }; + let ka: [u8; 32] = key_bytes.try_into().map_err(|_| "key error")?; + let cipher = Aes256Gcm::new_from_slice(&ka)?; + let combined = base64::Engine::decode( + &base64::engine::general_purpose::STANDARD, &ct_b64)?; + if combined.len() < 12 { return Err("too short".into()); } + let nonce = GenericArray::from_slice(&combined[..12]); + let pt = cipher.decrypt(nonce, &combined[12..]).map_err(|_| "dec failed")?; + Ok(String::from_utf8_lossy(&pt).to_string()) + })(); + stack.push(Value::Str(result.unwrap_or_else(|e| format!("error: {e}")))); + BuiltinResult::Handled + } + + // crypto_hash_sha256(input) -> String (hex) + "crypto_hash_sha256" => { + use sha2::{Sha256, Digest}; + let input = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + let mut h = Sha256::new(); h.update(input.as_bytes()); + stack.push(Value::Str(hex::encode(h.finalize()))); + BuiltinResult::Handled + } + + // crypto_hash_blake3(input) -> String (hex) + "crypto_hash_blake3" => { + let input = match stack.pop().unwrap_or(Value::Nil) { Value::Str(s) => s, _ => String::new() }; + stack.push(Value::Str(blake3::hash(input.as_bytes()).to_hex().to_string())); + BuiltinResult::Handled + } + + // crypto_random_bytes(n) -> String (hex) + "crypto_random_bytes" => { + use rand::RngCore; + let n = match stack.pop().unwrap_or(Value::Int(16)) { + Value::Int(i) => (i.max(0) as usize).min(4096), _ => 16, + }; + let mut bytes = vec![0u8; n]; + rand::rngs::OsRng.fill_bytes(&mut bytes); + stack.push(Value::Str(hex::encode(&bytes))); + BuiltinResult::Handled + } + + // crypto_uuid() -> String (UUID v4) + "crypto_uuid" => { + stack.push(Value::Str(uuid::Uuid::new_v4().to_string())); + BuiltinResult::Handled + } + + // ── Time extras ─────────────────────────────────────────────────────── + + // time_now_ms() -> Int + "time_now_ms" => { + let ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64).unwrap_or(0); + stack.push(Value::Int(ms)); + BuiltinResult::Handled + } + + // time_now_s() -> Int + "time_now_s" => { + let s = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64).unwrap_or(0); + stack.push(Value::Int(s)); + BuiltinResult::Handled + } + + // ── Timer / scheduling ──────────────────────────────────────────────── + + // timer_after(delay_ms, handler) + "timer_after" => { + let handler = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let delay_ms = match stack.pop().unwrap_or(Value::Int(0)) { + Value::Int(i) => i.max(0) as u64, _ => 0, + }; + let instr_arc = SERVE_INSTRUCTIONS.with(|si| si.borrow().clone()); + let fn_arc = SERVE_FN_TABLE.with(|sf| sf.borrow().clone()); + std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(delay_ms)); + if let (Some(instr), Some(fn_tbl)) = (instr_arc, fn_arc) { + if let Some(&entry) = fn_tbl.get(&handler) { + run_sub_interpreter(&instr, &fn_tbl, entry); + } + } + }); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // timer_every(interval_ms, handler) -> String (timer_id) + "timer_every" => { + let handler = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Str(String::new())); return BuiltinResult::Handled; } + }; + let interval_ms = match stack.pop().unwrap_or(Value::Int(1000)) { + Value::Int(i) => i.max(1) as u64, _ => 1000, + }; + let timer_id = next_timer_id(); + let (cancel_tx, cancel_rx) = crossbeam_channel::bounded::<()>(1); + timer_registry().lock().unwrap_or_else(|e| e.into_inner()) + .insert(timer_id.clone(), cancel_tx); + let instr_arc = SERVE_INSTRUCTIONS.with(|si| si.borrow().clone()); + let fn_arc = SERVE_FN_TABLE.with(|sf| sf.borrow().clone()); + let tid = timer_id.clone(); + std::thread::spawn(move || { + loop { + match cancel_rx.recv_timeout(std::time::Duration::from_millis(interval_ms)) { + Ok(_) | Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break, + Err(crossbeam_channel::RecvTimeoutError::Timeout) => { + if let (Some(ref instr), Some(ref fn_tbl)) = (&instr_arc, &fn_arc) { + if let Some(&entry) = fn_tbl.get(&handler) { + run_sub_interpreter(instr, fn_tbl, entry); + } + } + } + } + } + timer_registry().lock().unwrap_or_else(|e| e.into_inner()).remove(&tid); + }); + stack.push(Value::Str(timer_id)); + BuiltinResult::Handled + } + + // timer_cancel(timer_id) + "timer_cancel" => { + let timer_id = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + if let Some(tx) = timer_registry().lock().unwrap_or_else(|e| e.into_inner()) + .remove(&timer_id) { let _ = tx.send(()); } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // cron_schedule(expression, handler) -> String (job_id) + "cron_schedule" => { + let handler = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Str(String::new())); return BuiltinResult::Handled; } + }; + let expression = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Str(String::new())); return BuiltinResult::Handled; } + }; + let job_id = next_timer_id(); + let (cancel_tx, cancel_rx) = crossbeam_channel::bounded::<()>(1); + timer_registry().lock().unwrap_or_else(|e| e.into_inner()) + .insert(job_id.clone(), cancel_tx); + let instr_arc = SERVE_INSTRUCTIONS.with(|si| si.borrow().clone()); + let fn_arc = SERVE_FN_TABLE.with(|sf| sf.borrow().clone()); + let jid = job_id.clone(); + std::thread::spawn(move || { + loop { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH).unwrap_or_default(); + let nanos = 1_000_000_000u64.saturating_sub(now.subsec_nanos() as u64); + match cancel_rx.recv_timeout(std::time::Duration::from_nanos(nanos)) { + Ok(_) | Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break, + Err(crossbeam_channel::RecvTimeoutError::Timeout) => {} + } + if cancel_rx.try_recv().is_ok() { break; } + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + if cron_matches(&expression, ts) { + if let (Some(ref instr), Some(ref fn_tbl)) = (&instr_arc, &fn_arc) { + if let Some(&entry) = fn_tbl.get(&handler) { + run_sub_interpreter(instr, fn_tbl, entry); + } + } + match cancel_rx.recv_timeout(std::time::Duration::from_secs(60)) { + Ok(_) | Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break, + Err(crossbeam_channel::RecvTimeoutError::Timeout) => {} + } + } + } + timer_registry().lock().unwrap_or_else(|e| e.into_inner()).remove(&jid); + }); + stack.push(Value::Str(job_id)); + BuiltinResult::Handled + } + + // cron_cancel(job_id) + "cron_cancel" => { + let job_id = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + if let Some(tx) = timer_registry().lock().unwrap_or_else(|e| e.into_inner()) + .remove(&job_id) { let _ = tx.send(()); } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // ── Inter-thread channels ───────────────────────────────────────────── + + // chan_create(name, capacity) + "chan_create" => { + let capacity = match stack.pop().unwrap_or(Value::Int(64)) { + Value::Int(i) => i.max(1) as usize, _ => 64, + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let (tx, rx) = crossbeam_channel::bounded::(capacity); + channel_registry().lock().unwrap_or_else(|e| e.into_inner()) + .insert(name, ChannelEntry { sender: tx, receiver: rx }); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // chan_send(name, message) -> Bool + "chan_send" => { + let message = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, other => other.to_string(), + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Bool(false)); return BuiltinResult::Handled; } + }; + let sender = channel_registry().lock().unwrap_or_else(|e| e.into_inner()) + .get(&name).map(|e| e.sender.clone()); + stack.push(Value::Bool( + sender.map(|tx| tx.try_send(message).is_ok()).unwrap_or(false) + )); + BuiltinResult::Handled + } + + // chan_recv(name, timeout_ms) -> String ("" on timeout) + "chan_recv" => { + let timeout_ms = match stack.pop().unwrap_or(Value::Int(0)) { + Value::Int(i) => i.max(0) as u64, _ => 0, + }; + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Str(String::new())); return BuiltinResult::Handled; } + }; + let receiver = channel_registry().lock().unwrap_or_else(|e| e.into_inner()) + .get(&name).map(|e| e.receiver.clone()); + let msg = match receiver { + Some(rx) => if timeout_ms == 0 { + rx.recv().unwrap_or_default() + } else { + rx.recv_timeout(std::time::Duration::from_millis(timeout_ms)) + .unwrap_or_default() + }, + None => String::new(), + }; + stack.push(Value::Str(msg)); + BuiltinResult::Handled + } + + // chan_close(name) + "chan_close" => { + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + channel_registry().lock().unwrap_or_else(|e| e.into_inner()).remove(&name); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // chan_len(name) -> Int + "chan_len" => { + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { stack.push(Value::Int(0)); return BuiltinResult::Handled; } + }; + let len = channel_registry().lock().unwrap_or_else(|e| e.into_inner()) + .get(&name).map(|e| e.receiver.len() as i64).unwrap_or(0); + stack.push(Value::Int(len)); + BuiltinResult::Handled + } + + // ── Explicit observability builtins ────────────────────────────────── + + // log_debug(msg) / log_info(msg) / log_warn(msg) / log_error(msg) + "log_debug" => { + let msg = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + v => format!("{:?}", v), + }; + telemetry::log_debug(&msg); + stack.push(Value::Nil); + BuiltinResult::Handled + } + "log_info" => { + let msg = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + v => format!("{:?}", v), + }; + telemetry::log_info(&msg); + stack.push(Value::Nil); + BuiltinResult::Handled + } + "log_warn" => { + let msg = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + v => format!("{:?}", v), + }; + telemetry::log_warn(&msg); + stack.push(Value::Nil); + BuiltinResult::Handled + } + "log_error" => { + let msg = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + v => format!("{:?}", v), + }; + telemetry::log_error(&msg); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // trace_start(name) -> String (returns span_id) + "trace_start" => { + let name = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => "span".to_string(), + }; + let (trace_id, _parent_id) = telemetry::context::current_span() + .map(|(t, s)| (t, Some(s))) + .unwrap_or_else(|| (telemetry::new_trace_id(), None)); + let span_id = telemetry::new_span_id(); + let start_ns = telemetry::now_ns(); + EXPLICIT_SPANS.with(|m| { + m.borrow_mut().insert(span_id.clone(), (trace_id, span_id.clone(), start_ns, name, vec![])); + }); + stack.push(Value::Str(span_id)); + BuiltinResult::Handled + } + + // trace_end(span_id) + "trace_end" => { + let span_id = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let entry = EXPLICIT_SPANS.with(|m| m.borrow_mut().remove(&span_id)); + if let Some((trace_id, sid, start_ns, name, attrs)) = entry { + let end_ns = telemetry::now_ns(); + let parent_id = telemetry::context::current_span().map(|(_, s)| s); + telemetry::emit_span(telemetry::Span { + trace_id, + span_id: sid, + parent_id, + name, + start_ns, + end_ns, + attrs, + events: vec![], + status: telemetry::SpanStatus::Ok, + service: telemetry::service_name().to_string(), + }); + } + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // trace_tag(span_id, key, value) + "trace_tag" => { + let value = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + v => format!("{:?}", v), + }; + let key = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let span_id = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + EXPLICIT_SPANS.with(|m| { + if let Some(entry) = m.borrow_mut().get_mut(&span_id) { + entry.4.push((key, telemetry::AttrValue::Str(value))); + } + }); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // metric_counter(name, value, tags) tags = "k=v,k2=v2" + "metric_counter" => { + let tags = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => String::new(), + }; + let value = match stack.pop().unwrap_or(Value::Int(1)) { + Value::Int(n) => n, + Value::Float(f) => f as i64, + _ => 1, + }; + let name = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let attrs = telemetry::parse_tags_string(&tags); + telemetry::counter(&name, value as f64, attrs); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // metric_gauge(name, value, tags) + "metric_gauge" => { + let tags = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => String::new(), + }; + let value = match stack.pop().unwrap_or(Value::Int(0)) { + Value::Int(n) => n, + Value::Float(f) => f as i64, + _ => 0, + }; + let name = match stack.pop().unwrap_or(Value::Str(String::new())) { + Value::Str(s) => s, + _ => { stack.push(Value::Nil); return BuiltinResult::Handled; } + }; + let attrs = telemetry::parse_tags_string(&tags); + telemetry::gauge(&name, value as f64, attrs); + stack.push(Value::Nil); + BuiltinResult::Handled + } + + // graph_sql(query: String) -> String + // + // Execute a SQL query against the Engram graph via POST /api/sql and + // return the raw JSON response string. Uses ENGRAM_URL if set, + // falling back to http://localhost:8742. + "graph_sql" => { + let query = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + other => { + stack.push(Value::Str(format!( + "{{\"error\":\"graph_sql: expected string, got {:?}\"}}", + other + ))); + return BuiltinResult::Handled; + } + }; + let engram_url = std::env::var("ENGRAM_URL") + .unwrap_or_else(|_| "http://localhost:8742".to_string()); + let url = format!("{engram_url}/api/sql"); + let body = serde_json::json!({"query": query}).to_string(); + let result = net::with_single_retry(500, || { + reqwest::blocking::Client::new() + .post(&url) + .header("Content-Type", "application/json") + .body(body.clone()) + .send() + .and_then(|r| r.text()) + }); + stack.push(Value::Str(result)); + BuiltinResult::Handled + } + + // ── App block context builtins ─────────────────────────────────────────── + + "ctx_env" => { + let val = APP_ENV.with(|e| e.borrow().clone()); + let val = if val.is_empty() { + std::env::var("NEURON_ENV").unwrap_or_else(|_| "dev".to_string()) + } else { + val + }; + stack.push(Value::Str(val)); + BuiltinResult::Handled + } + + "ctx_service" => { + let val = APP_SERVICE.with(|s| s.borrow().clone()); + stack.push(Value::Str(val)); + BuiltinResult::Handled + } + + "ctx_version" => { + let val = APP_VERSION.with(|v| v.borrow().clone()); + stack.push(Value::Str(val)); + BuiltinResult::Handled + } + + "ctx_instance" => { + let val = APP_INSTANCE.with(|id| { + let current = id.borrow().clone(); + if current.is_empty() { + let new_id = uuid::Uuid::new_v4().to_string(); + *id.borrow_mut() = new_id.clone(); + new_id + } else { + current + } + }); + stack.push(Value::Str(val)); + BuiltinResult::Handled + } + + "config" => { + let default_val = if _arity >= 2 { + stack.pop().unwrap_or(Value::Nil) + } else { + Value::Nil + }; + let key = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { + stack.push(Value::Nil); + return BuiltinResult::Handled; + } + }; + // env var overrides everything + if let Ok(env_val) = std::env::var(&key) { + stack.push(Value::Str(env_val)); + return BuiltinResult::Handled; + } + let val = APP_CONFIG.with(|c| c.borrow().get(&key).cloned()); + match val { + Some(v) => stack.push(v), + None => { + if matches!(default_val, Value::Nil) { + eprintln!("Error: config key '{}' not found", key); + std::process::exit(1); + } + stack.push(default_val) + } + } + BuiltinResult::Handled + } + + "secret" => { + let key = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { + stack.push(Value::Str(String::new())); + return BuiltinResult::Handled; + } + }; + // Check env var first + if let Ok(env_val) = std::env::var(&key) { + stack.push(Value::Str(env_val)); + return BuiltinResult::Handled; + } + let val = APP_SECRETS.with(|s| s.borrow().get(&key).cloned()); + match val { + Some(v) => stack.push(Value::Str(v)), + None => { + // Try secrets.json directly + let secrets_file = load_neuron_secrets_json(); + match secrets_file.get(&key).cloned() { + Some(v) => stack.push(Value::Str(v)), + None => { + eprintln!( + "Error: required secret '{}' not found. Set environment variable or add to ~/.neuron/secrets.json", + key + ); + std::process::exit(1); + } + } + } + } + BuiltinResult::Handled + } + + "flag" => { + let default_val = if _arity >= 2 { + match stack.pop().unwrap_or(Value::Bool(false)) { + Value::Bool(b) => b, + _ => false, + } + } else { + false + }; + let key = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => { + stack.push(Value::Bool(false)); + return BuiltinResult::Handled; + } + }; + let val = APP_FLAGS.with(|f| f.borrow().get(&key).copied()); + stack.push(Value::Bool(val.unwrap_or(default_val))); + BuiltinResult::Handled + } + + // JSX rendering: __jsx__(tag, attrs_map, children_list) + // Renders a JSX element to an HTML string. + "__jsx__" => { + let children = stack.pop().unwrap_or(Value::Nil); + let attrs = stack.pop().unwrap_or(Value::Nil); + let tag_val = stack.pop().unwrap_or(Value::Nil); + + let tag = match &tag_val { + Value::Str(s) => s.clone(), + _ => "div".to_string(), + }; + + // Build attrs string + let attrs_str = match &attrs { + Value::Map(pairs) => { + pairs.iter().map(|(k, v)| { + let val_str = match v { + Value::Str(s) => s.clone(), + Value::Bool(b) => b.to_string(), + Value::Int(n) => n.to_string(), + _ => v.to_string(), + }; + // Skip event handlers (on:click etc.) in HTML output + if k.contains(':') || k.starts_with("on") { + String::new() + } else { + format!(" {}=\"{}\"", k, val_str) + } + }).collect::>().concat() + } + _ => String::new(), + }; + + // Build children string + let children_str = match children { + Value::List(items) => { + items.iter().map(|v| match v { + Value::Str(s) => s.clone(), + Value::Nil => String::new(), + other => other.to_string(), + }).collect::>().concat() + } + Value::Str(s) => s, + Value::Nil => String::new(), + other => other.to_string(), + }; + + // Self-closing tags + let is_void = matches!(tag.as_str(), "br" | "hr" | "img" | "input" | "link" | "meta"); + let html = if is_void || children_str.is_empty() && attrs_str.contains("/>") { + format!("<{tag}{attrs_str} />") + } else if tag == "fragment" { + children_str + } else { + format!("<{tag}{attrs_str}>{children_str}") + }; + + stack.push(Value::Str(html)); + BuiltinResult::Handled + } + + // Component render: __render_component__(name) — renders a named component + "__render_component__" => { + let name = match stack.pop().unwrap_or(Value::Nil) { + Value::Str(s) => s, + _ => "App".to_string(), + }; + stack.push(Value::Str(format!("[component: {name}]"))); + BuiltinResult::NotBuiltin // Let NotBuiltin branch call __component_ + } + _ => BuiltinResult::NotBuiltin, } } +/// Derive a 32-byte AES key from an arbitrary string via SHA-256. +fn derive_aes_key(s: &str) -> Vec { + use sha2::{Sha256, Digest}; + let mut h = Sha256::new(); + h.update(s.as_bytes()); + h.finalize().to_vec() +} + +/// Check whether a Unix timestamp (seconds) matches a 5-field cron expression. +/// Fields: minute hour day-of-month month day-of-week +fn cron_matches(expression: &str, unix_secs: u64) -> bool { + let total_days = unix_secs / 86_400; + let tod = unix_secs % 86_400; + let hour = (tod / 3600) as u32; + let min = ((tod % 3600) / 60) as u32; + let wday = ((total_days + 4) % 7) as u32; // 0=Sun + let (_year, month, mday) = cron_days_to_ymd(total_days); + let fields: Vec<&str> = expression.split_whitespace().collect(); + if fields.len() < 5 { return false; } + cron_field_matches(fields[0], min) + && cron_field_matches(fields[1], hour) + && cron_field_matches(fields[2], mday) + && cron_field_matches(fields[3], month) + && cron_field_matches(fields[4], wday) +} + +fn cron_days_to_ymd(days: u64) -> (u32, u32, u32) { + let mut rem = days; + let mut year = 1970u32; + loop { + let leap = (year % 4 == 0 && year % 100 != 0) || (year % 400 == 0); + let diy = if leap { 366 } else { 365 }; + if rem < diy { break; } + rem -= diy; + year += 1; + } + let leap = (year % 4 == 0 && year % 100 != 0) || (year % 400 == 0); + let mdays: [u32; 12] = [31, if leap { 29 } else { 28 }, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; + let mut month = 1u32; + let mut r = rem as u32; + for &md in &mdays { + if r < md { break; } + r -= md; month += 1; + } + (year, month, r + 1) +} + +fn cron_field_matches(field: &str, value: u32) -> bool { + if field == "*" { return true; } + if let Some(s) = field.strip_prefix("*/") { + let step: u32 = s.parse().unwrap_or(1); + return step > 0 && value % step == 0; + } + for part in field.split(',') { + let p = part.trim(); + if let Ok(n) = p.parse::() { + if n == value { return true; } + } else if let Some((s, st)) = p.split_once('/') { + let start: u32 = s.parse().unwrap_or(0); + let step: u32 = st.parse().unwrap_or(1); + if step > 0 && value >= start && (value - start) % step == 0 { return true; } + } + } + false +} + /// Return terminal dimensions as (cols, rows). /// Uses TIOCGWINSZ ioctl on Unix; falls back to (80, 24). fn term_size() -> (u16, u16) { @@ -6378,6 +8848,108 @@ thread_local! { static OBSERVER_ID_COUNTER: std::cell::Cell = std::cell::Cell::new(1); } +// ── GC / memory management state ───────────────────────────────────────────── + +/// Total allocations since process start (rough counter). +static GC_ALLOC_COUNT: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); +/// Timestamp (ms) of last gc_collect() call. +static GC_LAST_COLLECT_MS: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); +/// Soft memory limit in bytes (0 = no limit). +static MEM_LIMIT_BYTES: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); +/// GC threshold bytes: when estimated usage exceeds this, auto-trigger gc_collect (0 = off). +static GC_THRESHOLD_BYTES: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); + +fn now_ms_u64() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Estimate current process RSS in bytes. +fn estimate_mem_usage_bytes() -> u64 { + #[cfg(target_os = "linux")] + { + if let Ok(s) = std::fs::read_to_string("/proc/self/status") { + for line in s.lines() { + if line.starts_with("VmRSS:") { + let kb: u64 = line.split_whitespace() + .nth(1) + .and_then(|n| n.parse().ok()) + .unwrap_or(0); + return kb * 1024; + } + } + } + } + GC_ALLOC_COUNT.load(std::sync::atomic::Ordering::Relaxed) * 64 +} + +// ── Rate limiter state ──────────────────────────────────────────────────────── + +use std::sync::{OnceLock, Mutex}; + +static RATE_LIMITERS: OnceLock>>> = + OnceLock::new(); +static RATE_LIMITER_CONFIGS: OnceLock>> = + OnceLock::new(); + +fn rate_limiters() -> &'static Mutex>> { + RATE_LIMITERS.get_or_init(|| Mutex::new(std::collections::HashMap::new())) +} +fn rate_limiter_configs() -> &'static Mutex> { + RATE_LIMITER_CONFIGS.get_or_init(|| Mutex::new(std::collections::HashMap::new())) +} + +// ── Backpressure state ──────────────────────────────────────────────────────── + +#[derive(Clone)] +struct BackpressureController { + high_watermark: i64, + low_watermark: i64, + state: String, +} + +static BACKPRESSURE: OnceLock>> = + OnceLock::new(); + +fn backpressure_map() -> &'static Mutex> { + BACKPRESSURE.get_or_init(|| Mutex::new(std::collections::HashMap::new())) +} + +// ── Shutdown / error hooks ──────────────────────────────────────────────────── + +static SHUTDOWN_HOOKS: OnceLock>> = OnceLock::new(); +static ERROR_HOOKS: OnceLock>> = OnceLock::new(); + +fn shutdown_hooks() -> &'static Mutex> { + SHUTDOWN_HOOKS.get_or_init(|| Mutex::new(Vec::new())) +} +fn error_hooks() -> &'static Mutex> { + ERROR_HOOKS.get_or_init(|| Mutex::new(Vec::new())) +} + +fn run_shutdown_hooks() { + let hooks: Vec = { + let guard = shutdown_hooks().lock().unwrap_or_else(|e: std::sync::PoisonError>>| e.into_inner()); + guard.iter().rev().cloned().collect() + }; + if hooks.is_empty() { return; } + let instr_arc = SERVE_INSTRUCTIONS.with(|si| si.borrow().clone()); + let fn_arc = SERVE_FN_TABLE.with(|sf| sf.borrow().clone()); + if let (Some(instr), Some(fns)) = (instr_arc, fn_arc) { + for handler in hooks { + if let Some(&entry) = fns.get(&handler) { + run_sub_interpreter(&instr, &fns, entry); + } + } + } +} + thread_local! { /// LLM model name → (endpoint_url, api_key) static LLM_CONFIG: std::cell::RefCell> = @@ -6420,6 +8992,45 @@ thread_local! { }); } +// ── Inter-thread channel registry (global, cross-thread) ───────────────────── +// +// Named channels backed by crossbeam bounded channels. + +struct ChannelEntry { + sender: crossbeam_channel::Sender, + receiver: crossbeam_channel::Receiver, +} + +static CHANNEL_REGISTRY: std::sync::OnceLock< + std::sync::Mutex> +> = std::sync::OnceLock::new(); + +fn channel_registry() + -> &'static std::sync::Mutex> +{ + CHANNEL_REGISTRY.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new())) +} + +// ── Timer/scheduler registry (global, cross-thread) ────────────────────────── + +static TIMER_REGISTRY: std::sync::OnceLock< + std::sync::Mutex>> +> = std::sync::OnceLock::new(); + +fn timer_registry() + -> &'static std::sync::Mutex>> +{ + TIMER_REGISTRY.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new())) +} + +static TIMER_ID_COUNTER: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(1); + +fn next_timer_id() -> String { + let id = TIMER_ID_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + format!("tmr:{}", id) +} + // ── LLM helper functions ────────────────────────────────────────────────────── /// Map a short model name to the actual Anthropic model id string. @@ -6567,11 +9178,15 @@ fn is_builtin(name: &str) -> bool { | "str_slice" | "str_index_of" | "str_last_index_of" | "str_trim" | "str_upper" | "str_lower" | "str_len" | "string_len" | "list_get" | "list_len" | "list_join" + | "native_list_get" | "native_list_len" | "native_string_chars" + | "native_list_append" | "native_list_empty" | "native_int_to_str" + | "native_str_to_int" | "native_string_contains" | "uuid_new" | "uuid_v4" | "unix_timestamp" | "now_millis" | "timestamp" | "env" | "state_get" | "state_set" | "state_del" | "state_keys" | "http_get" | "http_post" | "http_put" | "http_delete" | "http_get_auth" | "http_post_auth" | "http_put_auth" | "http_delete_auth" + | "http_post_form_auth" | "http_post_engram" | "http_get_engram" | "engram_relate" | "engram_neighbors" | "engram_activate" | "fs_read" | "fs_write" | "fs_exists" | "fs_mkdir" | "fs_list" @@ -6607,6 +9222,45 @@ fn is_builtin(name: &str) -> bool { | "observe" | "unobserve" // LLM native functions | "llm_call" | "llm_call_system" | "llm_stream" | "llm_parallel" | "llm_models" | "llm_configure" + // GC / memory management + | "gc_collect" | "gc_stats" | "gc_set_threshold" | "mem_limit_set" | "mem_usage" + // Rate limiting + | "rate_limit_create" | "rate_limit_check" | "rate_limit_wait" + // Backpressure + | "backpressure_create" | "backpressure_signal" | "backpressure_check" + // App block context builtins + | "ctx_env" | "ctx_service" | "ctx_version" | "ctx_instance" + | "config" | "secret" | "flag" + // Lifecycle hooks + | "on_shutdown" | "on_error" | "exit_clean" + // Cache + | "cache_set" | "cache_get" | "cache_invalidate" | "cache_clear" + // HTTP retry + | "http_get_retry" | "http_post_retry" + // Circuit breaker + | "circuit_open" | "circuit_call" + // WebSocket server + | "ws_serve" | "ws_serve_poll" | "ws_server_send" | "ws_broadcast" | "ws_server_close" + // WebSocket client (async/poll-based) + | "ws_connect_async" | "ws_client_send" | "ws_client_close" | "ws_client_poll" + // Subprocess + | "sh" | "sh_async" + // Crypto + | "crypto_keygen_ed25519" | "crypto_sign_ed25519" | "crypto_verify_ed25519" + | "crypto_encrypt_aes" | "crypto_decrypt_aes" + | "crypto_hash_sha256" | "crypto_hash_blake3" + | "crypto_random_bytes" | "crypto_uuid" + // Time extras + | "time_now_ms" | "time_now_s" + // Timers + | "timer_after" | "timer_every" | "timer_cancel" + | "cron_schedule" | "cron_cancel" + // Channels + | "chan_create" | "chan_send" | "chan_recv" | "chan_close" | "chan_len" + // Explicit observability + | "log_debug" | "log_info" | "log_warn" | "log_error" + | "trace_start" | "trace_end" | "trace_tag" + | "metric_counter" | "metric_gauge" ) }