rayforce-sys 1.2.0

/*
 *   Copyright (c) 2025-2026 Anton Kundenko <singaraiona@gmail.com>
 *   All rights reserved.

 *   Permission is hereby granted, free of charge, to any person obtaining a copy
 *   of this software and associated documentation files (the "Software"), to deal
 *   in the Software without restriction, including without limitation the rights
 *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *   copies of the Software, and to permit persons to whom the Software is
 *   furnished to do so, subject to the following conditions:

 *   The above copyright notice and this permission notice shall be included in all
 *   copies or substantial portions of the Software.

 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 *   SOFTWARE.
 */

#if !defined(_WIN32) && !defined(_GNU_SOURCE)
#  define _GNU_SOURCE   /* popen()/pclose() and gethostname() — .sys.exec, .sys.info */
#endif

#include "lang/internal.h"
#include "lang/env.h"
#include "lang/eval.h"  /* LAMBDA_PARAMS */
#include "lang/nfo.h"   /* NFO_FILENAME */
#include "lang/parse.h"
#include "ops/ops.h"    /* ray_is_lazy, ray_lazy_materialize */
#include "ops/internal.h"   /* ray_group_perpart_runs — (.sys.mem) counter */
#include "mem/heap.h"
#include "mem/sys.h"
#include "core/block.h"
#include "ops/idxop.h"
#include "core/profile.h"   /* g_ray_profile — (.sys.prof) */
#include "core/qmeasure.h"  /* shared per-query measurement lifecycle */
#include "core/qlog.h"      /* g_qlog — (.sys.querylog) */
#include "store/serde.h"
#include "store/splay.h"
#include "store/part.h"
#include "core/ipc.h"
#include "core/mcast.h"
#include "core/poll.h"
#include "core/timer.h"

/* Forward decls: keep this TU decoupled from core/runtime.h — only
 * the opaque accessors are needed. */
void* ray_runtime_get_poll(void);
void  ray_runtime_set_sys_args(void* dict);
void* ray_runtime_get_sys_args(void);
#include <time.h>
#include <ctype.h>
#include <errno.h>
#include <limits.h>
#include <stdlib.h>
#include <string.h>

#if !defined(RAY_OS_WINDOWS)
#include <unistd.h>
#include <sys/wait.h>   /* WIFEXITED/WEXITSTATUS — .sys.exec exit codes */
#define RAY_POPEN(c, m)  popen((c), (m))
#define RAY_PCLOSE(f)    pclose(f)
#else
#include <io.h>         /* access, F_OK */
#define RAY_POPEN(c, m)  _popen((c), (m))
#define RAY_PCLOSE(f)    _pclose(f)
#endif

static int parse_sys_int_arg(const char* s, bool nonnegative, bool core_count,
                             int64_t* out) {
    if (!s || !*s) return 0;
    const unsigned char* p = (const unsigned char*)s;
    if (!nonnegative && (*p == '+' || *p == '-')) p++;
    if (!isdigit(*p)) return 0;
    for (const unsigned char* q = p; *q; q++)
        if (!isdigit(*q)) return 0;

    char* end = NULL;
    errno = 0;
    long long v = strtoll(s, &end, 10);
    if (errno == ERANGE || end == s || *end != '\0' ||
        (nonnegative && v < 0) || (core_count && v > INT_MAX))
        return 0;
    *out = (int64_t)v;
    return 1;
}

/* ══════════════════════════════════════════
 * Serialization / storage
 * ══════════════════════════════════════════ */

/* (ser val) -> serialize to U8 vector with IPC header */
ray_t* ray_ser_fn(ray_t* val) {
    return ray_ser(val);
}

/* (de bytes) -> deserialize from U8 vector */
ray_t* ray_de_fn(ray_t* val) {
    return ray_de(val);
}

/* True when `name` (n bytes) is partition-shaped: nonempty, digits and
 * dots only — the same loose classification collect_part_dirs
 * (store/part.c) applies to partition directory names. */
static bool sym_partition_shaped(const char* name, size_t n) {
    if (n == 0) return false;
    for (size_t i = 0; i < n; i++)
        if (name[i] != '.' && (name[i] < '0' || name[i] > '9')) return false;
    return true;
}

/* Symfile resolution (sym-domain spec, "Surface"): for a partition dir
 * (/db/2024.01.01/t/) the table root is the PARTED ROOT — derived from
 * the path shape — and the domain is root/.sym; for a standalone splayed
 * dir the root is the dir itself (dir/.sym).  The symfile is a dotfile so
 * it can never collide with a user column (e.g. a "sym" ticker column).
 * Writes default to the convention unconditionally; reads prefer an
 * existing dir/.sym, then an existing partition root/.sym, else NULL (the
 * load layer raises the
 * loud "sym" error only if SYM columns actually exist). */
static const char* splay_resolve_sym(const char* dir, char* buf, size_t bufsz,
                                     bool for_write) {
    size_t dlen = strlen(dir);
    while (dlen > 1 && dir[dlen - 1] == '/') dlen--; /* strip trailing '/' */

    /* dir/.sym */
    int n = snprintf(buf, bufsz, "%.*s/.sym", (int)dlen, dir);
    if (n < 0 || (size_t)n >= bufsz) return NULL;
    if (!for_write && access(buf, F_OK) == 0) return buf;

    /* partition-shaped parent → parted root's root/sym */
    const char* slash = NULL;
    for (size_t i = dlen; i > 0; i--)
        if (dir[i - 1] == '/') { slash = dir + i - 1; break; }
    if (slash && slash > dir) {
        size_t parent_end = (size_t)(slash - dir);
        const char* pslash = NULL;
        for (size_t i = parent_end; i > 0; i--)
            if (dir[i - 1] == '/') { pslash = dir + i - 1; break; }
        const char* pname = pslash ? pslash + 1 : dir;
        size_t pname_len = (size_t)(dir + parent_end - pname);
        if (sym_partition_shaped(pname, pname_len) && pslash) {
            size_t root_len = (size_t)(pslash - dir);
            if (root_len == 0) root_len = 1; /* "/" root */
            int rn = snprintf(buf, bufsz, "%.*s/.sym", (int)root_len, dir);
            if (rn < 0 || (size_t)rn >= bufsz) return NULL;
            if (for_write || access(buf, F_OK) == 0) return buf;
            return NULL;
        }
    }

    if (for_write) {
        /* Not partition-shaped: the dir itself is the table root. */
        n = snprintf(buf, bufsz, "%.*s/.sym", (int)dlen, dir);
        if (n < 0 || (size_t)n >= bufsz) return NULL;
        return buf;
    }
    return NULL; /* read: no symfile resolvable */
}

/* Helper: extract null-terminated path from a STR atom into a stack buffer.
 * Returns pointer to buf on success, NULL on failure. */
static const char* str_to_cpath(ray_t* s, char* buf, size_t bufsz) {
    if (!s || s->type != -RAY_STR) return NULL;
    const char* p = ray_str_ptr(s);
    size_t len = ray_str_len(s);
    if (!p || len == 0 || len >= bufsz) return NULL;
    memcpy(buf, p, len);
    buf[len] = '\0';
    return buf;
}

/* (.db.splayed.set "dir" table) or (.db.splayed.set "dir" table "sym_path") */
ray_t* ray_set_splayed_fn(ray_t** args, int64_t n) {
    if (n < 2 || n > 3)
        return ray_error("domain", ".db.splayed.set expects 2 or 3 arguments, got %lld", (long long)n);

    char dir[1024];
    if (!str_to_cpath(args[0], dir, sizeof(dir)))
        return ray_error("type", ".db.splayed.set expects a string dir path, got %s", ray_type_name(args[0]->type));

    ray_t* tbl = args[1];
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", ".db.splayed.set expects a table, got %s", ray_type_name(tbl->type));

    char sym[1024];
    const char* sym_path = NULL;
    if (n == 3 && args[2] && args[2]->type == -RAY_STR)
        sym_path = str_to_cpath(args[2], sym, sizeof(sym));
    else
        sym_path = splay_resolve_sym(dir, sym, sizeof(sym), true);

    ray_err_t err = ray_splay_save(tbl, dir, sym_path);
    if (err != RAY_OK) return ray_error(ray_err_code_str(err), NULL);

    ray_retain(tbl);
    return tbl;
}

/* (.db.splayed.get "dir") or (.db.splayed.get "dir" "sym_path") */
ray_t* ray_get_splayed_fn(ray_t** args, int64_t n) {
    if (n < 1 || n > 2)
        return ray_error("domain", ".db.splayed.get expects 1 or 2 arguments, got %lld", (long long)n);

    char dir[1024];
    if (!str_to_cpath(args[0], dir, sizeof(dir)))
        return ray_error("type", ".db.splayed.get expects a string dir path, got %s", ray_type_name(args[0]->type));

    char sym[1024];
    const char* sym_path = NULL;
    if (n == 2 && args[1] && args[1]->type == -RAY_STR)
        sym_path = str_to_cpath(args[1], sym, sizeof(sym));
    else
        sym_path = splay_resolve_sym(dir, sym, sizeof(sym), false);

    /* sym_path == NULL: no symfile resolvable (no explicit arg, no
     * dir/sym, no partition root/sym).  The load layer raises the loud
     * "sym" error if the table actually has SYM columns; symbol-free
     * tables load fine without one. */
    return ray_read_splayed(dir, sym_path);
}

/* (.db.parted.get "db_root" `table_name) -- load partitioned table */
ray_t* ray_get_parted_fn(ray_t** args, int64_t n) {
    if (n != 2)
        return ray_error("domain", ".db.parted.get expects 2 arguments, got %lld", (long long)n);

    char root[1024];
    if (!str_to_cpath(args[0], root, sizeof(root)))
        return ray_error("type", ".db.parted.get expects a string db root path, got %s", ray_type_name(args[0]->type));

    /* Table name as symbol atom */
    if (!args[1] || args[1]->type != -RAY_SYM)
        return ray_error("type", ".db.parted.get expects a sym table name, got %s", ray_type_name(args[1]->type));
    ray_t* name_atom = ray_sym_str(args[1]->i64);
    if (!name_atom) return ray_error("name", NULL);

    char name[256];
    size_t nlen = ray_str_len(name_atom);
    if (nlen == 0 || nlen >= sizeof(name))
        return ray_error("domain", ".db.parted.get table name length out of range, got %lld", (long long)nlen);
    memcpy(name, ray_str_ptr(name_atom), nlen);
    name[nlen] = '\0';

    return ray_read_parted(root, name);
}

/* (.db.parted.tables "db_root") → sym vector of table names available under
 * the root, suitable for passing to .db.parted.get. */
ray_t* ray_get_parted_tables_fn(ray_t** args, int64_t n) {
    if (n != 1)
        return ray_error("domain", ".db.parted.tables expects 1 argument, got %lld", (long long)n);

    char root[1024];
    if (!str_to_cpath(args[0], root, sizeof(root)))
        return ray_error("type", ".db.parted.tables expects a string db root path, got %s", ray_type_name(args[0]->type));

    return ray_parted_tables(root);
}

/* (.db.parted.fill "db_root") → sym vector of partition names that were
 * filled.  Writes an empty copy of every table into the partitions that lack
 * it (schema from the most recent partition with it). */
ray_t* ray_fill_parted_fn(ray_t** args, int64_t n) {
    if (n != 1)
        return ray_error("domain", ".db.parted.fill expects 1 argument, got %lld", (long long)n);

    char root[1024];
    if (!str_to_cpath(args[0], root, sizeof(root)))
        return ray_error("type", ".db.parted.fill expects a string db root path, got %s", ray_type_name(args[0]->type));

    return ray_parted_fill(root);
}

/* stat/dirent used by the .fs.* filesystem metadata builtins below. */
#include <sys/stat.h>
#include <dirent.h>

/* ══════════════════════════════════════════
 * Filesystem metadata: .fs.size / .fs.list
 *
 * Issue #36 asked for size + existence + listing primitives.  We
 * keep just two — `.fs.size` and `.fs.list` — because every other
 * predicate (exists, is-file, is-dir) is reachable either via
 * try-on-error against these or via the existing shell fallback
 * (`(.sys.cmd "test -e p")` etc.).  Both errors are flagged "io"
 * so a user wrapping the call in `try` can distinguish missing /
 * wrong-kind from a domain mistake without introspecting the
 * message.
 * ══════════════════════════════════════════ */

/* (.fs.size "path") → i64 file size in bytes.  Errors with "io"
 * when the path doesn't exist or names a directory — `try` it if
 * the caller wants those treated as "not a file" rather than a
 * hard error. */
ray_t* ray_fs_size_fn(ray_t* x) {
    if (!ray_is_atom(x) || x->type != -RAY_STR)
        return ray_error("type", ".fs.size expects a string path");
    char path[1024];
    /* x is already a STR atom here; str_to_cpath fails only when the path
     * is empty or exceeds the buffer — a domain/range issue, not a type
     * mismatch (old code: "type" → "domain"). */
    if (!str_to_cpath(x, path, sizeof(path)))
        return ray_error("domain", ".fs.size path is empty or too long, got %lld bytes", (long long)ray_str_len(x));

    struct stat st;
    if (stat(path, &st) != 0)
        return ray_error("io", "%s: %s", path, strerror(errno));
    if (S_ISDIR(st.st_mode))
        return ray_error("io", "%s: is a directory", path);
    return ray_i64((int64_t)st.st_size);
}

/* qsort comparator for sorting directory entries by name.  Filesystem
 * order from readdir is implementation-defined; sorting gives stable
 * output for tests and predictable iteration in user code. */
static int dir_entry_cmp(const void* a, const void* b) {
    const char* sa = *(const char* const*)a;
    const char* sb = *(const char* const*)b;
    return strcmp(sa, sb);
}

/* (.fs.list "path") → sym vec of entries, sorted, with `.` and `..`
 * filtered out.  Errors with "io" if the path isn't a directory or
 * doesn't exist — caller can use that as a file/dir discriminator
 * via `try` when they don't want to shell out for the predicate. */
ray_t* ray_fs_list_fn(ray_t* x) {
    if (!ray_is_atom(x) || x->type != -RAY_STR)
        return ray_error("type", ".fs.list expects a string path");
    char path[1024];
    /* x is already a STR atom here; str_to_cpath fails only when the path
     * is empty or exceeds the buffer — a domain/range issue, not a type
     * mismatch (old code: "type" → "domain"). */
    if (!str_to_cpath(x, path, sizeof(path)))
        return ray_error("domain", ".fs.list path is empty or too long, got %lld bytes", (long long)ray_str_len(x));

    DIR* d = opendir(path);
    if (!d) return ray_error("io", "%s: %s", path, strerror(errno));

    /* Collect names into a sys-allocator string array; capacity grows
     * geometrically so big directories don't quadratic-realloc.  Uses
     * ray_sys_alloc/realloc/strdup/free (NOT libc) per project policy —
     * mirrors the collect_part_dirs idiom in src/store/part.c. */
    char** names = NULL;
    int64_t count = 0;
    int64_t cap = 0;
    struct dirent* ent;
    while ((ent = readdir(d)) != NULL) {
        if (ent->d_name[0] == '.' &&
            (ent->d_name[1] == '\0' || (ent->d_name[1] == '.' && ent->d_name[2] == '\0')))
            continue;
        if (count >= cap) {
            int64_t new_cap = cap == 0 ? 16 : cap * 2;
            char** tmp = (char**)ray_sys_realloc(names, (size_t)new_cap * sizeof(char*));
            if (!tmp) {
                closedir(d);
                for (int64_t i = 0; i < count; i++) ray_sys_free(names[i]);
                ray_sys_free(names);
                return ray_error("oom", NULL);
            }
            names = tmp;
            cap = new_cap;
        }
        char* dup = ray_sys_strdup(ent->d_name);
        if (!dup) {
            closedir(d);
            for (int64_t i = 0; i < count; i++) ray_sys_free(names[i]);
            ray_sys_free(names);
            return ray_error("oom", NULL);
        }
        names[count++] = dup;
    }
    closedir(d);

    /* qsort(NULL, 0, ...) is undefined behaviour per C11 §7.22.5.2/2
     * even when nmemb == 0 (UBSan flags it via qsort's nonnull attr).
     * Skip the call when the directory was empty — an empty array is
     * already sorted. */
    if (count > 0)
        qsort(names, (size_t)count, sizeof(char*), dir_entry_cmp);

    ray_t* result = ray_vec_new(RAY_SYM, count);
    if (!result || RAY_IS_ERR(result)) {
        for (int64_t i = 0; i < count; i++) ray_sys_free(names[i]);
        ray_sys_free(names);
        return result ? result : ray_error("oom", NULL);
    }
    result->len = count;
    int64_t* out = (int64_t*)ray_data(result);
    for (int64_t i = 0; i < count; i++) {
        out[i] = ray_sym_intern(names[i], strlen(names[i]));
        ray_sys_free(names[i]);
    }
    ray_sys_free(names);
    return result;
}

static int64_t ray_os_pid(void);   /* defined with the .sys.info helpers */

/* Eight bytes of OS entropy.  False when no source answered, which the
 * caller treats as "fold in less", never as "seed with a constant". */
static bool guid_os_entropy(uint64_t* out) {
#if defined(RAY_OS_WINDOWS)
    (void)out;
    return false;   /* covered by the process/time material below */
#else
    FILE* f = fopen("/dev/urandom", "rb");
    if (!f) return false;
    size_t got = fread(out, 1, sizeof(*out), f);
    fclose(f);
    return got == sizeof(*out);
#endif
}

/* Seed for one thread's GUID stream.
 *
 * This used to come from rand(), on the stated assumption that "rand() is
 * itself seeded by the runtime".  Nothing in the tree has ever called
 * srand(), so rand() ran from libc's default seed of 1 and every process
 * produced the same GUID sequence — the reporter's first value on their
 * machine and build was byte-identical to ours (#571).  Uniqueness is the
 * one property the type exists to provide, and the failure was silent:
 * the values are well-formed and carry the correct version-4 nibble.
 *
 * Seeded from the OS independently of rand(), so rand()'s determinism —
 * which is defensible, and which reproducible tests rely on — is left
 * exactly as it was.  Process id, a high-resolution clock and a stack
 * address are always folded in, so even a platform or a chroot where the
 * entropy source cannot be reached still cannot hand two processes, or
 * two threads, the same stream. */
static uint64_t guid_seed(void) {
    uint64_t s = 0;
    if (!guid_os_entropy(&s)) s = 0;

    s ^= (uint64_t)ray_os_pid() * 0x9E3779B97F4A7C15ULL;
    s ^= (uint64_t)(uintptr_t)&s;
#if defined(RAY_OS_WINDOWS)
    LARGE_INTEGER cnt;
    QueryPerformanceCounter(&cnt);
    s ^= (uint64_t)cnt.QuadPart * 0xBF58476D1CE4E5B9ULL;
    FILETIME ft;
    GetSystemTimeAsFileTime(&ft);
    s ^= ((uint64_t)ft.dwHighDateTime << 32) | ft.dwLowDateTime;
#else
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) == 0)
        s ^= ((uint64_t)ts.tv_sec * 1000000000ULL + (uint64_t)ts.tv_nsec) * 0xBF58476D1CE4E5B9ULL;
    if (clock_gettime(CLOCK_REALTIME, &ts) == 0)
        s ^= (uint64_t)ts.tv_sec * 0x94D049BB133111EBULL ^ (uint64_t)ts.tv_nsec;
#endif

    /* splitmix64 finaliser: the folded material is structured (a small pid,
     * a slow-moving clock), and xorshift64* needs its state well mixed or
     * the first few outputs stay correlated with it. */
    s ^= s >> 30; s *= 0xBF58476D1CE4E5B9ULL;
    s ^= s >> 27; s *= 0x94D049BB133111EBULL;
    s ^= s >> 31;
    return s ? s : 0x9E3779B97F4A7C15ULL;
}

/* xorshift64* — ~1ns per 64-bit word, vs rand()'s ~10ns for 1 byte.
 * Per-thread state seeded once from guid_seed().  v4 UUID quality only
 * requires the version and variant nibbles to be correct; the remaining
 * 122 bits are pseudo-random and xorshift64* is more than sufficient. */
static __thread uint64_t guid_rng_state = 0;

static inline uint64_t guid_rng_next(void) {
    uint64_t x = guid_rng_state;
    if (RAY_UNLIKELY(x == 0)) x = guid_seed();
    x ^= x >> 12;
    x ^= x << 25;
    x ^= x >> 27;
    guid_rng_state = x;
    return x * 0x2545F4914F6CDD1DULL;
}

/* (guid n) -> generate n random GUIDs as GUID vector.
 * v4 UUID format: 122 random bits + 4 version-bits (0100) + 2 variant-bits (10). */
ray_t* ray_guid_fn(ray_t* n_arg) {
    if (!n_arg || !is_numeric(n_arg))
        return ray_error("type", "guid expects a numeric count, got %s", ray_type_name(n_arg->type));
    int64_t n = as_i64(n_arg);
    if (n < 0) return ray_error("domain", "guid count must be non-negative, got %lld", (long long)n);
    ray_t* result = ray_vec_new(RAY_GUID, n);
    if (RAY_IS_ERR(result)) return result;
    result->len = n;
    uint8_t* data = (uint8_t*)ray_data(result);
    /* Two 64-bit RNG calls per UUID give 16 random bytes; then we just
     * stamp the version/variant nibbles. */
    for (int64_t i = 0; i < n; i++) {
        uint64_t lo = guid_rng_next();
        uint64_t hi = guid_rng_next();
        memcpy(data + i * 16, &lo, 8);
        memcpy(data + i * 16 + 8, &hi, 8);
        data[i * 16 + 6] = (data[i * 16 + 6] & 0x0F) | 0x40;  /* version 4 */
        data[i * 16 + 8] = (data[i * 16 + 8] & 0x3F) | 0x80;  /* variant 10 */
    }
    return result;
}

/* ══════════════════════════════════════════
 * Eval, parse, print, system, env builtins
 * ══════════════════════════════════════════ */

/* (eval expr) -- evaluate a parsed expression */
ray_t* ray_eval_builtin_fn(ray_t* x) {
    return ray_eval(x);
}

/* (parse str) -- parse a string into an AST */
ray_t* ray_parse_builtin_fn(ray_t* x) {
    if (x->type != -RAY_STR) return ray_error("type", "parse expects a string");
    const char* src = ray_str_ptr(x);
    if (!src) return ray_error("domain", "parse: empty or invalid source string");
    ray_t* parsed = ray_parse(src);
    return parsed ? parsed : ray_error("parse", NULL);
}

/* (print val) -- print without newline, return the value */
/* print moved to builtins.c alongside println/show */

/* (meta x) -- return metadata about an object as a dict */
ray_t* ray_meta_fn(ray_t* x) {
    /* x is NULL here, so no ->type to report. */
    if (!x) return ray_error("type", "meta expects a value");

    const char* tname = ray_type_name(x->type);
    int64_t type_sym = ray_sym_intern("type", 4);
    int64_t type_id  = ray_sym_intern(tname, strlen(tname));

    /* Build keys SYM vec + vals LIST. */
    int64_t cap = ray_is_atom(x) ? 1 : 2;
    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, cap);
    if (RAY_IS_ERR(keys)) return keys;
    ray_t* vals = ray_list_new(cap);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }

    keys = ray_vec_append(keys, &type_sym);
    if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
    ray_t* tv = ray_sym(type_id);
    vals = ray_list_append(vals, tv);
    ray_release(tv);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }

    if (!ray_is_atom(x)) {
        int64_t len_sym = ray_sym_intern("len", 3);
        keys = ray_vec_append(keys, &len_sym);
        if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
        int64_t row_count;
        if (x->type == RAY_DICT)       row_count = ray_dict_len(x);
        else if (x->type == RAY_TABLE) row_count = ray_table_ncols(x);
        else                            row_count = x->len;
        ray_t* lv = make_i64(row_count);
        vals = ray_list_append(vals, lv);
        ray_release(lv);
        if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
    }

    return ray_dict_new(keys, vals);
}

/* Object-graph size -------------------------------------------------------
 *
 * `.mem.objsize` reports the logical bytes retained by a Rayfall value:
 * every reachable ray_t header and its live payload, with shared children
 * counted once.  It deliberately does not report buddy-block slack, global
 * symbol dictionaries, or opaque native handles (graphs/HNSW/lazy plans).
 * That makes the result stable across allocator tuning and useful for
 * inspecting materialised query results.
 *
 * The traversal is iterative and pointer-deduplicated.  Besides preventing
 * double-counting shared table columns, the visited set makes it safe for any
 * future compound object that introduces a cycle. */
typedef struct {
    ray_t** stack;
    size_t  stack_len;
    size_t  stack_cap;
    ray_t** seen;
    size_t  seen_count;
    size_t  seen_cap;
} ray_objsize_walk_t;

static uint64_t objsize_ptr_hash(const ray_t* p) {
    uint64_t x = (uint64_t)(uintptr_t)p;
    x ^= x >> 33;
    x *= UINT64_C(0xff51afd7ed558ccd);
    x ^= x >> 33;
    x *= UINT64_C(0xc4ceb9fe1a85ec53);
    return x ^ (x >> 33);
}

static bool objsize_stack_push(ray_objsize_walk_t* w, ray_t* child) {
    if (!child || RAY_IS_ERR(child)) return true;
    if (w->stack_len == w->stack_cap) {
        size_t cap = w->stack_cap ? w->stack_cap * 2 : 64;
        if (cap < w->stack_cap || cap > SIZE_MAX / sizeof(ray_t*)) return false;
        ray_t** grown = (ray_t**)ray_sys_realloc(w->stack, cap * sizeof(ray_t*));
        if (!grown) return false;
        w->stack = grown;
        w->stack_cap = cap;
    }
    w->stack[w->stack_len++] = child;
    return true;
}

static bool objsize_seen_grow(ray_objsize_walk_t* w) {
    size_t cap = w->seen_cap ? w->seen_cap * 2 : 128;
    if (cap < w->seen_cap || cap > SIZE_MAX / sizeof(ray_t*)) return false;
    ray_t** grown = (ray_t**)ray_sys_alloc(cap * sizeof(ray_t*));
    if (!grown) return false;
    memset(grown, 0, cap * sizeof(ray_t*));

    if (w->seen) {
        size_t mask = cap - 1;
        for (size_t i = 0; i < w->seen_cap; i++) {
            ray_t* p = w->seen[i];
            if (!p) continue;
            size_t slot = (size_t)objsize_ptr_hash(p) & mask;
            while (grown[slot]) slot = (slot + 1) & mask;
            grown[slot] = p;
        }
        ray_sys_free(w->seen);
    }
    w->seen = grown;
    w->seen_cap = cap;
    return true;
}

/* Returns true when p was newly inserted, false when already seen.  OOM is
 * reported separately through *ok so a duplicate is not confused with an
 * allocation failure. */
static bool objsize_seen_insert(ray_objsize_walk_t* w, ray_t* p, bool* ok) {
    if (!w->seen_cap || (w->seen_count + 1) * 10 >= w->seen_cap * 7) {
        if (!objsize_seen_grow(w)) { *ok = false; return false; }
    }
    size_t mask = w->seen_cap - 1;
    size_t slot = (size_t)objsize_ptr_hash(p) & mask;
    while (w->seen[slot]) {
        if (w->seen[slot] == p) return false;
        slot = (slot + 1) & mask;
    }
    w->seen[slot] = p;
    w->seen_count++;
    return true;
}

static size_t objsize_shallow(ray_t* v) {
    if (v->attrs & RAY_ATTR_SLICE) return sizeof(ray_t);
    if (v->type == RAY_LAMBDA)
        return sizeof(ray_t) + 7 * sizeof(ray_t*);
    if (v->type == RAY_INDEX)
        return sizeof(ray_t) + sizeof(ray_index_t);
    if (RAY_IS_PARTED(v->type))
        return sizeof(ray_t) + (v->len > 0 ? (size_t)v->len * sizeof(ray_t*) : 0);
    if (v->type == RAY_MAPCOMMON)
        return sizeof(ray_t) + 2 * sizeof(ray_t*);
    return ray_block_size(v);
}

static bool objsize_push_index_children(ray_objsize_walk_t* w, ray_index_t* ix) {
#define OBJSIZE_PUSH(child) do { if (!objsize_stack_push(w, (child))) return false; } while (0)
    switch ((ray_idx_kind_t)ix->kind) {
    case RAY_IDX_UKEY:       OBJSIZE_PUSH(ix->u.ukey.slots); break;
    case RAY_IDX_HASH:
        OBJSIZE_PUSH(ix->u.hash.table); OBJSIZE_PUSH(ix->u.hash.gkeys);
        OBJSIZE_PUSH(ix->u.hash.offs);  OBJSIZE_PUSH(ix->u.hash.rows);
        break;
    case RAY_IDX_SORT:       OBJSIZE_PUSH(ix->u.sort.perm); break;
    case RAY_IDX_BLOOM:      OBJSIZE_PUSH(ix->u.bloom.bits); break;
    case RAY_IDX_CHUNK_ZONE:
        OBJSIZE_PUSH(ix->u.chunk_zone.mins);
        OBJSIZE_PUSH(ix->u.chunk_zone.maxs);
        OBJSIZE_PUSH(ix->u.chunk_zone.null_bits);
        OBJSIZE_PUSH(ix->u.chunk_zone.aggs);
        break;
    case RAY_IDX_PART:
        OBJSIZE_PUSH(ix->u.part.keys); OBJSIZE_PUSH(ix->u.part.starts);
        OBJSIZE_PUSH(ix->u.part.lens);
        break;
    case RAY_IDX_DICT:
        OBJSIZE_PUSH(ix->u.dict.codes); OBJSIZE_PUSH(ix->u.dict.first_occ);
        break;
    case RAY_IDX_ZONE:
    case RAY_IDX_NONE:
        break;
    }
#undef OBJSIZE_PUSH
    return true;
}

static bool objsize_push_children(ray_objsize_walk_t* w, ray_t* v) {
#define OBJSIZE_PUSH(child) do { if (!objsize_stack_push(w, (child))) return false; } while (0)
    if (v->type == RAY_LAMBDA) {
        ray_t** slots = (ray_t**)ray_data(v);
        for (int i = 0; i < 4; i++) OBJSIZE_PUSH(slots[i]);
        OBJSIZE_PUSH(slots[5]);
        OBJSIZE_PUSH(slots[6]);
        return true;
    }
    if (ray_is_atom(v)) {
        if (v->type == -RAY_GUID && v->obj) OBJSIZE_PUSH(v->obj);
        if (v->type == -RAY_STR &&
            !((v->slen >= 1 && v->slen <= 7) || (v->slen == 0 && !v->obj)))
            OBJSIZE_PUSH(v->obj);
        return true;
    }
    if (v->attrs & RAY_ATTR_SLICE) {
        OBJSIZE_PUSH(v->slice_parent);
        return true;
    }
    if (v->type == RAY_INDEX)
        return objsize_push_index_children(w, ray_index_payload(v));
    if (v->attrs & RAY_ATTR_HAS_INDEX) {
        OBJSIZE_PUSH(v->index);
        /* An indexed STR vector still owns its external character pool in
         * aux bytes 8..15; the index pointer only occupies bytes 0..7. */
        if (v->type == RAY_STR && v->str_pool)
            OBJSIZE_PUSH(v->str_pool);
        /* A TABLE carries the keyed-upsert map here, but its schema and
         * columns are still its own children: returning now would report a
         * mapped table as nothing but its map.  Fall through to the arm
         * below.  Only a vector has nothing further to walk. */
        if (v->type != RAY_TABLE) return true;
    }
    if (v->type == RAY_STR && v->str_pool)
        OBJSIZE_PUSH(v->str_pool);
    if (RAY_IS_PARTED(v->type)) {
        ray_t** segs = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < v->len; i++) OBJSIZE_PUSH(segs[i]);
        return true;
    }
    if (v->type == RAY_MAPCOMMON || v->type == RAY_TABLE || v->type == RAY_DICT) {
        ray_t** slots = (ray_t**)ray_data(v);
        OBJSIZE_PUSH(slots[0]);
        OBJSIZE_PUSH(slots[1]);
        return true;
    }
    if (v->type == RAY_LIST) {
        ray_t** elems = (ray_t**)ray_data(v);
        for (int64_t i = 0; i < v->len; i++) OBJSIZE_PUSH(elems[i]);
    }
#undef OBJSIZE_PUSH
    return true;
}

ray_t* ray_mem_objsize_fn(ray_t* root) {
    if (!root) return ray_error("type", "objsize expects a value");
    ray_objsize_walk_t w = {0};
    if (!objsize_stack_push(&w, root)) return ray_error("oom", NULL);

    uint64_t total = 0;
    bool ok = true;
    while (w.stack_len && ok) {
        ray_t* v = w.stack[--w.stack_len];
        if (!objsize_seen_insert(&w, v, &ok)) continue;
        size_t shallow = objsize_shallow(v);
        if (UINT64_MAX - total < shallow) { ok = false; break; }
        total += shallow;
        if (!objsize_push_children(&w, v)) ok = false;
    }
    ray_sys_free(w.stack);
    ray_sys_free(w.seen);
    if (!ok) return ray_error("oom", "objsize traversal overflow or allocation failure");
    if (total > INT64_MAX) return ray_error("limit", "objsize exceeds I64 range");
    return ray_i64((int64_t)total);
}

/* (.mem.ts expr) — evaluate one expression under a process-wide allocation
 * trace and return a named measurement dictionary.  SPECIAL_FORM: expr is
 * deliberately not pre-evaluated.  Result sizing and dictionary construction
 * happen after both clocks stop, so they do not contaminate the measured
 * expression. */
ray_t* ray_mem_ts_fn(ray_t** args, int64_t n) {
    if (n != 1)
        return ray_error("arity", ".mem.ts expects 1 argument, got %lld",
                         (long long)n);
    if (!ray_mem_trace_begin())
        return ray_error("state", ".mem.ts measurement already active");

    ray_query_measure_t measure;
    ray_query_measure_begin(&measure);
    ray_t* result = ray_eval(args[0]);
    ray_query_metrics_t metrics;
    ray_query_measure_end(&measure, &metrics);

    ray_mem_trace_t mt;
    ray_mem_trace_end(&mt);

    if (!result) return ray_error("state", ".mem.ts expression returned no value");
    if (RAY_IS_ERR(result)) return result;

    ray_t* result_size = ray_mem_objsize_fn(result);
    if (!result_size || RAY_IS_ERR(result_size)) {
        ray_release(result);
        return result_size ? result_size : ray_error("oom", NULL);
    }

    enum { MEM_TS_FIELDS = 13 };
    static const char* names[MEM_TS_FIELDS] = {
        "result", "time-ns", "memory-bytes", "allocated-bytes", "freed-bytes",
        "net-bytes", "peak-live-bytes", "result-bytes", "alloc-count",
        "free-count", "workers", "worker-busy-ns", "parallelism"
    };
    static const uint8_t name_lens[MEM_TS_FIELDS] = {
        6, 7, 12, 15, 11, 9, 15, 12, 11, 10, 7, 14, 11
    };

    ray_t* items[MEM_TS_FIELDS] = {
        result,
        ray_i64(metrics.time_ns),
        ray_i64(metrics.memory_bytes),
        ray_i64((int64_t)mt.allocated_bytes),
        ray_i64((int64_t)mt.freed_bytes),
        ray_i64(mt.net_bytes),
        ray_i64((int64_t)mt.peak_live_bytes),
        result_size,
        ray_i64((int64_t)mt.alloc_count),
        ray_i64((int64_t)mt.free_count),
        ray_i64((int64_t)metrics.workers),
        ray_i64((int64_t)metrics.worker_busy_ns),
        ray_f64(metrics.parallelism)
    };
    for (int i = 0; i < MEM_TS_FIELDS; i++) {
        if (!items[i] || RAY_IS_ERR(items[i])) {
            ray_t* err = items[i] ? items[i] : ray_error("oom", NULL);
            for (int j = 0; j < MEM_TS_FIELDS; j++)
                if (j != i && items[j] && !RAY_IS_ERR(items[j])) ray_release(items[j]);
            return err;
        }
    }

    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, MEM_TS_FIELDS);
    ray_t* vals = ray_list_new(MEM_TS_FIELDS);
    if (!keys || RAY_IS_ERR(keys) || !vals || RAY_IS_ERR(vals)) {
        ray_t* err = (!keys || RAY_IS_ERR(keys))
                   ? (keys ? keys : ray_error("oom", NULL))
                   : vals;
        if (keys && !RAY_IS_ERR(keys)) ray_release(keys);
        if (vals && !RAY_IS_ERR(vals)) ray_release(vals);
        for (int i = 0; i < MEM_TS_FIELDS; i++) ray_release(items[i]);
        return err;
    }

    for (int i = 0; i < MEM_TS_FIELDS; i++) {
        int64_t sid = ray_sym_intern(names[i], name_lens[i]);
        keys = ray_vec_append(keys, &sid);
        if (RAY_IS_ERR(keys)) {
            ray_release(vals);
            for (int j = i; j < MEM_TS_FIELDS; j++) ray_release(items[j]);
            return keys;
        }
        vals = ray_list_append(vals, items[i]);
        ray_release(items[i]);
        if (RAY_IS_ERR(vals)) {
            ray_release(keys);
            for (int j = i + 1; j < MEM_TS_FIELDS; j++) ray_release(items[j]);
            return vals;
        }
    }
    return ray_dict_new(keys, vals);
}

/* (.sys.gc) -- perform the allocator maintenance pass used by v1: drain
 * foreign frees, flush slab caches, coalesce blocks and release reclaimable
 * pages/pools.  Rayforce values are reference-counted, so this is allocator
 * GC rather than a tracing collector.  Variadic to allow `(.sys.gc)`. */
ray_t* ray_gc_fn(ray_t** args, int64_t n) {
    (void)args;
    if (n != 0) return ray_error("arity", ".sys.gc takes no arguments");
    ray_heap_gc();
    /* Same statement-boundary rule as the REPL: an explicit maintenance
     * call is also a chance to notice the process has gone quiet. */
    ray_heap_decay();
    return ray_i64(0);
}

/* (system cmd) -- run shell command, return exit code */
/* Both helpers exist only for the live .sys.exec path, which RAY_FUZZING
 * compiles out — guard them so the fuzz build does not trip
 * -Werror,-Wunused-function. */
#ifndef RAY_FUZZING
/* Turn a wait(2)-style status into the exit code a shell would report.
 * system() and pclose() hand back an encoded status, not a code: "exit 3"
 * arrives as 768 (3 << 8).  Signals follow the shell's 128+N convention, so
 * a command killed by SIGTERM reads as 143 rather than as a bare 15 that
 * could be confused with a real exit status. */
static int64_t exec_exit_code(int status) {
#if defined(RAY_OS_WINDOWS)
    /* The CRT's system()/_pclose() already return the child's exit code. */
    return (int64_t)status;
#else
    if (WIFEXITED(status))   return (int64_t)WEXITSTATUS(status);
    if (WIFSIGNALED(status)) return (int64_t)(128 + WTERMSIG(status));
    return (int64_t)status;
#endif
}

/* Read everything the child writes to stdout.  Grows geometrically rather
 * than sizing from any stat(), so output from a pipe — which has no size to
 * ask for — comes back whole (cf. #572). */
static ray_t* exec_capture(const char* cmd, int64_t* code_out) {
    FILE* fp = RAY_POPEN(cmd, "r");
    if (!fp) return ray_error("io", ".sys.exec: failed to start command: %s", strerror(errno));

    /* A command that never stops writing has no end to capture, exactly as an
     * endless file has no EOF to read (#572).  Same ceiling, same reason. */
    const int64_t budget = ray_unsized_read_budget();

    size_t cap = 4096, len = 0;
    char*  buf = (char*)ray_alloc_raw(cap);
    if (!buf) { RAY_PCLOSE(fp); return ray_error("oom", ".sys.exec: capture buffer"); }

    for (;;) {
        if (len == cap) {
            /* Peak during the grow is the old buffer plus the new one. */
            if (budget > 0 && (int64_t)cap * 3 > budget) {
                ray_free_raw(buf); RAY_PCLOSE(fp);
                return ray_error("io", ".sys.exec: output exceeds the readable budget of %lld bytes",
                                 (long long)budget);
            }
            char* nbuf = (char*)ray_realloc_raw(buf, cap * 2);
            if (!nbuf) { ray_free_raw(buf); RAY_PCLOSE(fp); return ray_error("oom", ".sys.exec: capture buffer"); }
            buf = nbuf; cap *= 2;
        }
        size_t got = fread(buf + len, 1, cap - len, fp);
        len += got;
        if (got == 0) {
            if (ferror(fp)) { ray_free_raw(buf); RAY_PCLOSE(fp); return ray_error("io", ".sys.exec: read failed"); }
            break;  /* EOF */
        }
    }

    int status = RAY_PCLOSE(fp);
    if (status == -1) { ray_free_raw(buf); return ray_error("io", ".sys.exec: command did not complete: %s", strerror(errno)); }
    *code_out = exec_exit_code(status);

    ray_t* out = ray_str(buf, len);
    ray_free_raw(buf);
    return out;
}

#endif /* !RAY_FUZZING */

/* (.sys.exec cmd)      -> exit code
 * (.sys.exec cmd 'out) -> {code: <exit code> out: <stdout>}
 *
 * The capture reads to EOF on the child's stdout, and a backgrounded
 * grandchild inherits that pipe — so 'out waits for it too, not just for the
 * shell (#579).  Inherent to popen(3); the one-argument form does not have
 * it, and is the way to launch something that outlives the call.
 *
 * stderr is deliberately left on the process's own stderr: a shelled-out
 * command's diagnostics belong in the log, and merging them into the
 * captured value would corrupt anything parsing that value (#573). */
ray_t* ray_system_fn(ray_t** args, int64_t n) {
#ifdef RAY_FUZZING
    /* Shell escape — disabled under fuzzing so untrusted input can't run
     * arbitrary commands in the fuzzer process. */
    (void)args; (void)n;
    return ray_error("restricted", "shell disabled under fuzzing");
#else
    if (n != 1 && n != 2) return ray_error("domain", ".sys.exec expects 1 or 2 arguments");
    ray_t* x = args[0];
    if (x->type != -RAY_STR) return ray_error("type", "system expects a string");

    bool capture = false;
    if (n == 2) {
        ray_t* mode = args[1];
        if (mode->type != -RAY_SYM)
            return ray_error("type", ".sys.exec: second argument must be the symbol 'out");
        if (mode->i64 != ray_sym_intern("out", 3))
            return ray_error("domain", ".sys.exec: unknown mode, expected 'out");
        capture = true;
    }

    /* libc needs a NUL-terminated command; ray strings are not NUL-terminated
     * once pooled (len > RAY_STR_INLINE_MAX). */
    char  stackbuf[1024];
    char* heapbuf = NULL;
    int64_t clen  = ray_str_len(x);
    const char* raw = ray_str_ptr(x);
    if (!raw) return ray_error("domain", ".sys.exec: empty or invalid command string");
    char* cmd;
    if (clen < (int64_t)sizeof(stackbuf)) {
        cmd = stackbuf;
    } else {
        heapbuf = (char*)ray_alloc_raw((size_t)clen + 1);
        if (!heapbuf) return ray_error("oom", ".sys.exec: command buffer");
        cmd = heapbuf;
    }
    memcpy(cmd, raw, (size_t)clen);
    cmd[clen] = '\0';

    ray_t* result;
    if (!capture) {
        int rc = system(cmd);
        /* -1 means the shell could not be started at all.  Decoding it would
         * yield 128+127 = 255 — indistinguishable from a command that really
         * exited 255 — so report it the way the capture path reports its
         * equivalent pclose failure. */
        if (rc == -1) {
            if (heapbuf) ray_free_raw(heapbuf);
            return ray_error("io", ".sys.exec: failed to run command: %s", strerror(errno));
        }
        result = make_i64(exec_exit_code(rc));
    } else {
        int64_t code = 0;
        ray_t* out = exec_capture(cmd, &code);
        if (RAY_IS_ERR(out)) {
            result = out;
        } else {
            ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 2);
            ray_t* vals = RAY_IS_ERR(keys) ? keys : ray_list_new(2);
            if (RAY_IS_ERR(keys) || RAY_IS_ERR(vals)) {
                if (!RAY_IS_ERR(keys)) ray_release(keys);
                ray_release(out);
                if (heapbuf) ray_free_raw(heapbuf);
                return RAY_IS_ERR(keys) ? keys : vals;
            }
            int64_t k1 = ray_sym_intern("code", 4);
            keys = ray_vec_append(keys, &k1);
            ray_t* v1 = make_i64(code);
            vals = ray_list_append(vals, v1); ray_release(v1);
            int64_t k2 = ray_sym_intern("out", 3);
            keys = ray_vec_append(keys, &k2);
            vals = ray_list_append(vals, out); ray_release(out);
            result = ray_dict_new(keys, vals);
        }
    }

    if (heapbuf) ray_free_raw(heapbuf);
    return result;
#endif
}

/* (getenv name) -- get environment variable */
ray_t* ray_getenv_fn(ray_t* x) {
    if (x->type != -RAY_STR) return ray_error("type", "getenv expects a string");
    /* ray strings are length-prefixed and NOT NUL-terminated when pooled
     * (len > RAY_STR_INLINE_MAX), so the raw ray_str_ptr cannot be handed
     * to libc getenv directly — copy into a NUL-terminated buffer first. */
    char namebuf[256];
    const char* name = str_to_cpath(x, namebuf, sizeof(namebuf));
    if (!name) return ray_error("domain", ".os.getenv name is empty or too long, got %lld bytes", (long long)ray_str_len(x));
    const char* val = getenv(name);
    return val ? ray_str(val, strlen(val)) : ray_str("", 0);
}

/* (setenv name val) -- set environment variable */
#if !defined(RAY_OS_WINDOWS)
extern int setenv(const char*, const char*, int);
#endif
ray_t* ray_setenv_fn(ray_t* name, ray_t* val) {
    if (name->type != -RAY_STR || val->type != -RAY_STR)
        return ray_error("type", "setenv expects two strings");
    /* libc setenv needs NUL-terminated args; ray strings are not
     * NUL-terminated when pooled (len > RAY_STR_INLINE_MAX). */
    char namebuf[256];
    const char* n = str_to_cpath(name, namebuf, sizeof(namebuf));
    if (!n) return ray_error("domain", ".os.setenv name is empty or too long, got %lld bytes", (long long)ray_str_len(name));
    char valbuf[4096];
    size_t vlen = ray_str_len(val);
    if (vlen >= sizeof(valbuf))
        return ray_error("range", ".os.setenv value too long, got %lld bytes", (long long)vlen);
    const char* vp = ray_str_ptr(val);
    if (!vp) return ray_error("domain", ".os.setenv: invalid value string");
    if (vlen) memcpy(valbuf, vp, vlen);
    valbuf[vlen] = '\0';
#if defined(RAY_OS_WINDOWS)
    _putenv_s(n, valbuf);
#else
    setenv(n, valbuf, 1);
#endif
    ray_retain(val);
    return val;
}

/* ══════════════════════════════════════════
 * Quote, return, args, rc, diverse, get, remove,
 * timer, env, internals, memstat, sysinfo
 * ══════════════════════════════════════════ */

/* (quote expr) -- special form, returns argument unevaluated.
 * A bare-name symbol argument becomes a LITERAL symbol (≡ the tick form
 * 'x): return a flag-set COPY — args[0] is the shared unevaluated AST
 * node, re-evaluated in loops / reused lambda bodies, so never mutate it
 * in place.  ray_sym() freshly allocates (mirrors parse.c's tick path),
 * so setting ATTR_QUOTED on the copy is safe.  Already-literal symbols
 * and all non-symbol values pass through unchanged. */
ray_t* ray_quote_fn(ray_t** args, int64_t n) {
    if (n < 1) return ray_error("domain", "quote expects 1 argument");
    ray_t* a = args[0];
    if (a->type == -RAY_SYM && !(a->attrs & ATTR_QUOTED)) {
        ray_t* q = ray_sym(a->i64);
        if (RAY_IS_ERR(q)) return q;
        q->attrs |= ATTR_QUOTED;
        return q;
    }
    ray_retain(a);
    return a;
}

/* (return) | (return x) — early exit from enclosing compiled lambda.
 * Outside a compiled lambda (e.g. value position: (map return xs), or
 * REPL top-level) this collapses to: identity for one arg, null for
 * zero, domain error otherwise. The early-exit semantics are emitted
 * by the bytecode compiler — see compile.c. */
ray_t* ray_return_fn(ray_t** args, int64_t n) {
    if (n == 0) return RAY_NULL_OBJ;
    if (n == 1) { ray_retain(args[0]); return args[0]; }
    return ray_error("domain", "return expects 0 or 1 argument");
}

/* (rc x) -- return reference count of object */
ray_t* ray_rc_fn(ray_t* x) {
    if (!x || RAY_IS_ERR(x)) return make_i64(0);
    return make_i64((int64_t)x->rc);
}

/* (diverse x) -- check if all elements in a collection are unique */
ray_t* ray_diverse_fn(ray_t* x) {
    if (ray_is_atom(x)) return make_bool(1);
    if (!is_collection(x)) return ray_error("type", "diverse expects a collection");

    int64_t n = ray_len(x);
    if (n <= 1) return make_bool(1);

    ray_t* d = ray_distinct_fn(x);
    if (RAY_IS_ERR(d)) return d;
    /* ray_distinct_fn returns lazy for concrete vecs; materialise here
     * since we need the concrete length to compare. */
    if (ray_is_lazy(d)) {
        d = ray_lazy_materialize(d);
        if (!d || RAY_IS_ERR(d)) return d ? d : ray_error("oom", NULL);
    }
    int64_t dn = ray_len(d);
    ray_release(d);
    return make_bool(dn == n ? 1 : 0);
}

/* (get dict key) -- dictionary/table lookup (alias for at) */
ray_t* ray_get_fn(ray_t* dict, ray_t* key) {
    return ray_at_fn(dict, key);
}

/* (remove dict key) -- remove key from dict, return new dict */
ray_t* ray_remove_fn(ray_t* dict, ray_t* key) {
    if (!dict || dict->type != RAY_DICT)
        return ray_error("type", "remove expects a dict");
    ray_retain(dict);
    return ray_dict_remove(dict, key);
}

/* (.time.now) -- current monotonic time in milliseconds */
ray_t* ray_time_now_fn(ray_t** args, int64_t n) {
    (void)args;
    if (n != 0) return ray_error("domain", ".time.now takes no arguments");
    return make_i64(ray_time_now_ms());
}

/* (.time.timer.set ms num fn) -- schedule fn to fire every ms; returns id */
ray_t* ray_time_timer_set_fn(ray_t** args, int64_t n) {
    if (n != 3)
        return ray_error("domain", ".time.timer.set expects 3 arguments");
    if (args[0]->type != -RAY_I64) return ray_error("type", "ms must be i64");
    if (args[1]->type != -RAY_I64) return ray_error("type", "num must be i64");
    if (args[2]->type != RAY_LAMBDA)
        return ray_error("type", "fn must be a lambda");

    int64_t ms  = args[0]->i64;
    int64_t num = args[1]->i64;
    if (ms  < 0) return ray_error("domain", "ms must be non-negative");
    if (num < 0) return ray_error("domain", "num must be non-negative");

    /* Check lambda arity: declared params list must have exactly 1 element. */
    ray_t* params = LAMBDA_PARAMS(args[2]);
    if (!params || ray_len(params) != 1)
        return ray_error("domain", "fn must accept exactly 1 argument");

    ray_poll_t* poll = (ray_poll_t*)ray_runtime_get_poll();
    if (!poll) return ray_error("domain", "no poll loop active");

    if (!poll->timers) {
        poll->timers = ray_timers_create(16);
        if (!poll->timers) return ray_error("oom", NULL);
    }

    int64_t id = ray_timers_add((ray_timers_t*)poll->timers, ms, num, args[2]);
    if (id < 0) return ray_error("oom", NULL);
    return make_i64(id);
}

/* (.time.timer.del id) -- cancel a timer; returns null */
ray_t* ray_time_timer_del_fn(ray_t* id) {
    if (!id || id->type != -RAY_I64) return ray_error("type", "id must be i64");
    ray_poll_t* poll = (ray_poll_t*)ray_runtime_get_poll();
    if (!poll || !poll->timers) return RAY_NULL_OBJ;
    ray_timers_del((ray_timers_t*)poll->timers, id->i64);
    return RAY_NULL_OBJ;
}

/* (env) -- return dict of all global environment bindings */
ray_t* ray_env_fn(ray_t* x) {
    (void)x;
    int64_t sym_ids[1024];
    ray_t* vals_buf[1024];
    int32_t count = ray_env_list(sym_ids, vals_buf, 1024);

    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, count);
    if (RAY_IS_ERR(keys)) return keys;
    ray_t* vals = ray_list_new(count);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }

    for (int32_t i = 0; i < count; i++) {
        keys = ray_vec_append(keys, &sym_ids[i]);
        if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
        vals = ray_list_append(vals, vals_buf[i]);
        if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
    }
    return ray_dict_new(keys, vals);
}

/* (.sys.build) -- return dict with internal build information */
ray_t* ray_internals_fn(ray_t** args, int64_t n) {
    (void)args;
    if (n != 0) return ray_error("arity", ".sys.build takes no arguments");
    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 2);
    if (RAY_IS_ERR(keys)) return keys;
    ray_t* vals = ray_list_new(2);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }

    int64_t ver_sym = ray_sym_intern("version", 7);
    keys = ray_vec_append(keys, &ver_sym);
#ifdef RAYFORCE_VERSION
    ray_t* v1 = ray_str(RAYFORCE_VERSION, strlen(RAYFORCE_VERSION));
#else
    ray_t* v1 = ray_str("unknown", 7);
#endif
    vals = ray_list_append(vals, v1); ray_release(v1);

    int64_t date_sym = ray_sym_intern("build-date", 10);
    keys = ray_vec_append(keys, &date_sym);
#ifdef RAYFORCE_BUILD_DATE
    ray_t* v2 = ray_str(RAYFORCE_BUILD_DATE, strlen(RAYFORCE_BUILD_DATE));
#else
    ray_t* v2 = ray_str("unknown", 7);
#endif
    vals = ray_list_append(vals, v2); ray_release(v2);

    return ray_dict_new(keys, vals);
}

/* (.sys.mem) -- return dict with memory allocator statistics */
ray_t* ray_memstat_fn(ray_t** args, int64_t n) {
    (void)args; (void)n;
    ray_mem_stats_t st;
    ray_mem_stats(&st);

    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 16);
    if (RAY_IS_ERR(keys)) return keys;
    ray_t* vals = ray_list_new(16);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }

    struct { const char* name; size_t nlen; int64_t v; } rows[] = {
        { "alloc-count",     11, (int64_t)st.alloc_count       },
        { "bytes-allocated", 15, (int64_t)st.bytes_allocated   },
        { "direct-bytes",    12, (int64_t)st.direct_bytes      },
        { "peak-bytes",      10, (int64_t)st.peak_bytes        },
        { "slab-hits",        9, (int64_t)st.slab_hits         },
        { "sys-current",     11, (int64_t)st.sys_current       },
        { "sys-mapped",      10, (int64_t)st.sys_mapped        },
        { "sys-mapped-peak", 15, (int64_t)st.sys_mapped_peak   },
        { "anon-committed",  14, ray_heap_anon_committed()     },
        { "anon-peak",        9, ray_heap_anon_peak()          },
        { "anon-watermark",  14, ray_heap_anon_watermark()     },
        { "group-perpart-runs", 18, ray_group_perpart_runs()   },
        { "asof-perpart-runs",  17, ray_asof_perpart_runs()    },
        { "sort-perpart-runs",  17, ray_sort_perpart_runs()    },
        { "window-perpart-runs", 19, ray_window_perpart_runs() },
        { "join-perpart-runs",   17, ray_join_perpart_runs()   },
        /* the global symbol table: entries and the bytes of their strings,
         * so a test can assert that a query leaves it where it found it */
        { "sym-count",        9, (int64_t)ray_sym_count()       },
        { "sym-bytes",        9, (int64_t)ray_sym_bytes()       },
    };
    for (size_t i = 0; i < sizeof(rows)/sizeof(rows[0]); i++) {
        int64_t s = ray_sym_intern(rows[i].name, rows[i].nlen);
        keys = ray_vec_append(keys, &s);
        ray_t* v = make_i64(rows[i].v);
        vals = ray_list_append(vals, v); ray_release(v);
    }

    return ray_dict_new(keys, vals);
}

/* (.sys.prof) — materialize the last profiled query's span log as a table.
 *
 * One row per completed operator span (START/END pair) and per phase tick
 * (parse / eval / materialize / optimizer passes), in completion order.
 * Columns:
 *   operator       the span label (symbol)
 *   depth          nesting depth (0 = top level)
 *   cumulative-ms  wall-clock milliseconds, including children
 *   exclusive-ms   self time — this span minus its children
 *   percent        share of total query time (exclusive-based, sums to ~100)
 *   rows           result element/row count (operator spans)
 *   allocated-kib  net process bytes allocated across the span, KiB
 *   workers        worker threads that ran a task for this span
 *   busy-ms        summed worker busy time, ms (parallelism)
 *   parallelism    effective parallelism = busy time / wall time
 *
 * Because it returns an ordinary table, an IPC client gets the profile
 * with no protocol work.  Empty (0 rows) when the profiler is inactive or
 * no query has run.  Profiling is toggled with `:t` / the `-t` flag.
 */
ray_t* ray_prof_fn(ray_t** args, int64_t n) {
    (void)args; (void)n;

    /* Read the snapshot of the last completed query — NOT the in-flight
     * buffer, which currently holds this `(.sys.prof)` call's own spans.
     * The snapshot is taken at each eval boundary (ray_profile_snapshot),
     * so it holds the fully-materialized query that ran just before. */
    const ray_profile_t* p = &g_ray_profile_last;
    int32_t ns = p->n;

    /* Count output rows: one per END and per TICK span. */
    int64_t nrows = 0;
    for (int32_t i = 0; i < ns; i++)
        if (p->spans[i].type != RAY_PROF_SPAN_START) nrows++;

    ray_t* c_op    = ray_vec_new(RAY_SYM, nrows);
    ray_t* c_depth = ray_vec_new(RAY_I64, nrows);
    ray_t* c_dur   = ray_vec_new(RAY_F64, nrows);
    ray_t* c_self  = ray_vec_new(RAY_F64, nrows);
    ray_t* c_pct   = ray_vec_new(RAY_F64, nrows);
    ray_t* c_rows  = ray_vec_new(RAY_I64, nrows);
    ray_t* c_alloc = ray_vec_new(RAY_F64, nrows);
    ray_t* c_wrk   = ray_vec_new(RAY_I64, nrows);
    ray_t* c_busy  = ray_vec_new(RAY_F64, nrows);
    ray_t* c_par   = ray_vec_new(RAY_F64, nrows);
    ray_t* cols[] = { c_op, c_depth, c_dur, c_self, c_pct,
                      c_rows, c_alloc, c_wrk, c_busy, c_par };
    for (size_t i = 0; i < sizeof(cols)/sizeof(cols[0]); i++)
        if (!cols[i] || RAY_IS_ERR(cols[i])) {
            for (size_t j = 0; j < sizeof(cols)/sizeof(cols[0]); j++)
                if (cols[j] && !RAY_IS_ERR(cols[j])) ray_release(cols[j]);
            return ray_error("oom", "sys.prof: column alloc");
        }

    int64_t* op_ids  = (int64_t*)ray_data(c_op);
    int64_t* depths  = (int64_t*)ray_data(c_depth);
    double*  durs    = (double*)ray_data(c_dur);
    double*  selfs   = (double*)ray_data(c_self);
    double*  pcts    = (double*)ray_data(c_pct);
    int64_t* rows    = (int64_t*)ray_data(c_rows);
    double*  allocs  = (double*)ray_data(c_alloc);
    int64_t* wrks    = (int64_t*)ray_data(c_wrk);
    double*  busys   = (double*)ray_data(c_busy);
    double*  pars    = (double*)ray_data(c_par);

    /* Total wall time = the whole profiled window (first START → last END).
     * Used for the `pct` share-of-total column, like EXPLAIN ANALYZE. */
    int64_t total_ns = ns > 0 ? (p->spans[ns-1].ts - p->spans[0].ts) : 0;

    /* Walk the flat span log, pairing START/END via a depth stack.
     * child_ns[d] accumulates the wall time of the direct children of the
     * span currently open at depth d, so each span's EXCLUSIVE (self) time
     * is its own wall minus that sum — standard per-operator
     * exclusive timing (cumulative is derivable by summing the subtree). */
    int32_t stack[256];
    int64_t child_ns[256] = {0};
    int32_t sp = 0;
    int64_t prev_ts = ns > 0 ? p->spans[0].ts : 0;
    int64_t r = 0;
    for (int32_t i = 0; i < ns; i++) {
        const ray_prof_span_t* s = &p->spans[i];
        if (s->type == RAY_PROF_SPAN_START) {
            if (sp < 256) { stack[sp] = i; child_ns[sp] = 0; }
            sp++;
        } else if (s->type == RAY_PROF_SPAN_END) {
            if (sp > 0) sp--;
            const ray_prof_span_t* st = (sp < 256 && sp >= 0) ? &p->spans[stack[sp]] : s;
            int64_t wall = s->ts - st->ts;
            int64_t self = (sp < 256 && sp >= 0) ? wall - child_ns[sp] : wall;
            if (self < 0) self = 0;
            /* Fold this span's wall into its parent's child accumulator. */
            if (sp - 1 >= 0 && sp - 1 < 256) child_ns[sp - 1] += wall;
            int64_t busy = (int64_t)(s->qs_busy_ns - st->qs_busy_ns);
            const char* nm = s->msg ? s->msg : "?";
            op_ids[r] = ray_sym_intern(nm, strlen(nm));
            depths[r] = sp;
            durs[r]   = (double)wall / 1e6;
            selfs[r]  = (double)self / 1e6;
            /* Percent is EXCLUSIVE-based so the column partitions the query
             * and sums to ~100% — a direct "where did time go" ranking. */
            pcts[r]   = total_ns > 0 ? (double)self * 100.0 / (double)total_ns : 0.0;
            rows[r]   = s->rows_out;
            allocs[r] = (double)(s->sys_cur - st->sys_cur) / 1024.0;
            wrks[r]   = (int64_t)s->qs_workers;
            busys[r]  = (double)busy / 1e6;
            pars[r]   = wall > 0 ? (double)busy / (double)wall : 0.0;
            r++;
        } else { /* TICK: a point phase (parse / eval / optimizer pass). */
            int64_t wall = s->ts - prev_ts;
            /* A tick is a leaf phase — count it against its parent's self. */
            if (sp - 1 >= 0 && sp - 1 < 256) child_ns[sp - 1] += wall;
            const char* nm = s->msg ? s->msg : "?";
            op_ids[r] = ray_sym_intern(nm, strlen(nm));
            depths[r] = sp;
            durs[r]   = (double)wall / 1e6;
            selfs[r]  = (double)wall / 1e6;
            pcts[r]   = total_ns > 0 ? (double)wall * 100.0 / (double)total_ns : 0.0;
            rows[r]   = 0; allocs[r] = 0.0;
            wrks[r]   = 0; busys[r] = 0.0; pars[r] = 0.0;
            r++;
        }
        prev_ts = s->ts;
    }

    /* ray_vec_new allocates capacity but leaves len at 0; publish the
     * filled row count on every column. */
    for (size_t i = 0; i < sizeof(cols)/sizeof(cols[0]); i++)
        cols[i]->len = r;

    /* Assemble the table. */
    static const char* names[] = { "operator","depth","cumulative-ms","exclusive-ms",
                                   "percent","rows","allocated-kib",
                                   "workers","busy-ms","parallelism" };
    ray_t* tbl = ray_table_new(10);
    if (!tbl || RAY_IS_ERR(tbl)) {
        for (size_t i = 0; i < sizeof(cols)/sizeof(cols[0]); i++) ray_release(cols[i]);
        return tbl ? tbl : ray_error("oom", "sys.prof: table");
    }
    for (size_t i = 0; i < sizeof(cols)/sizeof(cols[0]); i++) {
        int64_t nid = ray_sym_intern(names[i], strlen(names[i]));
        tbl = ray_table_add_col(tbl, nid, cols[i]);
        ray_release(cols[i]);
        if (RAY_IS_ERR(tbl)) return tbl;
    }
    return tbl;
}

/* (.sys.querylog) — the in-memory query-statistics ring as a table.
 *
 * One row per completed query (oldest first), capped at the ring capacity.
 * Ambient server-side query-log statistics (a system query_log table): read
 * back with ordinary queries, no extra protocol.  Each call materializes a
 * fresh snapshot, so anything a caller does to the returned table (upsert,
 * delete) touches the copy, never the ring.  Empty until logging is enabled
 * (`-Q` flag or `(.sys.querylog.enable 1)`).  Columns:
 *
 *   time         wall-clock time the query finished (timestamp)
 *   duration-ms  total wall time
 *   result-rows  rows in the result (scalar => 1)
 *   memory-kib   net process allocation across the query, KiB
 *   workers      worker threads that ran a task
 *   parallelism  worker busy time / wall time
 *   status       `ok`, or the error kind on failure
 *   query        source text (truncated)
 */
ray_t* ray_qlog_fn(ray_t** args, int64_t n) {
    (void)args;
    if (n != 0) return ray_error("domain", ".sys.querylog takes no arguments");
    return ray_qlog_table();
}

/* (.sys.querylog.enable [flag]) — toggle query-statistics logging.  No arg
 * flips the current state; 0 disables; non-zero enables.  Returns the new
 * state as i64 (0/1).  Restricted: it changes server-wide behaviour. */
ray_t* ray_qlog_enable_fn(ray_t** args, int64_t n) {
    bool on;
    if (n == 0) {
        on = !ray_qlog_enabled();
    } else if (n == 1) {
        if (args[0]->type != -RAY_I64) return ray_error("type", "flag must be i64");
        on = args[0]->i64 != 0;
    } else {
        return ray_error("domain", ".sys.querylog.enable takes 0 or 1 arguments");
    }
    ray_qlog_set_enabled(on);
    return make_i64(on ? 1 : 0);
}

/* Process id of this process.  getpid() is already used internally for heap
 * swap-file names; this is the same value, made reachable from Rayfall. */
static int64_t ray_os_pid(void) {
#if defined(RAY_OS_WINDOWS)
    return (int64_t)GetCurrentProcessId();
#else
    return (int64_t)getpid();
#endif
}

/* Host name into `buf`, NUL-terminated.  False when the platform cannot
 * answer, in which case .sys.info reports "" rather than failing the whole
 * call — one unavailable field should not cost the caller the others. */
static bool ray_os_hostname(char* buf, size_t cap) {
    if (cap == 0) return false;
    buf[0] = '\0';
#if defined(RAY_OS_WINDOWS)
    DWORD n = (DWORD)cap;
    if (!GetComputerNameA(buf, &n)) return false;
#else
    if (gethostname(buf, cap) != 0) return false;
    /* POSIX leaves truncation unterminated. */
    buf[cap - 1] = '\0';
#endif
    return buf[0] != '\0';
}

ray_t* ray_sysinfo_fn(ray_t** args, int64_t n) {
    (void)args; (void)n;
    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 5);
    if (RAY_IS_ERR(keys)) return keys;
    ray_t* vals = ray_list_new(5);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }

#if !defined(RAY_OS_WINDOWS)
    int64_t s1 = ray_sym_intern("cores", 5);
    keys = ray_vec_append(keys, &s1);
    ray_t* v1 = make_i64(sysconf(_SC_NPROCESSORS_ONLN));
    vals = ray_list_append(vals, v1); ray_release(v1);

    int64_t s2 = ray_sym_intern("page-size", 9);
    keys = ray_vec_append(keys, &s2);
    ray_t* v2 = make_i64(sysconf(_SC_PAGESIZE));
    vals = ray_list_append(vals, v2); ray_release(v2);

    int64_t s3 = ray_sym_intern("total-mem", 9);
    keys = ray_vec_append(keys, &s3);
    ray_t* v3 = make_i64(ray_sys_total_ram());
    vals = ray_list_append(vals, v3); ray_release(v3);
#else
    SYSTEM_INFO si;
    GetSystemInfo(&si);

    int64_t s1 = ray_sym_intern("cores", 5);
    keys = ray_vec_append(keys, &s1);
    ray_t* v1 = make_i64((int64_t)si.dwNumberOfProcessors);
    vals = ray_list_append(vals, v1); ray_release(v1);

    int64_t s2 = ray_sym_intern("page-size", 9);
    keys = ray_vec_append(keys, &s2);
    ray_t* v2 = make_i64((int64_t)si.dwPageSize);
    vals = ray_list_append(vals, v2); ray_release(v2);

    int64_t s3 = ray_sym_intern("total-mem", 9);
    keys = ray_vec_append(keys, &s3);
    ray_t* v3 = make_i64(ray_sys_total_ram());
    vals = ray_list_append(vals, v3); ray_release(v3);
#endif

    /* Process and host identity (#573).  An embedded process previously had
     * no way to name itself: pid plus hostname plus a start timestamp is a
     * diagnosable epoch, where a random token only tells you that two runs
     * differ, not which machine or process either one was. */
    int64_t sp = ray_sym_intern("pid", 3);
    keys = ray_vec_append(keys, &sp);
    ray_t* vp = make_i64(ray_os_pid());
    vals = ray_list_append(vals, vp); ray_release(vp);

    char host[256];
    int64_t sh = ray_sym_intern("hostname", 8);
    keys = ray_vec_append(keys, &sh);
    ray_t* vh = ray_os_hostname(host, sizeof(host)) ? ray_str(host, strlen(host))
                                                    : ray_str("", 0);
    vals = ray_list_append(vals, vh); ray_release(vh);

    return ray_dict_new(keys, vals);
}

/* Parse argv into the .sys.args dict.  Recognized launcher flags become
 * typed top-level entries (i64 / bool / str); everything after `--` becomes
 * a str->str `user` subdict via `-k v` / `--k v` pairing.  Auth passwords
 * (-u/-U) are intentionally NOT exposed. */
ray_t* ray_build_sys_args(int argc, char** argv) {
    const char* file = "";
    const char* log  = "";
    int64_t port = 0, cores = 0;
    bool timeit = false, interactive = false, querylog = false;
    int user_start = -1;

    for (int i = 1; i < argc; i++) {
        if (strcmp(argv[i], "--") == 0) { user_start = i + 1; break; }
        else if (strcmp(argv[i], "-i") == 0 || strcmp(argv[i], "--interactive") == 0)
            interactive = true;
        else if ((strcmp(argv[i], "-p") == 0 || strcmp(argv[i], "--port") == 0) && i + 1 < argc)
            port = (int64_t)atoll(argv[++i]);
        else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "--cores") == 0) {
            if (i + 1 >= argc)
                return ray_error("rank", ".sys.args: %s requires an argument", argv[i]);
            int64_t v;
            if (!parse_sys_int_arg(argv[++i], true, true, &v))
                return ray_error("domain", ".sys.args: invalid %s value \"%s\"",
                                  argv[i - 1], argv[i]);
            cores = v;
        }
        else if (strcmp(argv[i], "-t") == 0 || strcmp(argv[i], "--timeit") == 0) {
            if (i + 1 >= argc)
                return ray_error("rank", ".sys.args: %s requires an argument", argv[i]);
            int64_t v;
            if (!parse_sys_int_arg(argv[++i], false, false, &v))
                return ray_error("domain", ".sys.args: invalid %s value \"%s\"",
                                  argv[i - 1], argv[i]);
            timeit = (v != 0);
        }
        else if (strcmp(argv[i], "-Q") == 0 || strcmp(argv[i], "--querylog") == 0) {
            if (i + 1 >= argc)
                return ray_error("rank", ".sys.args: %s requires an argument", argv[i]);
            int64_t v;
            if (!parse_sys_int_arg(argv[++i], false, false, &v))
                return ray_error("domain", ".sys.args: invalid %s value \"%s\"",
                                  argv[i - 1], argv[i]);
            querylog = (v != 0);
        }
        else if ((strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--file") == 0) && i + 1 < argc)
            file = argv[++i];
        else if (strcmp(argv[i], "-u") == 0 && i + 1 < argc) i++;   /* skip secret */
        else if (strcmp(argv[i], "-U") == 0 && i + 1 < argc) i++;   /* skip secret */
        else if ((strcmp(argv[i], "-l") == 0 || strcmp(argv[i], "-L") == 0) && i + 1 < argc)
            log = argv[++i];
        else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) { /* ignore */ }
        else file = argv[i];
    }

    /* user subdict (built by upsert so dup keys collapse, last wins) */
    ray_t* uk = ray_sym_vec_new(RAY_SYM_W64, 0);
    ray_t* uv = ray_list_new(0);
    ray_t* user = ray_dict_new(uk, uv);
    if (user_start >= 0) {
        for (int i = user_start; i < argc; ) {
            const char* tok = argv[i];
            if (tok[0] == '-') {
                const char* name = tok; while (*name == '-') name++;
                if (*name == '\0') { i++; continue; }   /* lone - / -- */
                const char* value = "";
                if (i + 1 < argc && argv[i + 1][0] != '-') { value = argv[i + 1]; i += 2; }
                else i += 1;
                ray_t* ka = ray_sym(ray_sym_intern(name, strlen(name)));
                ray_t* va = ray_str(value, strlen(value));
                ray_t* nu = ray_dict_upsert(user, ka, va);
                ray_release(ka); ray_release(va);
                if (RAY_IS_ERR(nu)) {
                    ray_release(nu);                              /* drop the error obj */
                    user = ray_dict_new(ray_sym_vec_new(RAY_SYM_W64, 0), ray_list_new(0));
                    break;                                        /* `user` already freed by upsert */
                }
                user = nu;
            } else {
                i++;   /* bare value with no key — ignore */
            }
        }
    }

    /* top-level dict: 8 entries (file, port, cores, timeit, querylog, interactive, log, user) */
    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 8);
    ray_t* vals = ray_list_new(8);

    int64_t k;
    ray_t* v;
    k = ray_sym_intern("file", 4);        keys = ray_vec_append(keys, &k);
    v = ray_str(file, strlen(file));      vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("port", 4);        keys = ray_vec_append(keys, &k);
    v = ray_i64(port);                    vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("cores", 5);       keys = ray_vec_append(keys, &k);
    v = ray_i64(cores);                   vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("timeit", 6);      keys = ray_vec_append(keys, &k);
    v = ray_bool(timeit);                 vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("querylog", 8);    keys = ray_vec_append(keys, &k);
    v = ray_bool(querylog);               vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("interactive", 11);keys = ray_vec_append(keys, &k);
    v = ray_bool(interactive);            vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("log", 3);         keys = ray_vec_append(keys, &k);
    v = ray_str(log, strlen(log));        vals = ray_list_append(vals, v); ray_release(v);
    k = ray_sym_intern("user", 4);        keys = ray_vec_append(keys, &k);
    vals = ray_list_append(vals, user);   ray_release(user);

    return ray_dict_new(keys, vals);
}

/* (.sys.args) -- return the application-arguments dict (empty if unset) */
/* The launcher dict is built once at startup; `source` is the one key
 * that changes while the process runs — the file currently being
 * evaluated (the innermost `load`, or the command-line script), by the
 * path that was actually opened, so a script can locate its neighbours
 * from any working directory (#506).  Empty at the REPL, under a pipe,
 * or in a hook or timer outside any file — the same convention as
 * `file`.  bash's $BASH_SOURCE next to $0. */
ray_t* ray_sys_args_fn(ray_t** args, int64_t n) {
    (void)args;
    if (n != 0) return ray_error("domain", ".sys.args takes no arguments");
    ray_t* d = (ray_t*)ray_runtime_get_sys_args();
    ray_t* keys; ray_t* vals;
    if (d) {
        keys = ray_dict_keys(d); ray_retain(keys);   /* append COWs the shared vectors */
        vals = ray_dict_vals(d); ray_retain(vals);
    } else {
        keys = ray_sym_vec_new(RAY_SYM_W64, 1);
        vals = ray_list_new(1);
    }
    ray_t* nfo = ray_eval_get_nfo();
    ray_t* src = (nfo && !RAY_IS_ERR(nfo)) ? NFO_FILENAME(nfo) : NULL;
    bool is_file = src && !RAY_IS_ERR(src) && src->type == -RAY_STR &&
                   !(ray_str_len(src) == strlen(RAY_NFO_REPL_NAME) &&
                     memcmp(ray_str_ptr(src), RAY_NFO_REPL_NAME, ray_str_len(src)) == 0);
    ray_t* sv = is_file ? (ray_retain(src), src) : ray_str("", 0);
    int64_t k = ray_sym_intern("source", 6);
    keys = ray_vec_append(keys, &k);
    if (RAY_IS_ERR(keys)) { ray_release(vals); ray_release(sv); return keys; }
    vals = ray_list_append(vals, sv); ray_release(sv);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
    return ray_dict_new(keys, vals);
}

/* ══════════════════════════════════════════
 * IPC builtins
 * ══════════════════════════════════════════ */

/* (hopen "host:port[:user:password]" [timeout-ms]) → i64 handle.
 * The optional second argument bounds the TCP connect and handshake in
 * milliseconds; omitted leaves the default budget. */
ray_t* ray_hopen_fn(ray_t** args, int64_t n) {
#ifdef RAY_FUZZING
    /* No outbound connections under fuzzing. */
    (void)args; (void)n;
    return ray_error("restricted", "ipc.open disabled under fuzzing");
#else
    if (n < 1 || n > 2)
        return ray_error("rank", ".ipc.open expects 1 or 2 arguments: \"host:port[:user:password]\" [timeout-ms]");
#endif

    ray_t* x = args[0];
    if (!ray_is_atom(x) || x->type != -RAY_STR)
        return ray_error("type", ".ipc.open expects a string \"host:port[:user:password]\", got %s", ray_type_name(x->type));

    /* Optional second argument: an integer connect timeout in
     * milliseconds (0 = use default), or an options dict carrying
     * `timeout`, `compress` (#541) and/or `keepalive` (#589). */
    int    timeout_ms = 0;
    size_t compress   = RAY_IPC_COMPRESS_AUTO;
    int    keepalive  = RAY_IPC_KEEPALIVE_AUTO;
    if (n == 2) {
        ray_t* err = ray_ipc_parse_open_opts(args[1], &timeout_ms, &compress,
                                             &keepalive);
        if (err) return err;
    }

    const char* s = ray_str_ptr(x);
    size_t slen = ray_str_len(x);

    /* Split on colons */
    const char* parts[4] = {0};
    size_t part_lens[4] = {0};
    int n_parts = 0;
    const char* start = s;
    for (size_t i = 0; i <= slen && n_parts < 4; i++) {
        if (i == slen || s[i] == ':') {
            parts[n_parts] = start;
            part_lens[n_parts] = (size_t)(&s[i] - start);
            n_parts++;
            start = &s[i + 1];
        }
    }
    if (n_parts < 2) return ray_error("domain", ".ipc.open expects \"host:port[:user:password]\", missing port");

    char host[256];
    if (part_lens[0] >= sizeof(host)) return ray_error("domain", ".ipc.open host too long, got %lld bytes", (long long)part_lens[0]);
    memcpy(host, parts[0], part_lens[0]);
    host[part_lens[0]] = '\0';

    char port_str[8];
    if (part_lens[1] >= sizeof(port_str)) return ray_error("domain", ".ipc.open port field too long, got %lld bytes", (long long)part_lens[1]);
    memcpy(port_str, parts[1], part_lens[1]);
    port_str[part_lens[1]] = '\0';
    /* Strict parse: atoi("1abc") returns 1, silently accepting a malformed
     * port with a numeric prefix.  Require the whole field to be digits. */
    char* endp;
    errno = 0;
    long port_l = strtol(port_str, &endp, 10);
    if (endp == port_str || *endp != '\0')
        return ray_error("domain", ".ipc.open port must be numeric 1..65535, got \"%s\"", port_str);
    if (errno == ERANGE || port_l <= 0 || port_l > 65535)
        return ray_error("domain", ".ipc.open port must be in 1..65535, got \"%s\"", port_str);
    int port = (int)port_l;

    char user[128] = "";
    char password[128] = "";
    if (n_parts >= 4) {
        if (part_lens[2] < sizeof(user)) {
            memcpy(user, parts[2], part_lens[2]);
            user[part_lens[2]] = '\0';
        }
        if (part_lens[3] < sizeof(password)) {
            memcpy(password, parts[3], part_lens[3]);
            password[part_lens[3]] = '\0';
        }
    }

    const char* pw_ptr = (n_parts >= 4) ? password : NULL;
    const char* us_ptr = (n_parts >= 4) ? user : NULL;

    int64_t h = ray_ipc_connect_opts(host, (uint16_t)port, us_ptr, pw_ptr, timeout_ms,
                                     compress, keepalive);
    if (h == -2) return ray_error("access", "server requires authentication");
    if (h == -3) return ray_error("access", "authentication failed");
    if (h == -4) return ray_error("io", "wire version mismatch: %s:%d", host, port);
    if (h == -5) return ray_error("io", "connection timed out: %s:%d", host, port);
    if (h == -6) return ray_error("io", "connect failed: %s:%d: %s", host, port, strerror(errno));
    if (h < 0) return ray_error("io", "connection refused: %s:%d", host, port);

    return make_i64(h);
}

/* (hclose handle) → null */
ray_t* ray_hclose_fn(ray_t* x) {
    if (!ray_is_atom(x) || (x->type != -RAY_I64 && x->type != -RAY_I32))
        return ray_error("type", ".ipc.close expects an i64 or i32 handle, got %s", ray_type_name(x->type));
    int64_t h = (x->type == -RAY_I64) ? x->i64 : x->i32;
    ray_ipc_close(h);
    return RAY_NULL_OBJ;
}

/* (.ipc.send handle msg [timeout-ms]) → result
 *
 * The optional timeout bounds the whole round trip (#589).  On expiry the
 * connection is closed — a late reply would otherwise answer the next
 * send — and an `io` error is returned.  0N means no deadline, as if the
 * argument were absent. */
ray_t* ray_hsend_fn(ray_t** args, int64_t n) {
    if (n < 2 || n > 3)
        return ray_error("rank", ".ipc.send expects 2 or 3 arguments: handle msg [timeout-ms]");
    ray_t* handle = args[0];
    ray_t* msg    = args[1];
    if (!ray_is_atom(handle) || (handle->type != -RAY_I64 && handle->type != -RAY_I32))
        return ray_error("type", ".ipc.send expects an i64 or i32 handle, got %s", ray_type_name(handle->type));
    int64_t h = (handle->type == -RAY_I64) ? handle->i64 : handle->i32;

    int64_t timeout_ms = 0;
    if (n == 3) {
        ray_t* t = args[2];
        if (!ray_is_atom(t) || (t->type != -RAY_I64 && t->type != -RAY_I32))
            return ray_error("type", ".ipc.send timeout must be an integer (ms), got %s", ray_type_name(t->type));
        int64_t tv = (t->type == -RAY_I64) ? t->i64
                   : (t->i32 == NULL_I32 ? NULL_I64 : (int64_t)t->i32);
        if (tv != NULL_I64) {
            /* 0 is refused rather than guessed at: "no deadline" or
             * "already expired"? */
            if (tv <= 0)
                return ray_error("domain", ".ipc.send timeout must be > 0 or 0N (none), got %lld", (long long)tv);
            timeout_ms = tv;
        }
    }

    /* Validate message is serializable (reject builtins, etc.) */
    if (ray_serde_size(msg) <= 0)
        return ray_error("type", "message not serializable");
    return ray_ipc_send_timeout(h, msg, timeout_ms);
}

/* (.ipc.post handle msg) → null on local send, error on failure.
 * Async fire-and-forget counterpart to (.ipc.send ...): the server runs
 * the message through `.ipc.on.async` (or default eval) and sends NO
 * response, so only a LOCAL send failure is observable here. */
ray_t* ray_hpost_fn(ray_t* handle, ray_t* msg) {
    if (!ray_is_atom(handle) || (handle->type != -RAY_I64 && handle->type != -RAY_I32))
        return ray_error("type", ".ipc.post expects an i64 or i32 handle, got %s", ray_type_name(handle->type));
    int64_t h = (handle->type == -RAY_I64) ? handle->i64 : handle->i32;
    /* Validate message is serializable (reject builtins, etc.) */
    if (ray_serde_size(msg) <= 0)
        return ray_error("type", "message not serializable");
    ray_err_t rc = ray_ipc_send_async(h, msg);
    if (rc != RAY_OK)
        return ray_error("io", "ipc async send failed");
    return RAY_NULL_OBJ;
}

/* (.ipc.handle) → i64 current connection handle inside any `.ipc.on.*`
 * hook, or -1 outside any hook.  Registered variadic so both
 * `(.ipc.handle)` (no args) and `(.ipc.handle 0)` (one arg, ignored)
 * work — matches the convention of `.sys.gc` / `.sys.info`. */
/* Build {handle limit_bytes limit_frames queued_bytes queued_frames hwm_bytes}
 * for a live connection. */
static ray_t* ipc_tx_info_dict(int64_t h, const ray_ipc_tx_info_t* ti) {
    static const char* const names[7] = { "handle", "inbound", "limit_bytes", "limit_frames",
                                          "queued_bytes", "queued_frames", "hwm_bytes" };
    int64_t vals_i[7] = { h, ti->inbound ? 1 : 0, ti->limit_bytes, ti->limit_frames,
                          ti->queued_bytes, ti->queued_frames, ti->hwm_bytes };
    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 7);
    ray_t* vals = ray_list_new(7);
    if (!keys || RAY_IS_ERR(keys) || !vals || RAY_IS_ERR(vals)) {
        if (keys && !RAY_IS_ERR(keys)) ray_release(keys);
        if (vals && !RAY_IS_ERR(vals)) ray_release(vals);
        return ray_error("oom", NULL);
    }
    for (int i = 0; i < 7; i++) {
        int64_t k = ray_sym_intern(names[i], strlen(names[i]));
        keys = ray_vec_append(keys, &k);
        if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
        /* `inbound` is the direction: accepted here, or opened by .ipc.open */
        ray_t* v = (i == 1) ? ray_bool(vals_i[i] != 0) : make_i64(vals_i[i]);
        vals = ray_list_append(vals, v); ray_release(v);
        if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
    }
    return ray_dict_new(keys, vals);
}

static bool ipc_arg_i64(ray_t* x, int64_t* out) {
    if (!x || !ray_is_atom(x)) return false;
    if (x->type == -RAY_I64) { *out = x->i64; return true; }
    if (x->type == -RAY_I32) { *out = x->i32; return true; }
    return false;
}

/* `(.ipc.handle)` (no args) and `(.ipc.handle 0)` (one arg, ignored)
 * answer the current connection's handle, -1 outside a hook.  Given a
 * LIVE connection handle, `(.ipc.handle h)` instead returns that
 * connection's transmit-queue view (#486): its effective backlog limits,
 * the bytes and frames queued right now, and its high-water mark. */
ray_t* ray_ipc_handle_fn(ray_t** args, int64_t n) {
    int64_t h;
    if (n == 1 && ipc_arg_i64(args[0], &h)) {
        ray_ipc_tx_info_t ti;
        if (ray_ipc_tx_info(h, &ti) == RAY_OK) return ipc_tx_info_dict(h, &ti);
    }
    return make_i64(ray_ipc_current_handle());
}

static ray_t* ipc_txlimit_dict(int64_t bytes, int64_t frames) {
    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, 2);
    ray_t* vals = ray_list_new(2);
    if (!keys || RAY_IS_ERR(keys) || !vals || RAY_IS_ERR(vals)) {
        if (keys && !RAY_IS_ERR(keys)) ray_release(keys);
        if (vals && !RAY_IS_ERR(vals)) ray_release(vals);
        return ray_error("oom", NULL);
    }
    int64_t k = ray_sym_intern("bytes", 5);  keys = ray_vec_append(keys, &k);
    if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
    k = ray_sym_intern("frames", 6);         keys = ray_vec_append(keys, &k);
    if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
    ray_t* v = make_i64(bytes);  vals = ray_list_append(vals, v); ray_release(v);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
    v = make_i64(frames);        vals = ray_list_append(vals, v); ray_release(v);
    if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
    return ray_dict_new(keys, vals);
}

/* (.ipc.txlimit)                 -> {bytes frames}: the process default
 * (.ipc.txlimit bytes [frames])  -> set the default; frames 0 = unlimited
 * (.ipc.txlimit h bytes frames)  -> override one live connection
 * Every form answers with the limits now in force for its target. */
ray_t* ray_ipc_txlimit_fn(ray_t** args, int64_t n) {
    if (n > 3) return ray_error("rank", ".ipc.txlimit takes 0 to 3 arguments, got %lld", (long long)n);
    int64_t a[3] = {0, 0, 0};
    for (int64_t i = 0; i < n; i++)
        if (!ipc_arg_i64(args[i], &a[i]))
            return ray_error("type", ".ipc.txlimit expects integer arguments, got %s", ray_type_name(args[i]->type));
    int64_t bytes, frames;
    if (n == 0) {
        ray_ipc_tx_limit_get(&bytes, &frames);
        return ipc_txlimit_dict(bytes, frames);
    }
    if (n == 3) {
        ray_err_t rc = ray_ipc_tx_limit_set_handle(a[0], a[1], a[2]);
        if (rc == RAY_ERR_DOMAIN)
            return ray_error("domain", ".ipc.txlimit: bytes must be >= %lld and frames >= 0", (long long)RAY_IPC_TX_MIN_BYTES);
        if (rc != RAY_OK)
            return ray_error("domain", ".ipc.txlimit: %lld is not a live connection handle", (long long)a[0]);
        return ipc_txlimit_dict(a[1], a[2]);
    }
    ray_ipc_tx_limit_get(&bytes, &frames);
    bytes = a[0];
    if (n == 2) frames = a[1];
    if (ray_ipc_tx_limit_set(bytes, frames) != RAY_OK)
        return ray_error("domain", ".ipc.txlimit: bytes must be >= %lld and frames >= 0", (long long)RAY_IPC_TX_MIN_BYTES);
    return ipc_txlimit_dict(bytes, frames);
}

ray_t* ray_mc_sub_fn(ray_t** args, int64_t n) {
    if (n < 1 || n > 2)
        return ray_error("domain", ".mc.sub expects 1 or 2 arguments");
    if (!ray_ipc_context_poll())
        return ray_error("domain", ".mc.sub requires a poll IPC context");
    ray_t* filter = (n == 2) ? args[1] : RAY_NULL_OBJ;
    return ray_mcast_sub(ray_ipc_active_poll(), ray_ipc_current_handle(), args[0], filter);
}

ray_t* ray_mc_unsub_fn(ray_t* topic) {
    if (!ray_ipc_context_poll())
        return ray_error("domain", ".mc.unsub requires a poll IPC context");
    return ray_mcast_unsub(ray_ipc_active_poll(), ray_ipc_current_handle(), topic);
}

ray_t* ray_mc_pub_fn(ray_t* topic, ray_t* payload) {
    return ray_mcast_pub(ray_ipc_active_poll(), topic, payload);
}

ray_t* ray_mc_stats_fn(ray_t** args, int64_t n) {
    (void)args;
    if (n != 0) return ray_error("domain", ".mc.stats takes no arguments");
    return ray_mcast_stats(ray_ipc_active_poll());
}