jan-cli 0.27.0

//! Jan-owned language runtime pool (warm interpreters).
//!
//! Workers are Jan-internal children (not a third MAS peer daemon). Each job runs
//! in an isolated child so scripts cannot leak state into the worker or crash it
//! permanently. Unix only; other platforms always cold-spawn.

use std::collections::BTreeMap;
use std::fs;
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::config;
use crate::remote;

const SOCKET_FILE: &str = "runtime.sock";
const PID_FILE: &str = "runtime.pid";
const MAX_WORKERS: usize = 16;
const IDLE_TTL_SECS: u64 = 15 * 60;
const WORKER_READY_TIMEOUT: Duration = Duration::from_secs(10);
const JOB_TIMEOUT: Duration = Duration::from_secs(60 * 30);

static STOP_REQUESTED: AtomicBool = AtomicBool::new(false);
static WARM_HITS: AtomicU64 = AtomicU64::new(0);
static COLD_FALLBACKS: AtomicU64 = AtomicU64::new(0);
static JOBS_OK: AtomicU64 = AtomicU64::new(0);
static JOBS_ERR: AtomicU64 = AtomicU64::new(0);

/// Whether the warm runtime pool may be used (`JAN_RUNTIME=0` disables).
pub fn runtime_enabled() -> bool {
    match std::env::var("JAN_RUNTIME") {
        Ok(v) => {
            let t = v.trim();
            !(t == "0" || t.eq_ignore_ascii_case("false") || t.eq_ignore_ascii_case("off"))
        }
        Err(_) => true,
    }
}

pub fn runtime_dir() -> PathBuf {
    if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") {
        let dir = dir.trim();
        if !dir.is_empty() {
            return PathBuf::from(dir).join("jan-cli");
        }
    }
    config::config_dir().join("run")
}

pub fn socket_path() -> PathBuf {
    runtime_dir().join(SOCKET_FILE)
}

pub fn pid_path() -> PathBuf {
    runtime_dir().join(PID_FILE)
}

fn worker_sock_path(key_hash: &str) -> PathBuf {
    runtime_dir().join(format!("runtime-w-{key_hash}.sock"))
}

fn jan_bin() -> Result<PathBuf> {
    let exe = std::env::current_exe().context("resolve jan binary path")?;
    exe.canonicalize()
        .with_context(|| format!("canonicalize {}", exe.display()))
}

fn short_hash(parts: &[&str]) -> String {
    let mut h = Sha256::new();
    for p in parts {
        h.update(p.as_bytes());
        h.update([0]);
    }
    let dig = h.finalize();
    hex::encode(&dig[..8])
}

/// Hex encode without adding a hex crate — small helper.
mod hex {
    pub fn encode(bytes: &[u8]) -> String {
        const HEX: &[u8] = b"0123456789abcdef";
        let mut s = String::with_capacity(bytes.len() * 2);
        for &b in bytes {
            s.push(HEX[(b >> 4) as usize] as char);
            s.push(HEX[(b & 0xf) as usize] as char);
        }
        s
    }
}

fn materialize_script(name: &str, body: &str) -> Result<PathBuf> {
    let dir = remote::cache_root()?.join("runtime");
    fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
    let hash = short_hash(&[body]);
    let path = dir.join(format!("{name}-{hash}"));
    if !path.is_file() {
        fs::write(&path, body).with_context(|| format!("write {}", path.display()))?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut perms = fs::metadata(&path)?.permissions();
            perms.set_mode(0o755);
            fs::set_permissions(&path, perms)?;
        }
    }
    Ok(path)
}

fn python_worker_script() -> Result<PathBuf> {
    materialize_script(
        "python_worker.py",
        include_str!("../runtime/python_worker.py"),
    )
}

fn node_worker_script() -> Result<PathBuf> {
    materialize_script("node_worker.js", include_str!("../runtime/node_worker.js"))
}

fn shell_worker_script() -> Result<PathBuf> {
    materialize_script("shell_worker.py", include_str!("../runtime/shell_worker.py"))
}

fn kotlin_worker_script() -> Result<PathBuf> {
    materialize_script(
        "kotlin_worker.py",
        include_str!("../runtime/kotlin_worker.py"),
    )
}

/// Language identity for a warm worker.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum RuntimeLang {
    Python,
    Node,
    Bash,
    Sh,
    Zsh,
    Kotlin,
}

impl RuntimeLang {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Python => "python",
            Self::Node => "node",
            Self::Bash => "bash",
            Self::Sh => "sh",
            Self::Zsh => "zsh",
            Self::Kotlin => "kotlin",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkerKey {
    pub lang: RuntimeLang,
    /// Absolute interpreter / java / python used to boot the worker.
    pub interpreter: String,
    /// Package env root (uv/pnpm/gradle cache dir) or empty for system.
    pub env_root: String,
    /// Optional NODE_PATH baked into the worker process.
    #[serde(default)]
    pub node_path: Option<String>,
}

impl WorkerKey {
    pub fn hash(&self) -> String {
        short_hash(&[
            self.lang.as_str(),
            &self.interpreter,
            &self.env_root,
            self.node_path.as_deref().unwrap_or(""),
        ])
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobSource {
    pub kind: String,
    #[serde(default)]
    pub value: String,
    #[serde(default)]
    pub classpath: Option<String>,
    #[serde(default)]
    pub main_class: Option<String>,
    #[serde(default)]
    pub java: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobRequest {
    pub cwd: String,
    pub env: BTreeMap<String, String>,
    pub source: JobSource,
    #[serde(default)]
    pub argv: Vec<String>,
    #[serde(default)]
    pub shell: Option<String>,
    #[serde(default)]
    pub argv0: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct JobResponse {
    ok: bool,
    #[serde(default)]
    exit_code: Option<i32>,
    #[serde(default)]
    error: Option<String>,
    #[serde(default)]
    stdout_b64: Option<String>,
    #[serde(default)]
    stderr_b64: Option<String>,
}

struct WorkerHandle {
    key: WorkerKey,
    key_hash: String,
    child: Child,
    sock: PathBuf,
    last_used: Instant,
    jobs: u64,
}

struct SupervisorState {
    workers: Vec<WorkerHandle>,
    started: Instant,
}

impl SupervisorState {
    fn status_line(&self) -> String {
        let warm = WARM_HITS.load(Ordering::Relaxed);
        let cold = COLD_FALLBACKS.load(Ordering::Relaxed);
        let ok = JOBS_OK.load(Ordering::Relaxed);
        let err = JOBS_ERR.load(Ordering::Relaxed);
        let up = self.started.elapsed().as_secs();
        let mut parts: Vec<String> = self
            .workers
            .iter()
            .map(|w| {
                format!(
                    "{}:{}:jobs={}",
                    w.key.lang.as_str(),
                    &w.key_hash[..8.min(w.key_hash.len())],
                    w.jobs
                )
            })
            .collect();
        parts.sort();
        format!(
            "ok uptime={up}s workers={} warm_hits={warm} cold_fallbacks={cold} jobs_ok={ok} jobs_err={err} [{}]",
            self.workers.len(),
            parts.join(" ")
        )
    }

    fn reap_idle(&mut self) {
        let ttl = Duration::from_secs(IDLE_TTL_SECS);
        let mut keep = Vec::new();
        for mut w in self.workers.drain(..) {
            if w.last_used.elapsed() > ttl {
                let _ = w.child.kill();
                let _ = w.child.wait();
                let _ = fs::remove_file(&w.sock);
            } else {
                keep.push(w);
            }
        }
        self.workers = keep;
    }

    fn stop_all(&mut self) {
        for mut w in self.workers.drain(..) {
            let _ = w.child.kill();
            let _ = w.child.wait();
            let _ = fs::remove_file(&w.sock);
        }
    }

    fn find_mut(&mut self, hash: &str) -> Option<&mut WorkerHandle> {
        self.workers.iter_mut().find(|w| w.key_hash == hash)
    }

    fn register_worker(&mut self, key: WorkerKey, hash: String, child: Child, sock: PathBuf) {
        self.workers.push(WorkerHandle {
            key,
            key_hash: hash,
            child,
            sock,
            last_used: Instant::now(),
            jobs: 0,
        });
    }
}

fn spawn_worker_process(key: &WorkerKey, sock: &Path) -> Result<Child> {
    let _ = fs::remove_file(sock);
    let mut cmd = match key.lang {
        RuntimeLang::Python => {
            let script = python_worker_script()?;
            let mut c = Command::new(&key.interpreter);
            c.arg("-u").arg(script).arg(sock);
            c
        }
        RuntimeLang::Node => {
            let script = node_worker_script()?;
            let mut c = Command::new(&key.interpreter);
            c.arg(script).arg(sock);
            if let Some(np) = &key.node_path {
                c.env("NODE_PATH", np);
            }
            c
        }
        RuntimeLang::Bash | RuntimeLang::Sh | RuntimeLang::Zsh => {
            // Shell worker is a Python supervisor that spawns shell children.
            let script = shell_worker_script()?;
            let py = which_python()?;
            let mut c = Command::new(py);
            c.arg(script).arg(sock);
            c
        }
        RuntimeLang::Kotlin => {
            let script = kotlin_worker_script()?;
            let py = which_python()?;
            let mut c = Command::new(py);
            c.arg(script).arg(sock);
            c
        }
    };
    cmd.stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::inherit());
    if !key.env_root.is_empty() {
        cmd.env("JAN_RUNTIME_ENV_ROOT", &key.env_root);
    }
    cmd.spawn()
        .with_context(|| format!("spawn {} worker", key.lang.as_str()))
}

fn which_python() -> Result<PathBuf> {
    for name in ["python3", "python"] {
        if let Ok(p) = crate::deps::resolve_program(name, &[]) {
            return Ok(p);
        }
    }
    bail!("python3 required to host shell/kotlin runtime workers")
}

fn spawn_worker(key: &WorkerKey, hash: &str) -> Result<PathBuf> {
    // Called from ensure path that registers — see ensure_worker_locked
    let sock = worker_sock_path(hash);
    let mut child = spawn_worker_process(key, &sock)?;
    let stdout = child.stdout.take().context("worker stdout")?;
    let mut reader = BufReader::new(stdout);
    let mut line = String::new();
    let start = Instant::now();
    loop {
        if start.elapsed() > WORKER_READY_TIMEOUT {
            let _ = child.kill();
            bail!(
                "{} worker failed to become ready within {:?}",
                key.lang.as_str(),
                WORKER_READY_TIMEOUT
            );
        }
        line.clear();
        match reader.read_line(&mut line) {
            Ok(0) => {
                let _ = child.kill();
                bail!("{} worker exited before ready", key.lang.as_str());
            }
            Ok(_) => {
                if line.starts_with("ready ") {
                    break;
                }
            }
            Err(e) => {
                let _ = child.kill();
                bail!("{} worker ready read: {e}", key.lang.as_str());
            }
        }
    }
    // Detach remaining stdout so the pipe does not fill.
    thread::spawn(move || {
        let mut r = reader;
        let mut sink = Vec::new();
        let _ = r.read_to_end(&mut sink);
    });
    // Registration happens in caller with child — we need to return child too.
    // Refactor: store in thread-local? Better change signature.
    // Use a static pending slot — ugly. Change ensure to call spawn_and_register.
    PENDING_CHILD
        .lock()
        .unwrap()
        .replace((key.clone(), hash.to_string(), child, sock.clone()));
    Ok(sock)
}

static PENDING_CHILD: Mutex<Option<(WorkerKey, String, Child, PathBuf)>> = Mutex::new(None);

fn ensure_worker_locked(state: &mut SupervisorState, key: WorkerKey) -> Result<PathBuf> {
    state.reap_idle();
    let hash = key.hash();
    if let Some(w) = state.find_mut(&hash) {
        match w.child.try_wait() {
            Ok(None) => {
                w.last_used = Instant::now();
                return Ok(w.sock.clone());
            }
            _ => {
                let _ = fs::remove_file(&w.sock);
            }
        }
        state.workers.retain(|w| w.key_hash != hash);
    }
    if state.workers.len() >= MAX_WORKERS {
        if let Some(idx) = state
            .workers
            .iter()
            .enumerate()
            .min_by_key(|(_, w)| w.last_used)
            .map(|(i, _)| i)
        {
            let mut old = state.workers.remove(idx);
            let _ = old.child.kill();
            let _ = old.child.wait();
            let _ = fs::remove_file(&old.sock);
        }
    }
    let sock = spawn_worker(&key, &hash)?;
    if let Some((k, h, child, s)) = PENDING_CHILD.lock().unwrap().take() {
        state.register_worker(k, h, child, s);
    }
    Ok(sock)
}

fn write_pid_file() -> Result<()> {
    let dir = runtime_dir();
    fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
    fs::write(pid_path(), format!("{}\n", std::process::id()))
        .with_context(|| format!("write {}", pid_path().display()))?;
    Ok(())
}

fn remove_runtime_files() {
    let _ = fs::remove_file(socket_path());
    let _ = fs::remove_file(pid_path());
}

#[cfg(unix)]
fn accept_control(state: Arc<Mutex<SupervisorState>>) {
    use std::os::unix::net::UnixListener;

    let listener = match UnixListener::bind(socket_path()) {
        Ok(l) => l,
        Err(e) => {
            eprintln!(
                "jan runtime daemon: bind {}: {e:#}",
                socket_path().display()
            );
            STOP_REQUESTED.store(true, Ordering::SeqCst);
            return;
        }
    };
    let _ = listener.set_nonblocking(false);
    for stream in listener.incoming() {
        if STOP_REQUESTED.load(Ordering::SeqCst) {
            break;
        }
        match stream {
            Ok(s) => {
                let st = Arc::clone(&state);
                thread::spawn(move || handle_client(s, st));
            }
            Err(e) => {
                eprintln!("jan runtime daemon: accept: {e:#}");
            }
        }
    }
}

#[cfg(unix)]
fn handle_client(mut stream: std::os::unix::net::UnixStream, state: Arc<Mutex<SupervisorState>>) {
    use std::os::unix::net::UnixStream;

    let reader = BufReader::new(
        stream
            .try_clone()
            .unwrap_or_else(|_| stream.try_clone().expect("clone runtime control socket")),
    );
    let line = match reader.lines().next() {
        Some(Ok(l)) => l,
        _ => return,
    };
    let line = line.trim();
    let (cmd, rest) = match line.split_once(char::is_whitespace) {
        Some((c, r)) => (c, r.trim()),
        None => (line, ""),
    };
    let reply = match cmd.to_ascii_lowercase().as_str() {
        "ping" => "pong".to_string(),
        "status" => state.lock().unwrap().status_line(),
        "stop" => {
            STOP_REQUESTED.store(true, Ordering::SeqCst);
            state.lock().unwrap().stop_all();
            "ok stopping".to_string()
        }
        "ensure" => match serde_json::from_str::<WorkerKey>(rest) {
            Ok(key) => match ensure_worker_locked(&mut state.lock().unwrap(), key) {
                Ok(sock) => format!("ok sock={}", sock.display()),
                Err(e) => format!("error {e:#}"),
            },
            Err(e) => format!("error bad key json: {e}"),
        },
        "note_cold" => {
            COLD_FALLBACKS.fetch_add(1, Ordering::Relaxed);
            "ok".to_string()
        }
        "note_warm" => {
            WARM_HITS.fetch_add(1, Ordering::Relaxed);
            JOBS_OK.fetch_add(1, Ordering::Relaxed);
            if !rest.is_empty() {
                if let Some(w) = state.lock().unwrap().find_mut(rest) {
                    w.jobs += 1;
                    w.last_used = Instant::now();
                }
            }
            "ok".to_string()
        }
        other => format!("error unknown command `{other}`"),
    };
    let _ = writeln!(stream, "{reply}");
    let _ = stream.flush();
    // stop closes listener by exiting process after flag — wake by connecting
    if cmd.eq_ignore_ascii_case("stop") {
        // Best-effort connect to unblock accept after STOP — actually accept
        // blocks until next client; spawn a ping to ourselves.
        let path = socket_path();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(50));
            let _ = UnixStream::connect(path);
        });
    }
}

/// Run the runtime supervisor in the foreground (`jan runtime daemon`).
pub fn run_foreground() -> Result<i32> {
    #[cfg(not(unix))]
    {
        bail!("jan runtime daemon requires Unix");
    }
    #[cfg(unix)]
    {
        STOP_REQUESTED.store(false, Ordering::SeqCst);
        fs::create_dir_all(runtime_dir())
            .with_context(|| format!("create {}", runtime_dir().display()))?;
        remove_runtime_files();
        write_pid_file()?;
        let state = Arc::new(Mutex::new(SupervisorState {
            workers: Vec::new(),
            started: Instant::now(),
        }));
        let state_bg = Arc::clone(&state);
        let control = thread::spawn(move || accept_control(state_bg));
        while !STOP_REQUESTED.load(Ordering::SeqCst) {
            thread::sleep(Duration::from_millis(200));
            if let Ok(mut st) = state.try_lock() {
                st.reap_idle();
            }
        }
        state.lock().unwrap().stop_all();
        remove_runtime_files();
        let _ = control.join();
        Ok(0)
    }
}

pub fn send_command(cmd: &str) -> Result<String> {
    #[cfg(not(unix))]
    {
        let _ = cmd;
        bail!("jan runtime requires Unix");
    }
    #[cfg(unix)]
    {
        use std::os::unix::net::UnixStream;
        let path = socket_path();
        if !path.exists() {
            bail!(
                "jan runtime daemon is not running (no socket at {})",
                path.display()
            );
        }
        let mut stream = UnixStream::connect(&path)
            .with_context(|| format!("connect to {}", path.display()))?;
        stream.set_read_timeout(Some(Duration::from_secs(5)))?;
        stream.set_write_timeout(Some(Duration::from_secs(5)))?;
        writeln!(stream, "{cmd}").context("write runtime command")?;
        let mut reader = BufReader::new(stream);
        let mut reply = String::new();
        reader.read_line(&mut reply).context("read runtime reply")?;
        Ok(reply.trim().to_string())
    }
}

pub fn daemon_running() -> bool {
    match send_command("ping") {
        Ok(r) => r == "pong",
        Err(_) => false,
    }
}

pub fn start_daemon_background() -> Result<()> {
    if !runtime_enabled() {
        return Ok(());
    }
    #[cfg(not(unix))]
    {
        return Ok(());
    }
    #[cfg(unix)]
    {
        if daemon_running() {
            return Ok(());
        }
        let jan = jan_bin()?;
        let mut cmd = Command::new(&jan);
        cmd.args(["--no-log", "runtime", "daemon"]);
        cmd.stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null());
        cmd.spawn()
            .with_context(|| format!("spawn `{} runtime daemon`", jan.display()))?;
        for _ in 0..50 {
            if daemon_running() {
                return Ok(());
            }
            thread::sleep(Duration::from_millis(100));
        }
        bail!(
            "jan runtime daemon failed to start (no response on {})",
            socket_path().display()
        );
    }
}

pub fn stop_daemon() -> Result<()> {
    if !socket_path().exists() {
        println!("(jan runtime daemon is not running)");
        return Ok(());
    }
    let reply = send_command("stop")?;
    for _ in 0..30 {
        if !socket_path().exists() {
            println!("stopped jan runtime daemon");
            return Ok(());
        }
        thread::sleep(Duration::from_millis(100));
    }
    bail!("jan runtime daemon did not stop: {reply}");
}

pub fn daemon_status() -> Result<i32> {
    match send_command("status") {
        Ok(reply) if reply.starts_with("ok ") => {
            println!("jan runtime daemon running ({reply})");
            Ok(0)
        }
        Ok(reply) => {
            println!("jan runtime daemon: {reply}");
            Ok(1)
        }
        Err(e) => {
            println!("jan runtime daemon is not running ({e:#})");
            Ok(1)
        }
    }
}

/// Ensure a worker exists for `key`; returns the worker socket path.
pub fn ensure_worker(key: &WorkerKey) -> Result<PathBuf> {
    start_daemon_background()?;
    let json = serde_json::to_string(key).context("serialize worker key")?;
    let reply = send_command(&format!("ensure {json}"))?;
    if let Some(sock) = reply.strip_prefix("ok sock=") {
        return Ok(PathBuf::from(sock));
    }
    if let Some(err) = reply.strip_prefix("error ") {
        bail!("{err}");
    }
    bail!("unexpected ensure reply: {reply}");
}

fn b64_decode(s: &str) -> Result<Vec<u8>> {
    fn val(c: u8) -> Option<u8> {
        match c {
            b'A'..=b'Z' => Some(c - b'A'),
            b'a'..=b'z' => Some(c - b'a' + 26),
            b'0'..=b'9' => Some(c - b'0' + 52),
            b'+' => Some(62),
            b'/' => Some(63),
            _ => None,
        }
    }
    let bytes = s.as_bytes();
    let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
    let mut buf = 0u32;
    let mut n = 0;
    for &c in bytes {
        if c == b'=' || c.is_ascii_whitespace() {
            continue;
        }
        let Some(v) = val(c) else {
            bail!("invalid base64");
        };
        buf = (buf << 6) | u32::from(v);
        n += 6;
        if n >= 8 {
            n -= 8;
            out.push((buf >> n) as u8);
        }
    }
    Ok(out)
}

/// Send a job to a worker socket and write captured stdio to this process.
pub fn run_job_on_worker(sock: &Path, job: &JobRequest) -> Result<i32> {
    #[cfg(not(unix))]
    {
        let _ = (sock, job);
        bail!("warm runtime requires Unix");
    }
    #[cfg(unix)]
    {
        use std::os::unix::net::UnixStream;
        let mut stream = UnixStream::connect(sock)
            .with_context(|| format!("connect to worker {}", sock.display()))?;
        stream.set_read_timeout(Some(JOB_TIMEOUT))?;
        stream.set_write_timeout(Some(Duration::from_secs(30)))?;
        let payload = serde_json::to_string(job).context("serialize job")?;
        writeln!(stream, "{payload}").context("write job")?;
        let mut reader = BufReader::new(stream);
        let mut reply = String::new();
        reader.read_line(&mut reply).context("read job reply")?;
        let resp: JobResponse =
            serde_json::from_str(reply.trim()).with_context(|| format!("parse job reply: {reply}"))?;
        if !resp.ok {
            JOBS_ERR.fetch_add(1, Ordering::Relaxed);
            bail!(
                "worker error: {}",
                resp.error.unwrap_or_else(|| "unknown".into())
            );
        }
        if let Some(b64) = &resp.stdout_b64 {
            let bytes = b64_decode(b64)?;
            let _ = std::io::stdout().write_all(&bytes);
            let _ = std::io::stdout().flush();
        }
        if let Some(b64) = &resp.stderr_b64 {
            let bytes = b64_decode(b64)?;
            let _ = std::io::stderr().write_all(&bytes);
            let _ = std::io::stderr().flush();
        }
        let code = resp.exit_code.unwrap_or(255);
        // Status counters live in the supervisor process.
        let key_hint = sock
            .file_name()
            .and_then(|s| s.to_str())
            .and_then(|s| s.strip_prefix("runtime-w-"))
            .and_then(|s| s.strip_suffix(".sock"))
            .unwrap_or("");
        let _ = send_command(&format!("note_warm {key_hint}"));
        Ok(code)
    }
}

/// Try to run via the warm pool. Returns `Ok(None)` to signal cold fallback.
pub fn try_run_warm(key: &WorkerKey, job: &JobRequest) -> Result<Option<i32>> {
    if !runtime_enabled() {
        return Ok(None);
    }
    #[cfg(not(unix))]
    {
        let _ = (key, job);
        return Ok(None);
    }
    #[cfg(unix)]
    {
        match (|| -> Result<i32> {
            let sock = ensure_worker(key)?;
            run_job_on_worker(&sock, job)
        })() {
            Ok(code) => Ok(Some(code)),
            Err(e) => {
                let _ = send_command("note_cold");
                if std::env::var_os("JAN_RUNTIME_DEBUG").is_some() {
                    eprintln!("jan: warm runtime fallback: {e:#}");
                }
                Ok(None)
            }
        }
    }
}

/// CLI: `jan runtime …`
pub fn dispatch_runtime(args: &[std::ffi::OsString]) -> Result<i32> {
    let mut it = args.iter();
    let sub = it
        .next()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_else(|| "status".into());
    match sub.as_str() {
        "help" | "--help" | "-h" => {
            print_runtime_help();
            Ok(0)
        }
        "daemon" => run_foreground(),
        "start" => {
            start_daemon_background()?;
            println!("started jan runtime daemon");
            Ok(0)
        }
        "stop" => {
            stop_daemon()?;
            Ok(0)
        }
        "restart" => {
            let _ = stop_daemon();
            start_daemon_background()?;
            println!("restarted jan runtime daemon");
            Ok(0)
        }
        "status" => daemon_status(),
        other => {
            bail!("unknown runtime subcommand `{other}` (try `jan runtime --help`)");
        }
    }
}

fn print_runtime_help() {
    print!(
        "\
jan runtime — warm language interpreter pool (Jan-internal)

Usage:
  jan runtime start
  jan runtime stop
  jan runtime restart
  jan runtime status
  jan runtime daemon     Run supervisor in the foreground

The pool hosts long-lived Python/Node (and shell/Kotlin) workers. Each job runs
in an isolated child. Set JAN_RUNTIME=0 to force cold interpreter spawns.
"
    );
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn worker_key_hash_stable() {
        let k = WorkerKey {
            lang: RuntimeLang::Python,
            interpreter: "/usr/bin/python3".into(),
            env_root: "/tmp/env".into(),
            node_path: None,
        };
        assert_eq!(k.hash(), k.hash());
        assert_eq!(k.hash().len(), 16);
    }

    #[test]
    fn runtime_enabled_default() {
        // Do not assert global env; just ensure function is callable.
        let _ = runtime_enabled();
    }
}