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.
 */

/**   I/O builtins, type casting, and misc builtins extracted from eval.c.
 */

#include <errno.h>
#include <stdint.h>
#include "lang/eval.h"
#include "lang/internal.h"
#include "ops/idxop.h"       /* ray_index_carry_append: keep a hash index across concat */
#include "lang/env.h"
#include "core/platform.h"   /* ray_vm_map_fd_ro / ray_vm_unmap_file (tracked) */
#include "vec/vec.h"
#include "vec/str.h"
#include "lang/nfo.h"
#include "lang/parse.h"
#include "core/pool.h"
#include "core/types.h"
#include "io/csv.h"
#include "ops/ops.h"

static inline double clear_neg_zero(double v) {
    uint64_t bits; memcpy(&bits, &v, 8);
    if (bits == UINT64_C(0x8000000000000000)) v = 0.0;
    return v;
}
#include "ops/hash.h"
#include "store/part.h"
#include "store/splay.h"
#include "table/sym.h"
#include "core/profile.h"
#include "mem/sys.h"
#include "lang/format.h"

#include <string.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <time.h>
#include <sys/stat.h>
#include "mem/heap.h"   /* ray_heap_anon_watermark/committed — read budget */
#if !defined(RAY_OS_WINDOWS)
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#endif

/* ══════════════════════════════════════════
 * I/O builtins: println, show, format, read-csv, write-csv, as, type
 * ══════════════════════════════════════════ */

/* Helper: return the null literal string for a typed null atom (e.g. "0Ni" for I32). */
static const char* null_literal_str(int8_t type) {
    switch (-type) {
        case RAY_I16:       return "0Nh";
        case RAY_I32:       return "0Ni";
        case RAY_I64:       return "0Nl";
        case RAY_F32:       return "0Ne";
        case RAY_F64:       return "0Nf";
        case RAY_DATE:      return "0Nd";
        case RAY_TIME:      return "0Nt";
        case RAY_TIMESTAMP: return "0Np";
        /* SYM/STR use their empty literal as the canonical null spelling;
         * callers keep them on their ordinary formatting paths. */
        default:            return "null";
    }
}

/* Helper: print a ray_t value to a file handle */
void ray_lang_print(FILE* fp, ray_t* val) {
    if (!val || RAY_IS_ERR(val)) { fprintf(fp, "error"); return; }
    if (RAY_IS_NULL(val)) { fprintf(fp, "null"); return; }
    /* Materialize lazy handles before printing.  val is BORROWED and
     * ray_lazy_materialize consumes its input, so give it its own ref;
     * print the owned concrete result and release it before returning. */
    if (ray_is_lazy(val)) {
        ray_retain(val);
        ray_t* mat = ray_lazy_materialize(val);
        ray_lang_print(fp, mat);
        if (mat && !RAY_IS_ERR(mat)) ray_release(mat);
        return;
    }
    if (!val || RAY_IS_ERR(val)) { fprintf(fp, "error"); return; }
    /* STR/SYM use their empty literals as canonical null spellings. */
    if (val->type != -RAY_STR && val->type != -RAY_SYM &&
        RAY_ATOM_IS_NULL(val)) {
        fprintf(fp, "%s", null_literal_str(val->type));
        return;
    }
    switch (val->type) {
    case -RAY_I64:  fprintf(fp, "%" PRId64, val->i64); break;
    case -RAY_F64: {
        double fv = val->f64;
        fv = clear_neg_zero(fv);
        fprintf(fp, "%g", fv);
        break;
    }
    case -RAY_BOOL: fprintf(fp, "%s", val->b8 ? "true" : "false"); break;
    case -RAY_SYM: {
        ray_t* s = ray_sym_str(val->i64);
        if (s) fprintf(fp, "'%.*s", (int)ray_str_len(s), ray_str_ptr(s));
        else fprintf(fp, "'?");
        break;
    }
    case -RAY_STR: {
        const char* s = ray_str_ptr(val);
        size_t slen = ray_str_len(val);
        fprintf(fp, "%.*s", (int)slen, s);
        break;
    }
    case RAY_LIST: {
        fprintf(fp, "[");
        int64_t len = ray_len(val);
        ray_t** elems = (ray_t**)ray_data(val);
        for (int64_t i = 0; i < len; i++) {
            if (i > 0) fprintf(fp, " ");
            ray_lang_print(fp, elems[i]);
        }
        fprintf(fp, "]");
        break;
    }
    case RAY_TABLE:
        fprintf(fp, "<table %" PRId64 "x%" PRId64 ">",
                ray_table_nrows(val), ray_table_ncols(val));
        break;
    case RAY_UNARY: case RAY_BINARY: case RAY_VARY: {
        const char* name = ray_fn_name(val);
        fprintf(fp, "%s", name[0] ? name : "builtin");
        break;
    }
    default: {
        /* Fall back to ray_fmt for everything else: i16, i32, u8, all
         * vector types (I16/I32/F64/SYM/...), DICT, GUID, temporal, etc.
         * Without this println on (println 5i) printed "<type:-4>" — a
         * debug placeholder, not the value. */
        ray_t* s = ray_fmt(val, 0);
        if (s && !RAY_IS_ERR(s)) {
            fprintf(fp, "%.*s", (int)ray_str_len(s), ray_str_ptr(s));
            ray_release(s);
        } else {
            fprintf(fp, "<type:%d>", val->type);
            if (s) ray_release(s);
        }
        break;
    }
    }
}

/* Helper: format string with % placeholders, substituting args.
 * Returns a heap-allocated char* (caller must ray_free_raw) and sets *out_len.
 * If fmt has no %, returns NULL (caller falls back to plain print). */
static char* fmt_interpolate(const char* fmt, size_t flen, ray_t** args, int64_t nargs, int64_t arg_start, size_t* out_len) {
    /* Quick scan: any % in fmt? */
    int has_pct = 0;
    for (size_t i = 0; i < flen; i++) if (fmt[i] == '%') { has_pct = 1; break; }
    if (!has_pct) return NULL;

    /* Build result in a dynamic buffer */
    size_t cap = flen + 256;
    char* buf = ray_calloc_raw(cap);
    if (!buf) return NULL;
    size_t pos = 0;
    int64_t ai = arg_start;

    for (size_t i = 0; i < flen; i++) {
        if (fmt[i] == '%' && ai < nargs) {
            /* Format the arg into a temp buffer */
            char tmp[256];
            ray_t* a = args[ai++];
            /* Defensive: args are pre-materialized by eval, but if a lazy
             * slips through, materialize a PRIVATE ref — a is borrowed and
             * ray_lazy_materialize consumes its input. */
            ray_t* a_owned = NULL;
            if (a && ray_is_lazy(a)) {
                ray_retain(a);
                a = a_owned = ray_lazy_materialize(a);
            }
            int tlen = 0;
            if (!a || RAY_IS_ERR(a)) {
                tlen = snprintf(tmp, sizeof(tmp), "error");
            } else if (a->type != -RAY_STR && a->type != -RAY_SYM &&
                       RAY_ATOM_IS_NULL(a)) {
                tlen = snprintf(tmp, sizeof(tmp), "%s", null_literal_str(a->type));
            } else if (a->type == -RAY_I64) {
                tlen = snprintf(tmp, sizeof(tmp), "%" PRId64, a->i64);
            } else if (a->type == -RAY_F64) {
                double fv = a->f64;
                fv = clear_neg_zero(fv);
                tlen = snprintf(tmp, sizeof(tmp), "%g", fv);
            } else if (a->type == -RAY_BOOL) {
                tlen = snprintf(tmp, sizeof(tmp), "%s", a->b8 ? "true" : "false");
            } else if (a->type == -RAY_STR) {
                const char* sp = ray_str_ptr(a);
                size_t sl = ray_str_len(a);
                while (pos + sl + 1 > cap) { cap *= 2; buf = ray_realloc_raw(buf, cap); }
                memcpy(buf + pos, sp, sl);
                pos += sl;
                if (a_owned) ray_release(a_owned);
                continue;
            } else if (a->type == -RAY_SYM) {
                ray_t* ss = ray_sym_str(a->i64);
                if (ss) {
                    const char* sp = ray_str_ptr(ss);
                    size_t sl = ray_str_len(ss);
                    /* sym 0 resolves to "" — the empty symbol shows as ' */
                    if (sl == 0) { sp = "'"; sl = 1; }
                    while (pos + sl + 1 > cap) { cap *= 2; buf = ray_realloc_raw(buf, cap); }
                    memcpy(buf + pos, sp, sl);
                    pos += sl;
                    ray_release(ss);
                    if (a_owned) ray_release(a_owned);
                    continue;
                }
                tlen = snprintf(tmp, sizeof(tmp), "'?");
            } else {
                /* Fall back to ray_fmt */
                ray_t* formatted = ray_fmt(a, 0);
                if (formatted && !RAY_IS_ERR(formatted)) {
                    const char* sp = ray_str_ptr(formatted);
                    size_t sl = ray_str_len(formatted);
                    while (pos + sl + 1 > cap) { cap *= 2; buf = ray_realloc_raw(buf, cap); }
                    memcpy(buf + pos, sp, sl);
                    pos += sl;
                    ray_release(formatted);
                    if (a_owned) ray_release(a_owned);
                    continue;
                }
                if (formatted) ray_release(formatted);
                tlen = snprintf(tmp, sizeof(tmp), "<type:%d>", a->type);
            }
            if (a_owned && !RAY_IS_ERR(a_owned)) ray_release(a_owned);
            while (pos + (size_t)tlen + 1 > cap) { cap *= 2; buf = ray_realloc_raw(buf, cap); }
            memcpy(buf + pos, tmp, (size_t)tlen);
            pos += (size_t)tlen;
        } else {
            if (pos + 2 > cap) { cap *= 2; buf = ray_realloc_raw(buf, cap); }
            buf[pos++] = fmt[i];
        }
    }
    buf[pos] = '\0';
    *out_len = pos;
    return buf;
}

/* (println val1 val2 ...) — print values to stdout, newline at end.
 * If first arg is a string with % placeholders, substitutes remaining args. */
ray_t* ray_println_fn(ray_t** args, int64_t n) {
    /* args are pre-materialized by eval (materialize_owned_args runs for
     * every non-LAZY_AWARE vary builtin); a stray lazy is handled safely
     * by ray_lang_print.  Materializing borrowed args here would consume
     * eval's refs. */

    /* Format string mode: first arg is a string with % placeholders */
    if (n >= 2 && args[0] && args[0]->type == -RAY_STR) {
        const char* fmt = ray_str_ptr(args[0]);
        size_t flen = ray_str_len(args[0]);
        size_t out_len = 0;
        char* result = fmt_interpolate(fmt, flen, args, n, 1, &out_len);
        if (result) {
            fwrite(result, 1, out_len, stdout);
            fputc('\n', stdout);
            fflush(stdout);
            ray_free_raw(result);
            return RAY_NULL_OBJ;
        }
    }

    for (int64_t i = 0; i < n; i++) {
        if (i > 0) fputc(' ', stdout);
        ray_lang_print(stdout, args[i]);
    }
    fputc('\n', stdout);
    fflush(stdout);
    return RAY_NULL_OBJ;
}

/* (print val1 val2 ...) — like println but without trailing newline */
ray_t* ray_print_fn(ray_t** args, int64_t n) {
    /* args are pre-materialized by eval — see ray_println_fn. */

    /* Format string mode: first arg is a string with % placeholders */
    if (n >= 2 && args[0] && args[0]->type == -RAY_STR) {
        const char* fmt = ray_str_ptr(args[0]);
        size_t flen = ray_str_len(args[0]);
        size_t out_len = 0;
        char* result = fmt_interpolate(fmt, flen, args, n, 1, &out_len);
        if (result) {
            fwrite(result, 1, out_len, stdout);
            fflush(stdout);
            ray_free_raw(result);
            return RAY_NULL_OBJ;
        }
    }

    for (int64_t i = 0; i < n; i++) {
        if (i > 0) fputc(' ', stdout);
        ray_lang_print(stdout, args[i]);
    }
    fflush(stdout);
    return RAY_NULL_OBJ;
}

/* (show val1 val2 ...) — print values to stdout using ray_fmt, newline at end */
ray_t* ray_show_fn(ray_t** args, int64_t n) {
    /* args are pre-materialized by eval — see ray_println_fn. */
    for (int64_t i = 0; i < n; i++) {
        if (!args[i] || RAY_IS_ERR(args[i])) { fprintf(stdout, "error"); continue; }
        /* show prints the whole value, streamed; the display limits belong
         * to the REPL and to print. */
        ray_err_t e = ray_fmt_write(stdout, args[i]);
        if (e == RAY_OK || e == RAY_ERR_IO) continue;   /* written, partly written, or stdout failed */
        /* out of memory before any output: fall back to the bounded REPL display */
        ray_t* formatted = ray_fmt(args[i], 1);
        if (formatted && !RAY_IS_ERR(formatted)) {
            const char* sp = ray_str_ptr(formatted);
            size_t sl = ray_str_len(formatted);
            fwrite(sp, 1, sl, stdout);
            ray_release(formatted);
        } else {
            if (formatted) ray_release(formatted);
            ray_lang_print(stdout, args[i]);
        }
    }
    fputc('\n', stdout);
    fflush(stdout);
    return RAY_NULL_OBJ;
}

/* (format "hello % world %" a b) — string formatting with % placeholders */
ray_t* ray_format_fn(ray_t** args, int64_t n) {
    if (n < 1) return ray_error("domain", "format: expects at least 1 argument, got %lld", (long long)n);
    /* args are pre-materialized by eval — see ray_println_fn. */
    if (!args[0] || args[0]->type != -RAY_STR) return ray_error("type", "format: format string must be str, got %s", args[0] ? ray_type_name(args[0]->type) : "null");
    const char* fmt = ray_str_ptr(args[0]);
    size_t flen = ray_str_len(args[0]);
    size_t out_len = 0;
    char* result = fmt_interpolate(fmt, flen, args, n, 1, &out_len);
    if (result) {
        ray_t* s = ray_str(result, out_len);
        ray_free_raw(result);
        return s;
    }
    /* No placeholders: return fmt as-is */
    ray_retain(args[0]);
    return args[0];
}

/* (resolve 'name) — check if name exists in env, return value or null.
 * SPECIAL_FORM: does not evaluate args. */
/* (resolve tbl) — replace I64 columns with SYM columns where values are valid sym IDs.
 * This makes query results human-readable (sym names instead of intern IDs).
 * Also accepts (resolve db tbl) for compat — just ignores db. */
ray_t* ray_resolve_fn(ray_t** args, int64_t n) {
    if (n < 1) return ray_error("arity", "resolve expects at least 1 argument");

    /* Evaluate all args */
    ray_t* tbl = NULL;
    if (n == 1) {
        tbl = ray_eval(args[0]);
    } else {
        /* (resolve db tbl) — ignore db, use tbl */
        ray_t* db = ray_eval(args[0]);
        if (db && !RAY_IS_ERR(db)) ray_release(db);
        tbl = ray_eval(args[1]);
    }
    if (!tbl || RAY_IS_ERR(tbl)) return tbl ? tbl : ray_error("type", "resolve: null argument");

    /* Materialize lazy tables.  ray_lazy_materialize CONSUMES the lazy
     * handle (releases its input) — the extra ray_release here was a
     * double free of the handle. */
    if (ray_is_lazy(tbl)) {
        tbl = ray_lazy_materialize(tbl);
        if (!tbl || RAY_IS_ERR(tbl)) return tbl ? tbl : ray_error("domain", "resolve: materialization failed");
    }

    /* If not a table, return as-is */
    if (tbl->type != RAY_TABLE) {
        if (tbl->type == -RAY_SYM) {
            ray_t* val = ray_env_get(tbl->i64);
            ray_release(tbl);
            if (!val) return RAY_NULL_OBJ;
            ray_retain(val);
            return val;
        }
        return tbl;
    }

    int64_t ncols = ray_table_ncols(tbl);
    int64_t nrows = ray_table_nrows(tbl);

    /* Build a new table replacing I64 columns with SYM columns where possible */
    ray_t* result = ray_table_new(ncols);
    if (RAY_IS_ERR(result)) { ray_release(tbl); return result; }

    for (int64_t c = 0; c < ncols; c++) {
        ray_t* col = ray_table_get_col_idx(tbl, c);
        int64_t col_name = ray_table_col_name(tbl, c);
        if (!col) continue;

        if (col->type == RAY_I64) {
            /* Try to resolve: convert to SYM only if ALL positive values
             * are valid sym IDs. This avoids converting entity-ID columns
             * where values are plain integers that happen to collide with
             * low sym IDs. */
            int64_t* data = (int64_t*)ray_data(col);
            bool all_user_sym = (nrows > 0);
            /* Only convert if ALL values resolve to user-defined symbols
             * (length >= 2, not single-char operators). This distinguishes
             * symbol references (name='Alice') from entity IDs (e=1). */
            for (int64_t r = 0; r < nrows; r++) {
                if (data[r] <= 0) { all_user_sym = false; break; }
                ray_t* sn = ray_sym_str(data[r]);
                if (!sn) { all_user_sym = false; break; }
                size_t slen = ray_str_len(sn);
                const char* sp = ray_str_ptr(sn);
                /* Single-char or starts with digit/operator -> not a user symbol */
                if (slen < 2 || (sp[0] >= '0' && sp[0] <= '9') ||
                    sp[0] == '+' || sp[0] == '-' || sp[0] == '*' || sp[0] == '/' ||
                    sp[0] == '<' || sp[0] == '>' || sp[0] == '=' || sp[0] == '!' ||
                    sp[0] == '?' || sp[0] == '_') {
                    all_user_sym = false; break;
                }
            }
            if (all_user_sym) {
                /* Convert to SYM column */
                ray_t* sym_col = ray_vec_new(RAY_SYM, nrows);
                if (RAY_IS_ERR(sym_col)) { ray_release(result); ray_release(tbl); return sym_col; }
                for (int64_t r = 0; r < nrows; r++) {
                    sym_col = ray_vec_append(sym_col, &data[r]);
                    if (RAY_IS_ERR(sym_col)) { ray_release(result); ray_release(tbl); return sym_col; }
                }
                result = ray_table_add_col(result, col_name, sym_col);
                ray_release(sym_col);
            } else {
                /* Keep as I64 */
                result = ray_table_add_col(result, col_name, col);
            }
        } else {
            /* Non-I64 column: keep as-is */
            result = ray_table_add_col(result, col_name, col);
        }
        if (RAY_IS_ERR(result)) { ray_release(tbl); return result; }
    }

    ray_release(tbl);
    return result;
}

/* (timeit expr) — evaluate expression and return time in ms as F64.
 * SPECIAL_FORM: does not pre-evaluate args. */
ray_t* ray_timeit_fn(ray_t** args, int64_t n) {
    if (n < 1) return ray_error("arity", "timeit: expects 1 argument, got %lld", (long long)n);
    int64_t t0 = ray_profile_now_ns();
    ray_t* result = ray_eval(args[0]);
    int64_t t1 = ray_profile_now_ns();
    /* Propagate errors instead of swallowing them — otherwise a failing
     * expression silently reports a timing as if it had succeeded. */
    if (result && RAY_IS_ERR(result)) return result;
    if (result) ray_release(result);
    double ms = (double)(t1 - t0) / 1e6;
    return make_f64(ms);
}

/* (exit code) — exit the process */
ray_t* ray_exit_fn(ray_t* arg) {
#ifdef RAY_FUZZING
    /* A fuzz driver evaluates untrusted input in-process; letting it call
     * exit() would tear the fuzzer down mid-run.  Surface it as an error. */
    (void)arg;
    return ray_error("restricted", "exit disabled under fuzzing");
#else
    int code = 0;
    if (arg && is_numeric(arg)) code = (int)as_i64(arg);
    exit(code);
    return NULL; /* unreachable */
#endif
}

/* (read-csv path) — read CSV file, return RAY_TABLE */
/* Helper: resolve a type name symbol to a ray type code */
static int8_t resolve_type_name(int64_t sym_id) {
    ray_t* s = ray_sym_str(sym_id);
    if (!s) return -1;
    const char* name = ray_str_ptr(s);
    size_t len = ray_str_len(s);
    int8_t result = -1;
    if (len == 3 && memcmp(name, "I64", 3) == 0) result = RAY_I64;
    else if (len == 3 && memcmp(name, "I32", 3) == 0) result = RAY_I32;
    else if (len == 3 && memcmp(name, "I16", 3) == 0) result = RAY_I16;
    /* INT — schema-only marker; csv.c resolves it to the narrowest int width
     * (I16/I32/I64) from the column's parsed min/max/has_null. */
    else if (len == 3 && memcmp(name, "INT", 3) == 0) result = RAY_CSV_AUTO_TAG;
    else if (len == 3 && memcmp(name, "F64", 3) == 0) result = RAY_F64;
    else if (len == 2 && memcmp(name, "B8", 2) == 0) result = RAY_BOOL;
    else if (len == 2 && memcmp(name, "U8", 2) == 0) result = RAY_U8;
    else if (len == 3 && memcmp(name, "SYM", 3) == 0) result = RAY_SYM;
    else if (len == 6 && memcmp(name, "SYMBOL", 6) == 0) result = RAY_SYM;
    else if (len == 3 && memcmp(name, "STR", 3) == 0) result = RAY_STR;
    /* CSV cells are scalar text, so the historical LIST schema spelling is
     * materialized as the native packed STR column rather than a boxed list.
     * This keeps schemas copied from older table displays loadable. */
    else if (len == 4 && memcmp(name, "LIST", 4) == 0) result = RAY_STR;
    else if (len == 3 && memcmp(name, "F32", 3) == 0) result = RAY_F32;
    else if (len == 4 && memcmp(name, "DATE", 4) == 0) result = RAY_DATE;
    else if (len == 4 && memcmp(name, "TIME", 4) == 0) result = RAY_TIME;
    else if (len == 9 && memcmp(name, "TIMESTAMP", 9) == 0) result = RAY_TIMESTAMP;
    else if (len == 4 && memcmp(name, "GUID", 4) == 0) result = RAY_GUID;
    ray_release(s);
    return result;
}

ray_t* ray_read_csv_fn(ray_t** args, int64_t n) {
    if (n < 1 || n > 3) return ray_error("arity", "read-csv: expects 1 to 3 arguments, got %lld", (long long)n);

    /* (read-csv [types] "path"), (read-csv [names] [types] "path"), or (read-csv "path") */
    ray_t* path_obj = NULL;
    ray_t* schema = NULL;
    ray_t* names = NULL;
    if (n >= 3 && ray_is_vec(args[0]) && args[0]->type == RAY_SYM &&
        ray_is_vec(args[1]) && args[1]->type == RAY_SYM) {
        names = args[0];
        schema = args[1];
        path_obj = args[2];
    } else if (n >= 2 && ray_is_vec(args[0]) && args[0]->type == RAY_SYM) {
        schema = args[0];
        path_obj = args[1];
    } else {
        path_obj = args[0];
    }

    const char* path = NULL;
    if (path_obj->type == -RAY_STR)
        path = ray_str_ptr(path_obj);
    else
        return ray_error("type", "read-csv: path must be str, got %s", ray_type_name(path_obj->type));
    if (!path) return ray_error("domain", "read-csv: empty path");

    if (schema) {
        int64_t ncols = schema->len;
        int8_t col_types[256];
        if (ncols > 256) return ray_error("limit", NULL);
        void* sym_data = ray_data(schema);
        for (int64_t i = 0; i < ncols; i++) {
            int64_t sid = ray_read_sym(sym_data, i, schema->type, schema->attrs);
            col_types[i] = resolve_type_name(sid);
            if (col_types[i] < 0) return ray_error("type", "read-csv: unknown column type name in schema");
        }
        int64_t col_names[256];
        if (names) {
            if (names->len != ncols) return ray_error("length", "read-csv: names and types must match, got %lld names and %lld types", (long long)names->len, (long long)ncols);
            void* name_data = ray_data(names);
            for (int64_t i = 0; i < ncols; i++)
                col_names[i] = ray_read_sym(name_data, i, names->type, names->attrs);
        }
        ray_t* tbl = names
            ? ray_read_csv_named_opts(path, 0, false, col_types, (int32_t)ncols,
                                      col_names, (int32_t)ncols)
            : ray_read_csv_opts(path, 0, true, col_types, (int32_t)ncols);
        if (!tbl) return ray_error("io", NULL);
        if (RAY_IS_ERR(tbl)) return tbl;
        return tbl;
    }

    ray_t* tbl = ray_read_csv(path);
    if (!tbl) return ray_error("io", NULL);
    if (RAY_IS_ERR(tbl)) return tbl;
    return tbl;
}

static const char* csv_str_arg(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;
}

static const char* csv_sym_arg(ray_t* s, char* buf, size_t bufsz) {
    if (!s || s->type != -RAY_SYM) return NULL;
    ray_t* text = ray_sym_str(s->i64);
    if (!text) return NULL;
    const char* p = ray_str_ptr(text);
    size_t len = ray_str_len(text);
    if (!p || len == 0 || len >= bufsz) {
        ray_release(text);
        return NULL;
    }
    memcpy(buf, p, len);
    buf[len] = '\0';
    ray_release(text);
    return buf;
}

static int csv_valid_table_name(const char* name) {
    return name && name[0] != '\0' && name[0] != '.' &&
           !strchr(name, '/') && !strchr(name, '\\') && !strstr(name, "..");
}

static const char* csv_default_sym_path(const char* dir, char* buf, size_t bufsz) {
    int n = snprintf(buf, bufsz, "%s/.sym", dir);
    if (n < 0 || (size_t)n >= bufsz) return NULL;
    return buf;
}

ray_t* ray_read_csv_splayed_fn(ray_t** args, int64_t n) {
    if (n < 2 || n > 4) return ray_error("arity", "csv.splayed: expects 2 to 4 arguments, got %lld", (long long)n);

    char dir[1024];
    if (!csv_str_arg(args[n - 1], dir, sizeof(dir))) return ray_error("type", "csv.splayed: target dir must be str, got %s", args[n - 1] ? ray_type_name(args[n - 1]->type) : "null");

    int64_t data_n = n - 1;
    ray_t* path_obj = NULL;
    ray_t* schema = NULL;
    ray_t* names = NULL;
    if (data_n >= 3 && ray_is_vec(args[0]) && args[0]->type == RAY_SYM &&
        ray_is_vec(args[1]) && args[1]->type == RAY_SYM) {
        names = args[0];
        schema = args[1];
        path_obj = args[2];
    } else if (data_n >= 2 && ray_is_vec(args[0]) && args[0]->type == RAY_SYM) {
        schema = args[0];
        path_obj = args[1];
    } else if (data_n == 1) {
        path_obj = args[0];
    } else {
        return ray_error("domain", "csv.splayed: invalid argument arrangement");
    }
    if (!path_obj || path_obj->type != -RAY_STR) return ray_error("type", "csv.splayed: source path must be str, got %s", path_obj ? ray_type_name(path_obj->type) : "null");
    const char* path = ray_str_ptr(path_obj);
    if (!path) return ray_error("domain", "csv.splayed: empty source path");

    int8_t col_types[256];
    int64_t col_names[256];
    int32_t ncols = 0;
    const int8_t* types_arg = NULL;
    const int64_t* names_arg = NULL;
    bool header = true;

    if (schema) {
        if (schema->len > 256) return ray_error("limit", NULL);
        ncols = (int32_t)schema->len;
        void* sym_data = ray_data(schema);
        for (int64_t i = 0; i < schema->len; i++) {
            int64_t sid = ray_read_sym(sym_data, i, schema->type, schema->attrs);
            col_types[i] = resolve_type_name(sid);
            if (col_types[i] < 0) return ray_error("type", "csv.splayed: unknown column type name in schema");
        }
        types_arg = col_types;
        if (names) {
            if (names->len != schema->len) return ray_error("length", "csv.splayed: names and types must match, got %lld names and %lld types", (long long)names->len, (long long)schema->len);
            void* name_data = ray_data(names);
            for (int64_t i = 0; i < names->len; i++)
                col_names[i] = ray_read_sym(name_data, i, names->type, names->attrs);
            names_arg = col_names;
            header = false;
        }
    }

    ray_err_t err = ray_csv_save_splayed_named_opts(path, 0, header,
                                                    types_arg, ncols,
                                                    names_arg, ncols,
                                                    dir, 0);
    if (err != RAY_OK) return ray_error(ray_err_code_str(err), NULL);

    char sym_path[1024];
    const char* sym = csv_default_sym_path(dir, sym_path, sizeof(sym_path));
    if (!sym) return ray_error("io", NULL);

    /* The writer builds indexes before publishing its generation. */
    return ray_read_splayed(dir, sym);
}

ray_t* ray_read_csv_parted_fn(ray_t** args, int64_t n) {
    if (n < 3 || n > 6) return ray_error("arity", "csv.parted: expects 3 to 6 arguments, got %lld", (long long)n);

    char root[1024];
    if (!csv_str_arg(args[n - 2], root, sizeof(root))) return ray_error("type", "csv.parted: root dir must be str, got %s", args[n - 2] ? ray_type_name(args[n - 2]->type) : "null");

    char table_name[256];
    if (!csv_sym_arg(args[n - 1], table_name, sizeof(table_name)))
        return ray_error("type", "csv.parted: table name must be sym, got %s", args[n - 1] ? ray_type_name(args[n - 1]->type) : "null");
    if (!csv_valid_table_name(table_name)) return ray_error("domain", "csv.parted: invalid table name");

    int64_t rows_per_part = 0;
    int64_t data_n = n - 2;
    if (data_n >= 2 && is_numeric(args[n - 3])) {
        rows_per_part = as_i64(args[n - 3]);
        if (rows_per_part <= 0) return ray_error("range", "csv.parted: rows per partition must be positive, got %lld", (long long)rows_per_part);
        data_n--;
    }
    ray_t* path_obj = NULL;
    ray_t* schema = NULL;
    ray_t* names = NULL;
    if (data_n >= 3 && ray_is_vec(args[0]) && args[0]->type == RAY_SYM &&
        ray_is_vec(args[1]) && args[1]->type == RAY_SYM) {
        names = args[0];
        schema = args[1];
        path_obj = args[2];
    } else if (data_n >= 2 && ray_is_vec(args[0]) && args[0]->type == RAY_SYM) {
        schema = args[0];
        path_obj = args[1];
    } else if (data_n == 1) {
        path_obj = args[0];
    } else {
        return ray_error("domain", "csv.parted: invalid argument arrangement");
    }

    if (!path_obj || path_obj->type != -RAY_STR) return ray_error("type", "csv.parted: source path must be str, got %s", path_obj ? ray_type_name(path_obj->type) : "null");
    const char* path = ray_str_ptr(path_obj);
    if (!path) return ray_error("domain", "csv.parted: empty source path");

    int8_t col_types[256];
    int64_t col_names[256];
    int32_t ncols = 0;
    const int8_t* types_arg = NULL;
    const int64_t* names_arg = NULL;
    bool header = true;

    if (schema) {
        if (schema->len > 256) return ray_error("limit", NULL);
        ncols = (int32_t)schema->len;
        void* sym_data = ray_data(schema);
        for (int64_t i = 0; i < schema->len; i++) {
            int64_t sid = ray_read_sym(sym_data, i, schema->type, schema->attrs);
            col_types[i] = resolve_type_name(sid);
            if (col_types[i] < 0) return ray_error("type", "csv.parted: unknown column type name in schema");
        }
        types_arg = col_types;
        if (names) {
            if (names->len != schema->len) return ray_error("length", "csv.parted: names and types must match, got %lld names and %lld types", (long long)names->len, (long long)schema->len);
            void* name_data = ray_data(names);
            for (int64_t i = 0; i < names->len; i++)
                col_names[i] = ray_read_sym(name_data, i, names->type, names->attrs);
            names_arg = col_names;
            header = false;
        }
    }

    /* A new root is staged in <root>.csv-partial, which a failed import
     * leaves behind for diagnosis and which then blocks the root.  Name it
     * in the error rather than returning a bare `io`. */
    char dest[1024], staging[1100];
    bool stale = false;
    if (ray_csv_parted_paths(root, dest, sizeof(dest), staging, sizeof(staging)) == RAY_OK) {
        struct stat st;
        stale = stat(dest, &st) != 0 && stat(staging, &st) == 0;
    }
    if (stale)
        return ray_error("io", "csv.parted: %s is left from an earlier failed import; remove it to retry",
                         staging);

    ray_err_t err = ray_csv_save_parted_named_opts(path, 0, header,
                                                   types_arg, ncols,
                                                   names_arg, ncols,
                                                   root, table_name, rows_per_part);
    if (err != RAY_OK) {
        struct stat st;
        if (stat(dest, &st) != 0 && stat(staging, &st) == 0)
            return ray_error(ray_err_code_str(err), "csv.parted: import failed; partial output retained in %s",
                             staging);
        return ray_error(ray_err_code_str(err), NULL);
    }

    ray_t* out = ray_read_parted(root, table_name);
    if (getenv("RAY_CSV_TRACE") && out && RAY_IS_ERR(out)) {
        fprintf(stderr, "csv.parted: read_parted failure root=%s table=%s err=%s\n",
                root, table_name, ray_err_code(out));
    }
    return out;
}

/* (write-csv table path) — write table to CSV file */
ray_t* ray_write_csv_fn(ray_t** args, int64_t n) {
#ifdef RAY_FUZZING
    /* No filesystem writes to attacker-controlled paths under fuzzing. */
    (void)args; (void)n;
    return ray_error("restricted", "write-csv disabled under fuzzing");
#endif
    if (n < 2) return ray_error("arity", "write-csv: expects 2 arguments, got %lld", (long long)n);
    ray_t* tbl = args[0];
    ray_t* path_obj = args[1];
    if (tbl->type != RAY_TABLE) return ray_error("type", "write-csv: first argument must be table, got %s", ray_type_name(tbl->type));
    const char* path = NULL;
    if (path_obj->type == -RAY_STR)
        path = ray_str_ptr(path_obj);
    else
        return ray_error("type", "write-csv: path must be str, got %s", ray_type_name(path_obj->type));
    if (!path) return ray_error("domain", "write-csv: empty path");
    ray_err_t err = ray_write_csv(tbl, path);
    if (err != RAY_OK) return ray_error(ray_err_code_str(err), NULL);
    return make_i64(0);
}

/* (as 'TypeName value) — type cast */
/* Case-insensitive type name match helper */
static int cast_match(const char* tname, size_t tlen, const char* target) {
    size_t tgt_len = strlen(target);
    if (tlen != tgt_len) return 0;
    for (size_t i = 0; i < tlen; i++) {
        char a = tname[i], b = target[i];
        if (a >= 'a' && a <= 'z') a -= 32;
        if (b >= 'a' && b <= 'z') b -= 32;
        if (a != b) return 0;
    }
    return 1;
}

/* Helper: copy null state from source vec/list to destination vec. */
static ray_t* cast_vec_copy_nulls(ray_t* vec, ray_t* val) {
    /* BOOL / U8 destinations are non-nullable — there is no slot for a
     * null marker.  Casting a nullable source to one of these types
     * silently collapses the null to the type's zero value (already
     * written by the cast loop). */
    if (vec->type == RAY_BOOL || vec->type == RAY_U8) return vec;

    if (ray_is_vec(val)) {
        if (ray_vec_copy_nulls(vec, val) != RAY_OK)
            { ray_release(vec); return ray_error("oom", NULL); }
    } else if (val->type == RAY_LIST) {
        ray_t** le = (ray_t**)ray_data(val);
        for (int64_t j = 0; j < vec->len; j++)
            if (le[j] && RAY_ATOM_IS_NULL(le[j]))
                ray_vec_set_null(vec, j, true);
        /* ray_vec_set_null writes the sentinel into the payload, so the
         * LIST branch needs no post-fill. */
        return vec;
    }
    /* VEC source: ray_vec_copy_nulls bulk-copies the source's HAS_NULLS
     * state into the destination, but never touches the payload — the
     * cast loop already filled the dest payload with raw cast results.
     * For each null source slot, overwrite the dest payload with the
     * dest-width sentinel so consumers reading the raw payload honor the
     * null contract.  Narrowing casts require writing the dest-width
     * sentinel directly — propagating through the cast macro produces
     * (int16_t)NULL_I32 = 0 etc., which collides with a legitimate value. */
    if (ray_vec_may_have_nulls(val)) {
        switch (vec->type) {
            case RAY_F64: {
                double* d = (double*)ray_data(vec);
                for (int64_t j = 0; j < vec->len; j++)
                    if (ray_vec_is_null(val, j)) d[j] = NULL_F64;
                break;
            }
            case RAY_I64: case RAY_TIMESTAMP: {
                int64_t* d = (int64_t*)ray_data(vec);
                for (int64_t j = 0; j < vec->len; j++)
                    if (ray_vec_is_null(val, j)) d[j] = NULL_I64;
                break;
            }
            case RAY_I32: case RAY_DATE: case RAY_TIME: {
                int32_t* d = (int32_t*)ray_data(vec);
                for (int64_t j = 0; j < vec->len; j++)
                    if (ray_vec_is_null(val, j)) d[j] = NULL_I32;
                break;
            }
            case RAY_I16: {
                int16_t* d = (int16_t*)ray_data(vec);
                for (int64_t j = 0; j < vec->len; j++)
                    if (ray_vec_is_null(val, j)) d[j] = NULL_I16;
                break;
            }
            default: break;
        }
    }
    return vec;
}

/* Bulk-cast loop over [_lo, _hi).  Reads `R` from `_src_p`, writes `W`
 * to `_dst_p`.  No atom allocations.  The single-threaded path passes
 * the whole [0, n2) range; the parallel worker passes its slice. */
#define CAST_LOOP_RANGE(R, W, EXPR, _lo, _hi) do {                     \
    const R* _src = (const R*)_src_p;                                  \
    W* _dst = (W*)_dst_p;                                              \
    for (int64_t _i = (_lo); _i < (_hi); _i++) {                       \
        R _v = _src[_i];                                               \
        _dst[_i] = (EXPR);                                             \
    }                                                                  \
} while (0)
#define CAST_LOOP(R, W, EXPR) CAST_LOOP_RANGE(R, W, EXPR, 0, n2)

/* Same-byte-rep type relabels (I64↔TIMESTAMP, I32↔DATE↔TIME): the
 * per-element data is identical, so a single memcpy populates the new
 * vector.  Returns true on hit. */
static bool cast_vec_relabel_compat(int8_t a, int8_t b) {
    if (a == b) return true;
    if ((a == RAY_I64 || a == RAY_TIMESTAMP) &&
        (b == RAY_I64 || b == RAY_TIMESTAMP)) return true;
    if ((a == RAY_I32 || a == RAY_DATE || a == RAY_TIME) &&
        (b == RAY_I32 || b == RAY_DATE || b == RAY_TIME)) return true;
    return false;
}

/* Vec→vec numeric cast on raw arrays (no per-element atom allocs).
 * Returns the populated `vec` on success, or NULL if the (in_type,
 * out_type) pair is unsupported here — caller falls back to the generic
 * path.
 *
 * Temporal cross-unit pairs (matched between the per-atom slow path
 * and the fast path):
 *   DATE → TIMESTAMP : days * NS_PER_DAY
 *   TIMESTAMP → DATE : floor-div by NS_PER_DAY (so ns=-1 → -1 day,
 *                       i.e. 1999-12-31, not 2000-01-01).
 *   TIMESTAMP → TIME : floor-mod by NS_PER_DAY then /1_000_000
 *                       (ns→ms within day, always in [0, 86_400_000)).
 * Plain `% / /` would truncate toward zero per C semantics and give
 * wrong components for pre-2000 timestamps; the helpers below give
 * Python-style floor semantics for a positive divisor. */
#define NS_PER_DAY 86400000000000LL

static inline int64_t ts_days_floor(int64_t ns) {
    int64_t q = ns / NS_PER_DAY;
    int64_t r = ns - q * NS_PER_DAY;
    if (r < 0) q -= 1;
    return q;
}
static inline int64_t ts_ns_in_day(int64_t ns) {
    int64_t r = ns % NS_PER_DAY;
    if (r < 0) r += NS_PER_DAY;
    return r;
}

/* Element-wise cast worker: writes _dst_p[lo..hi) from _src_p[lo..hi).
 * Used by both the single-threaded fast path and the parallel dispatch.
 * Returns true on hit; false means caller falls back to the generic
 * (atom) path. */
static bool cast_range_worker(const void* _src_p, void* _dst_p,
                              int64_t lo, int64_t hi,
                              int8_t in_type, int8_t out_type) {
    /* Temporal unit conversions. */
    if (in_type == RAY_DATE && out_type == RAY_TIMESTAMP) {
        CAST_LOOP_RANGE(int32_t, int64_t, (int64_t)_v * NS_PER_DAY, lo, hi);
        return true;
    }
    if (in_type == RAY_TIMESTAMP && out_type == RAY_DATE) {
        /* Floor-div, not truncate-toward-zero: ns=-1 must give -1 day
         * (1999-12-31), not 0 (2000-01-01). */
        CAST_LOOP_RANGE(int64_t, int32_t, (int32_t)ts_days_floor(_v), lo, hi);
        return true;
    }
    if (in_type == RAY_TIMESTAMP && out_type == RAY_TIME) {
        /* Floor-mod ns within day, then ns→ms. */
        CAST_LOOP_RANGE(int64_t, int32_t,
                        (int32_t)(ts_ns_in_day(_v) / 1000000LL), lo, hi);
        return true;
    }
    /* Generic numeric pairs.  The big switch dispatches on (out_type,
     * in_type); each leaf is a tight typed loop the compiler vectorizes. */
#define CL(R, W, EXPR) do { CAST_LOOP_RANGE(R, W, EXPR, lo, hi); return true; } while (0)
    switch (out_type) {
    case RAY_I64: case RAY_TIMESTAMP:
        switch (in_type) {
        case RAY_BOOL:  CL(uint8_t,  int64_t, _v ? 1 : 0);
        case RAY_U8:    CL(uint8_t,  int64_t, (int64_t)_v);
        case RAY_I16:   CL(int16_t,  int64_t, (int64_t)_v);
        case RAY_I32: case RAY_DATE: case RAY_TIME:
                        CL(int32_t,  int64_t, (int64_t)_v);
        case RAY_F64:   CL(double,   int64_t, ray_cast_f64_to_i64_null(_v));
        }
        break;
    case RAY_I32: case RAY_DATE: case RAY_TIME:
        switch (in_type) {
        case RAY_BOOL:  CL(uint8_t,  int32_t, _v ? 1 : 0);
        case RAY_U8:    CL(uint8_t,  int32_t, (int32_t)_v);
        case RAY_I16:   CL(int16_t,  int32_t, (int32_t)_v);
        case RAY_I64: case RAY_TIMESTAMP:
                        CL(int64_t,  int32_t, (int32_t)_v);
        case RAY_F64:   CL(double,   int32_t, ray_cast_f64_to_i32_null(_v));
        }
        break;
    case RAY_I16:
        switch (in_type) {
        case RAY_BOOL:  CL(uint8_t,  int16_t, _v ? 1 : 0);
        case RAY_U8:    CL(uint8_t,  int16_t, (int16_t)_v);
        case RAY_I32: case RAY_DATE: case RAY_TIME:
                        CL(int32_t,  int16_t, (int16_t)_v);
        case RAY_I64: case RAY_TIMESTAMP:
                        CL(int64_t,  int16_t, (int16_t)_v);
        case RAY_F64:   CL(double,   int16_t, ray_cast_f64_to_i16_null(_v));
        }
        break;
    case RAY_U8:
        switch (in_type) {
        case RAY_BOOL:  CL(uint8_t,  uint8_t, _v ? 1 : 0);
        case RAY_I16:   CL(int16_t,  uint8_t, (uint8_t)_v);
        case RAY_I32: case RAY_DATE: case RAY_TIME:
                        CL(int32_t,  uint8_t, (uint8_t)_v);
        case RAY_I64: case RAY_TIMESTAMP:
                        CL(int64_t,  uint8_t, (uint8_t)_v);
        case RAY_F64:   CL(double,   uint8_t, ray_cast_f64_to_u8_null(_v));
        }
        break;
    case RAY_F64:
        switch (in_type) {
        case RAY_BOOL:  CL(uint8_t,  double, _v ? 1.0 : 0.0);
        case RAY_U8:    CL(uint8_t,  double, (double)_v);
        case RAY_I16:   CL(int16_t,  double, (double)_v);
        case RAY_I32: case RAY_DATE: case RAY_TIME:
                        CL(int32_t,  double, (double)_v);
        case RAY_I64: case RAY_TIMESTAMP:
                        CL(int64_t,  double, (double)_v);
        }
        break;
    case RAY_BOOL:
        switch (in_type) {
        case RAY_U8:    CL(uint8_t,  uint8_t, _v != 0 ? 1 : 0);
        case RAY_I16:   CL(int16_t,  uint8_t, _v != 0 ? 1 : 0);
        case RAY_I32: case RAY_DATE: case RAY_TIME:
                        CL(int32_t,  uint8_t, _v != 0 ? 1 : 0);
        case RAY_I64: case RAY_TIMESTAMP:
                        CL(int64_t,  uint8_t, _v != 0 ? 1 : 0);
        case RAY_F64:   CL(double,   uint8_t, ray_cast_f64_to_bool_null(_v));
        }
        break;
    }
#undef CL
    return false;
}

typedef struct {
    const void* src;
    void*       dst;
    int8_t      in_type;
    int8_t      out_type;
} cast_par_ctx_t;

static void cast_par_fn(void* arg, uint32_t worker_id, int64_t lo, int64_t hi) {
    (void)worker_id;
    /* Honor SIGINT (ray_request_interrupt / ray_interrupted) per task —
     * the pool's own per-task gate checks `pool->cancelled` only, so
     * a Ctrl-C arriving during dispatch wouldn't otherwise short-
     * circuit the workers.  Skip the task on interrupt; the caller
     * post-checks via CANCELLED() and returns an error. */
    if (ray_interrupted()) return;
    cast_par_ctx_t* ctx = (cast_par_ctx_t*)arg;
    cast_range_worker(ctx->src, ctx->dst, lo, hi, ctx->in_type, ctx->out_type);
}

/* Threshold below which the dispatch overhead outweighs the speedup.
 * Memory-bound conversions saturate ~3 GB/s single-thread; with 8
 * workers we approach DRAM peak (~25 GB/s).  Below ~256 K elements the
 * 50 µs dispatch cost dominates. */
#define CAST_PAR_MIN_ELEMS 262144

static ray_t* cast_vec_numeric_fast(ray_t* val, ray_t* vec, int8_t out_type) {
    int8_t in_type = val->type;
    int64_t n2 = val->len;
    ray_pool_t* pool = ray_pool_get();

/* A cast is "cancelled" if EITHER:
 *   (a) the pool's per-query cancel flag is set (e.g. via ray_cancel
 *       from another thread or a long-query timeout), or
 *   (b) the eval-loop interrupt flag is set (Ctrl-C / SIGINT, signalled
 *       by ray_request_interrupt and observed via ray_interrupted).
 * Both must be polled — they're independent signals and either one
 * means the user wants the operation to abort. */
#define CANCELLED() ((pool && atomic_load_explicit(&pool->cancelled,   \
                                                   memory_order_acquire)) \
                     || ray_interrupted())
#define CHECK_CANCEL_OR(retval) do {                                   \
    if (CANCELLED()) return ray_error("cancel", NULL);                 \
    return (retval);                                                   \
} while (0)

    /* Function-entry cancel check — gates ALL paths below (relabel,
     * parallel, and chunked single-thread).  Without this, a cancel
     * pending at entry would still execute the first ~50 µs of any
     * path before being observed. */
    if (CANCELLED()) return ray_error("cancel", NULL);

    /* Same byte-rep types: chunked memcpy.  A single
     * memcpy(_, _, n*esz) on a 10M-element TIMESTAMP relabel is ~80 MB
     * and ~10 ms of opaque work — cancel arriving during it can't
     * interrupt the libc copy, so we'd happily return `vec` even if
     * the user asked to abort.  Break the copy into ~1 MB chunks and
     * poll cancel between them; max in-flight work between checks is
     * one chunk (~100 µs at realistic bandwidth). */
    if (cast_vec_relabel_compat(in_type, out_type)) {
        size_t esz = (size_t)ray_elem_size(out_type);
        if (n2 > 0 && esz > 0) {
            const char* sp = (const char*)ray_data(val);
            char* dp = (char*)ray_data(vec);
            size_t total = (size_t)n2 * esz;
            const size_t chunk_bytes = (size_t)1 << 20;  /* 1 MiB */
            size_t off = 0;
            while (off < total) {
                if (CANCELLED()) return ray_error("cancel", NULL);
                size_t cn = total - off;
                if (cn > chunk_bytes) cn = chunk_bytes;
                memcpy(dp + off, sp + off, cn);
                off += cn;
            }
        }
        /* Post-check: a cancel landing in the final chunk would have
         * been missed by the in-loop check (we copy then exit). */
        if (CANCELLED()) return ray_error("cancel", NULL);
        return vec;
    }

    const void* src_p = ray_data(val);
    void* dst_p = ray_data(vec);

    /* Three return states from this point on (helper does NOT touch
     * `vec`'s reference count):
     *
     *   - `vec`           : success, fully populated, no cancel observed
     *   - error pointer   : cancellation observed at any point — the
     *                       helper bails out as soon as it notices,
     *                       even mid-loop in the single-thread path
     *   - NULL            : (in_type, out_type) pair unsupported here
     *                       AND no cancellation observed — caller may
     *                       safely fall through to the per-atom slow
     *                       path with `vec` still valid */

    if (pool && n2 >= CAST_PAR_MIN_ELEMS && ray_pool_total_workers(pool) >= 2) {
        cast_par_ctx_t pctx = { .src = src_p, .dst = dst_p,
                                .in_type = in_type, .out_type = out_type };
        /* Probe the worker on a single element to verify the pair is
         * supported here.  If unsupported, fall through (NULL) — but
         * still re-check cancel first so a cancel raced into the probe
         * window is not swallowed. */
        if (n2 > 0 && cast_range_worker(src_p, dst_p, 0, 1, in_type, out_type)) {
            ray_pool_dispatch(pool, cast_par_fn, &pctx, n2);
            if (CANCELLED()) return ray_error("cancel", NULL);
            return vec;
        }
        CHECK_CANCEL_OR(NULL);
    }

    /* Chunked single-thread path.  Tight typed loops vectorize well
     * but block cancellation for the whole `n2` range — chunk into
     * cache-sized pieces so cancel is honored within ~one chunk
     * (64K elements ≈ 50 µs at realistic bandwidth). */
    if (n2 == 0)
        CHECK_CANCEL_OR(vec);
    /* Re-check cancel right before the first chunk runs (entry cancel
     * check above is over the whole helper, but if a cancel raced in
     * between the relabel path and here we want to bail before doing
     * any work). */
    if (CANCELLED()) return ray_error("cancel", NULL);
    int64_t chunk = (int64_t)65536;
    int64_t lo = 0;
    int64_t hi = (n2 < chunk) ? n2 : chunk;
    /* Probe the first chunk; if it fails, the (in, out) pair is
     * unsupported here and the caller falls through. */
    if (!cast_range_worker(src_p, dst_p, lo, hi, in_type, out_type))
        CHECK_CANCEL_OR(NULL);
    lo = hi;
    while (lo < n2) {
        if (CANCELLED()) return ray_error("cancel", NULL);
        hi = lo + chunk;
        if (hi > n2) hi = n2;
        cast_range_worker(src_p, dst_p, lo, hi, in_type, out_type);
        lo = hi;
    }
    CHECK_CANCEL_OR(vec);
#undef CHECK_CANCEL_OR
#undef CANCELLED
}

/* Helper: cast a vector/list to a numeric/temporal/bool type.
 * Handles I64, I32, I16, U8, F64, BOOL, DATE, TIME, TIMESTAMP, SYM.
 * Fast path for typed numeric input vectors (no per-element atoms);
 * generic path for RAY_LIST and other shapes. */
/* Null propagation: a cast can land the target type's RESERVED
 * null sentinel in the output without it being a *source* null — e.g.
 * STR→F64 "inf"→0Nf, or an out-of-domain value mapping to NULL_I64.
 * cast_vec_copy_nulls only carries source nulls, so scan the output and
 * flip HAS_NULLS for any reserved sentinel.
 *
 * Scope MATCHES the validator (src/vec/validate.c): only F64/F32/I64/
 * TIMESTAMP sentinels are reserved.  I16/I32 (and I32-backed DATE/TIME) are
 * DELIBERATELY excluded — narrow-int arithmetic/truncation wraps modulo the
 * width (documented wrap rule), so INT16_MIN/INT32_MIN are legitimate
 * computed values, not nulls (see test/rfl/expr/narrow_binary.rfl).  Forcing
 * HAS_NULLS for them would make a wrapped result read back as null. */
static void cast_mark_output_sentinels(ray_t* vec, int8_t out_type, int64_t n) {
    const void* out = ray_data(vec);
    switch (out_type) {
        case RAY_F64: {
            const double* d = (const double*)out;
            for (int64_t i = 0; i < n; i++)
                if (d[i] != d[i]) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_F32: {
            const float* d = (const float*)out;
            for (int64_t i = 0; i < n; i++)
                if (d[i] != d[i]) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_I64: case RAY_TIMESTAMP: {
            const int64_t* d = (const int64_t*)out;
            for (int64_t i = 0; i < n; i++)
                if (d[i] == NULL_I64) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_SYM:
            for (int64_t i = 0; i < n; i++) {
                if (ray_read_sym(out, i, RAY_SYM, vec->attrs) == 0) {
                    vec->attrs |= RAY_ATTR_HAS_NULLS;
                    return;
                }
            }
            return;
        default: return;
    }
}

static ray_t* cast_vec_numeric(ray_t* type_sym, ray_t* val, int8_t out_type) {
    int64_t n2 = val->len;
    ray_t* vec = ray_vec_new(out_type, n2);
    if (RAY_IS_ERR(vec)) return vec;
    vec->len = n2;

    /* Fast path: typed numeric vec → numeric vec, no list/string. */
    if (ray_is_vec(val) && val->type != RAY_STR && val->type != RAY_SYM &&
        val->type != RAY_GUID && out_type != RAY_SYM) {
        ray_t* fast = cast_vec_numeric_fast(val, vec, out_type);
        /* Three return states (helper does NOT release `vec`):
         *   - vec on success
         *   - error pointer on cancellation — caller releases `vec`
         *   - NULL on unsupported (in_type, out_type) — fall through */
        if (RAY_IS_ERR(fast)) { ray_release(vec); return fast; }
        if (fast != NULL) {
            /* Close the cancellation gap that surrounds the post-cast
             * null-bitmap copy.  cast_vec_copy_nulls runs after the
             * cancel-aware fast cast — for nullable inputs it does a
             * bitmap copy (and a per-element scan on RAY_LIST inputs
             * of length n2).  A cancel arriving in that window would
             * otherwise be masked by the success return.  Pre-check
             * gates the null-bitmap work; post-check catches a cancel
             * landing during it. */
            ray_pool_t* fp = ray_pool_get();
#define _FP_CANCELLED() ((fp && atomic_load_explicit(&fp->cancelled, \
                                                     memory_order_acquire)) \
                         || ray_interrupted())
            if (_FP_CANCELLED()) { ray_release(vec); return ray_error("cancel", NULL); }
            ray_t* result = cast_vec_copy_nulls(vec, val);
            if (RAY_IS_ERR(result)) return result;
            if (_FP_CANCELLED()) { ray_release(vec); return ray_error("cancel", NULL); }
#undef _FP_CANCELLED
            /* Null propagation: a narrowing fast cast may have produced a
             * reserved sentinel not present in the source. */
            cast_mark_output_sentinels(vec, out_type, n2);
            return vec;
        }
    }

    /* Fast path: STR vec → SYM vec.  Direct intern from each element's
     * (ptr, len), no atom alloc or recursive cast.  ray_sym_intern uses
     * the table's coarse lock so this stays single-threaded — but it
     * skips ~150 ns of overhead per row. */
    if (out_type == RAY_SYM && ray_is_vec(val) && val->type == RAY_STR) {
        int64_t* ids = (int64_t*)ray_data(vec);
        for (int64_t i = 0; i < n2; i++) {
            size_t slen = 0;
            const char* sp = ray_str_vec_get(val, i, &slen);
            int64_t id = ray_sym_intern(sp ? sp : "", sp ? slen : 0);
            if (id < 0) { ray_release(vec); return ray_error("oom", NULL); }
            ids[i] = id;
        }
        ray_t* result = cast_vec_copy_nulls(vec, val);
        if (RAY_IS_ERR(result)) return result;
        return vec;
    }

    void* out = ray_data(vec);
    for (int64_t i = 0; i < n2; i++) {
        int alloc = 0;
        ray_t* elem = collection_elem(val, i, &alloc);
        if (RAY_IS_ERR(elem)) { ray_release(vec); return elem; }
        ray_t* cast = ray_cast_fn(type_sym, elem);
        if (alloc) ray_release(elem);
        if (RAY_IS_ERR(cast)) { ray_release(vec); return cast; }
        switch (out_type) {
        case RAY_I64: case RAY_TIMESTAMP: case RAY_SYM:
            ((int64_t*)out)[i] = cast->i64; break;
        case RAY_I32: case RAY_DATE: case RAY_TIME:
            ((int32_t*)out)[i] = cast->i32; break;
        case RAY_I16:  ((int16_t*)out)[i] = cast->i16; break;
        case RAY_U8:   ((uint8_t*)out)[i] = cast->u8;  break;
        case RAY_F64:  ((double*)out)[i]  = cast->f64;  break;
        case RAY_BOOL: ((bool*)out)[i]    = cast->b8;   break;
        default: break;
        }
        ray_release(cast);
    }
    /* STAGE 2 (ingest/cast → F64 vector): the per-element recursion routes
     * STR→F64 through make_f64, which canonicalizes any non-finite parse
     * (strtod "inf"/"1e400"/"nan") to NULL_F64 (0Nf).  cast_vec_copy_nulls
     * below only propagates *source* nulls, so a 0Nf produced from a
     * non-null source cell would not flip HAS_NULLS.  Scan the F64 output
     * and set HAS_NULLS if any canonical 0Nf was produced. */
    /* Null propagation: scan the output for any reserved sentinel produced by
     * the cast (narrowing/truncation, or STR→F64 0Nf canonicalization) that
     * is not a *source* null cast_vec_copy_nulls would carry. */
    cast_mark_output_sentinels(vec, out_type, n2);
    ray_t* result = cast_vec_copy_nulls(vec, val);
    if (RAY_IS_ERR(result)) return result;
    return vec;
}

ray_t* ray_cast_fn(ray_t* type_sym, ray_t* val) {
    if (type_sym->type != -RAY_SYM) return ray_error("type", "as: target type must be sym, got %s", ray_type_name(type_sym->type));
    /* Null propagation: casting a typed null atom produces a typed null of target type */
    if (ray_is_atom(val) && RAY_ATOM_IS_NULL(val)) {
        ray_t* s2 = ray_sym_str(type_sym->i64);
        if (!s2) return ray_error("domain", NULL);
        const char* tn = ray_str_ptr(s2);
        size_t tl = ray_str_len(s2);
        int8_t tt = 0;
        if (cast_match(tn, tl, "I64") || cast_match(tn, tl, "i64")) tt = -RAY_I64;
        else if (cast_match(tn, tl, "I32") || cast_match(tn, tl, "i32")) tt = -RAY_I32;
        else if (cast_match(tn, tl, "I16") || cast_match(tn, tl, "i16")) tt = -RAY_I16;
        else if (cast_match(tn, tl, "U8") || cast_match(tn, tl, "u8")) tt = -RAY_U8;
        else if (cast_match(tn, tl, "F64") || cast_match(tn, tl, "f64")) tt = -RAY_F64;
        else if (cast_match(tn, tl, "BOOL") || cast_match(tn, tl, "bool") || cast_match(tn, tl, "B8") || cast_match(tn, tl, "b8")) tt = -RAY_BOOL;
        else if (cast_match(tn, tl, "SYMBOL") || cast_match(tn, tl, "symbol") || cast_match(tn, tl, "sym")) tt = -RAY_SYM;
        else if (cast_match(tn, tl, "DATE") || cast_match(tn, tl, "date")) tt = -RAY_DATE;
        else if (cast_match(tn, tl, "TIME") || cast_match(tn, tl, "time")) tt = -RAY_TIME;
        else if (cast_match(tn, tl, "TIMESTAMP") || cast_match(tn, tl, "timestamp")) tt = -RAY_TIMESTAMP;
        else if (cast_match(tn, tl, "GUID") || cast_match(tn, tl, "guid")) tt = -RAY_GUID;
        else if (cast_match(tn, tl, "STR") || cast_match(tn, tl, "str")) { ray_release(s2); return ray_str("", 0); }
        ray_release(s2);
        if (tt) return ray_typed_null(tt);
        return ray_error("domain", "as: unknown target type name");
    }
    ray_t* s = ray_sym_str(type_sym->i64);
    if (!s) return ray_error("domain", NULL);
    const char* tname = ray_str_ptr(s);
    size_t tlen = ray_str_len(s);

    /* Cast to I64 / i64 */
    if (cast_match(tname, tlen, "I64") || cast_match(tname, tlen, "i64")) {
        ray_release(s);
        if (val->type == -RAY_I64) { ray_retain(val); return val; }
        if (val->type == -RAY_F64) return make_i64(ray_cast_f64_to_i64_null(val->f64));
        if (val->type == -RAY_BOOL) return make_i64(val->b8 ? 1 : 0);
        if (val->type == -RAY_I32 || val->type == -RAY_DATE || val->type == -RAY_TIME)
            return make_i64(val->i32);
        if (val->type == -RAY_TIMESTAMP) return make_i64(val->i64);
        if (val->type == -RAY_I16) return make_i64(val->i16);
        if (val->type == -RAY_U8) return make_i64(val->u8);
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val);
            if (!sp) return ray_error("domain", "as: cannot parse empty str as i64");
            char* end;
            errno = 0;
            int64_t v = strtoll(sp, &end, 10);
            if (end == sp) return ray_error("domain", "as: cannot parse str as i64");
            if (*end != '\0')
                return ray_error("domain", "as: cannot parse str as i64, unexpected trailing characters");
            if (errno == ERANGE)
                return ray_error("domain", "as: cannot parse str as i64, value out of int64 range");
            /* INT64_MIN is NULL_I64: a non-null i64 cannot hold it, so reject
             * rather than silently returning 0Nl.  The string->vector cast
             * (cast_vec_numeric) routes each element through here too, so
             * vectors reject it consistently — string ingest is loud, unlike
             * numeric narrowing which nulls the sentinel (see as.rfl). */
            if (v == NULL_I64)
                return ray_error("domain", "as: cannot parse str as i64, value is the i64 null sentinel");
            return make_i64(v);
        }
        /* Vector/list cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_I64);
        return ray_error("type", "as: cannot cast %s to i64", ray_type_name(val->type));
    }
    /* Cast to I32 / i32 */
    if (cast_match(tname, tlen, "I32") || cast_match(tname, tlen, "i32")) {
        ray_release(s);
        if (val->type == -RAY_I32) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return ray_i32(val->b8 ? 1 : 0);
        if (val->type == -RAY_U8)  return ray_i32((int32_t)val->u8);
        if (val->type == -RAY_I16) return ray_i32(val->i16);
        if (val->type == -RAY_I64) return ray_i32((int32_t)val->i64);
        if (val->type == -RAY_F64) return ray_i32(ray_cast_f64_to_i32_null(val->f64));
        if (val->type == -RAY_DATE || val->type == -RAY_TIME) return ray_i32(val->i32);
        if (val->type == -RAY_TIMESTAMP) return ray_i32((int32_t)val->i64);
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val); char* end;
            errno = 0;
            int64_t v = strtoll(sp, &end, 10);
            if (end == sp) return ray_error("domain", "as: cannot parse str as i32");
            if (*end != '\0')
                return ray_error("domain", "as: cannot parse str as i32, unexpected trailing characters");
            if (errno == ERANGE)
                return ray_error("domain", "as: cannot parse str as i32, value out of int64 range");
            if (v < INT32_MIN || v > INT32_MAX)
                return ray_error("domain", "as: cannot parse str as i32, value out of i32 range");
            /* INT32_MIN is NULL_I32: a non-null i32 cannot hold it, so reject
             * rather than silently returning 0Ni.  The string->vector cast
             * (cast_vec_numeric) routes each element through here too, so
             * vectors reject it consistently — string ingest is loud, unlike
             * numeric narrowing which nulls the sentinel (see as.rfl). */
            if ((int32_t)v == NULL_I32)
                return ray_error("domain", "as: cannot parse str as i32, value is the i32 null sentinel");
            return ray_i32((int32_t)v);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_I32);
        return ray_error("type", "as: cannot cast %s to i32", ray_type_name(val->type));
    }
    /* Cast to I16 / i16 */
    if (cast_match(tname, tlen, "I16") || cast_match(tname, tlen, "i16")) {
        ray_release(s);
        if (val->type == -RAY_I16) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return ray_i16(val->b8 ? 1 : 0);
        if (val->type == -RAY_U8)  return ray_i16((int16_t)val->u8);
        if (val->type == -RAY_I32) return ray_i16((int16_t)val->i32);
        if (val->type == -RAY_I64) return ray_i16((int16_t)val->i64);
        if (val->type == -RAY_F64) return ray_i16(ray_cast_f64_to_i16_null(val->f64));
        if (val->type == -RAY_DATE || val->type == -RAY_TIME) return ray_i16((int16_t)val->i32);
        if (val->type == -RAY_TIMESTAMP) return ray_i16((int16_t)val->i64);
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val); char* end;
            errno = 0;
            int64_t v = strtoll(sp, &end, 10);
            if (end == sp) return ray_error("domain", "as: cannot parse str as i16");
            if (*end != '\0')
                return ray_error("domain", "as: cannot parse str as i16, unexpected trailing characters");
            if (errno == ERANGE)
                return ray_error("domain", "as: cannot parse str as i16, value out of int64 range");
            if (v < INT16_MIN || v > INT16_MAX)
                return ray_error("domain", "as: cannot parse str as i16, value out of i16 range");
            /* INT16_MIN is NULL_I16: a non-null i16 cannot hold it, so reject
             * rather than silently returning 0Nh.  The string->vector cast
             * (cast_vec_numeric) routes each element through here too, so
             * vectors reject it consistently — string ingest is loud, unlike
             * numeric narrowing which nulls the sentinel (see as.rfl). */
            if ((int16_t)v == NULL_I16)
                return ray_error("domain", "as: cannot parse str as i16, value is the i16 null sentinel");
            return ray_i16((int16_t)v);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_I16);
        return ray_error("type", "as: cannot cast %s to i16", ray_type_name(val->type));
    }
    /* Cast to F64 / f64 */
    if (cast_match(tname, tlen, "F64") || cast_match(tname, tlen, "f64")) {
        ray_release(s);
        if (val->type == -RAY_F64) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return make_f64(val->b8 ? 1.0 : 0.0);
        if (val->type == -RAY_I64) return make_f64((double)val->i64);
        if (val->type == -RAY_I32) return make_f64((double)val->i32);
        if (val->type == -RAY_I16) return make_f64((double)val->i16);
        if (val->type == -RAY_U8)  return make_f64((double)val->u8);
        if (val->type == -RAY_DATE || val->type == -RAY_TIME) return make_f64((double)val->i32);
        if (val->type == -RAY_TIMESTAMP) return make_f64((double)val->i64);
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val);
            if (!sp) return ray_error("domain", "as: cannot parse empty str as f64");
            char* end;
            errno = 0;
            double v = strtod(sp, &end);
            if (end == sp) return ray_error("domain", "as: cannot parse str as f64");
            if (*end != '\0')
                return ray_error("domain", "as: cannot parse str as f64, unexpected trailing characters");
            /* STAGE 2 (ingest/cast STR→F64): canonicalize at the ingest entry
             * point.  strtod("inf")/strtod("1e400")/strtod("nan") would yield a
             * non-finite F64; make_f64 maps every non-finite to NULL_F64 (0Nf),
             * closing the single-null float model on the cast surface.  An atom
             * 0Nf is itself the null — there is no HAS_NULLS attr to set. */
            return make_f64(v);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_F64);
        return ray_error("type", "as: cannot cast %s to f64", ray_type_name(val->type));
    }
    /* Cast to B8/BOOL/b8 */
    if (cast_match(tname, tlen, "BOOL") || cast_match(tname, tlen, "B8") || cast_match(tname, tlen, "b8")) {
        ray_release(s);
        if (val->type == -RAY_BOOL) { ray_retain(val); return val; }
        if (val->type == -RAY_I64) return make_bool(val->i64 != 0 ? 1 : 0);
        if (val->type == -RAY_I32) return make_bool(val->i32 != 0 ? 1 : 0);
        if (val->type == -RAY_I16) return make_bool(val->i16 != 0 ? 1 : 0);
        if (val->type == -RAY_U8) return make_bool(val->u8 != 0 ? 1 : 0);
        if (val->type == -RAY_F64) return make_bool(val->f64 != 0.0 ? 1 : 0);
        if (val->type == -RAY_DATE) return make_bool(val->i32 != 0 ? 1 : 0);
        if (val->type == -RAY_TIME) return make_bool(val->i32 != 0 ? 1 : 0);
        if (val->type == -RAY_TIMESTAMP) return make_bool(val->i64 != 0 ? 1 : 0);
        if (val->type == -RAY_STR) return make_bool(ray_str_len(val) > 0 ? 1 : 0);
        /* Vector cast: b8/B8 */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_BOOL);
        return ray_error("type", "as: cannot cast %s to b8", ray_type_name(val->type));
    }
    /* Cast to STR/str */
    if (cast_match(tname, tlen, "STR") || cast_match(tname, tlen, "str")) {
        ray_release(s);
        if (val->type == -RAY_STR) { ray_retain(val); return val; }
        if (val->type == -RAY_SYM) {
            ray_t* sym_str = ray_sym_str(val->i64);
            return sym_str ? sym_str : ray_str("", 0);
        }
        if (val->type == -RAY_I64) {
            char buf[32]; int n2 = snprintf(buf, sizeof(buf), "%lld", (long long)val->i64);
            return ray_str(buf, (size_t)n2);
        }
        if (val->type == -RAY_I32) {
            char buf[32]; int n2 = snprintf(buf, sizeof(buf), "%d", (int)val->i32);
            return ray_str(buf, (size_t)n2);
        }
        if (val->type == -RAY_I16) {
            char buf[32]; int n2 = snprintf(buf, sizeof(buf), "%d", (int)val->i16);
            return ray_str(buf, (size_t)n2);
        }
        if (val->type == -RAY_F64) {
            double fv = val->f64;
            fv = clear_neg_zero(fv);
            char buf[32]; int n2 = snprintf(buf, sizeof(buf), "%g", fv);
            return ray_str(buf, (size_t)n2);
        }
        if (val->type == -RAY_BOOL) {
            return val->b8 ? ray_str("true", 4) : ray_str("false", 5);
        }
        /* Fallback: use ray_fmt for any other atom type */
        if (ray_is_atom(val)) {
            ray_t* formatted = ray_fmt(val, 0);
            if (formatted && !RAY_IS_ERR(formatted)) return formatted;
            if (formatted) ray_release(formatted);
        }
        /* Vector/list -> STR vector: cast each element to string */
        if (ray_is_vec(val) || val->type == RAY_LIST) {
            int64_t n2 = val->len;
            ray_t* vec = ray_vec_new(RAY_STR, n2);
            if (!vec || RAY_IS_ERR(vec)) return vec ? vec : ray_error("oom", NULL);
            for (int64_t i = 0; i < n2; i++) {
                int alloc = 0;
                ray_t* elem = collection_elem(val, i, &alloc);
                if (RAY_IS_ERR(elem)) { ray_release(vec); return elem; }
                ray_t* cast = ray_cast_fn(type_sym, elem);
                if (alloc) ray_release(elem);
                if (RAY_IS_ERR(cast)) { ray_release(vec); return cast; }
                const char* sp = ray_str_ptr(cast);
                size_t slen = ray_str_len(cast);
                vec = ray_str_vec_append(vec, sp ? sp : "", sp ? slen : 0);
                ray_release(cast);
                if (RAY_IS_ERR(vec)) return vec;
            }
            ray_t* result = cast_vec_copy_nulls(vec, val);
            if (RAY_IS_ERR(result)) return result;
            return vec;
        }
        return ray_error("type", "as: cannot cast %s to str", ray_type_name(val->type));
    }
    /* Cast to SYMBOL/sym */
    if (cast_match(tname, tlen, "SYMBOL") || cast_match(tname, tlen, "sym") || cast_match(tname, tlen, "symbol")) {
        ray_release(s);
        if (val->type == -RAY_SYM) { ray_retain(val); return val; }
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val);
            size_t slen = ray_str_len(val);
            int64_t id = ray_sym_intern(sp, slen);
            return ray_sym(id);
        }
        /* Integer/bool atom -> symbol: convert to plain number string */
        if (ray_is_atom(val) && (is_numeric(val) || val->type == -RAY_BOOL)) {
            char buf[64]; int n2;
            if (val->type == -RAY_BOOL)     n2 = snprintf(buf, sizeof(buf), "%d", (int)val->b8);
            else if (val->type == -RAY_U8)  n2 = snprintf(buf, sizeof(buf), "%u", (unsigned)val->u8);
            else if (val->type == -RAY_I16) n2 = snprintf(buf, sizeof(buf), "%d", (int)val->i16);
            else if (val->type == -RAY_I32) n2 = snprintf(buf, sizeof(buf), "%d", (int)val->i32);
            else if (val->type == -RAY_F64) {
                double fv = val->f64;
                fv = clear_neg_zero(fv);
                n2 = snprintf(buf, sizeof(buf), "%.17g", fv);
            }
            else n2 = snprintf(buf, sizeof(buf), "%lld", (long long)as_i64(val));
            if (n2 > 0) {
                int64_t id = ray_sym_intern(buf, (size_t)n2);
                return ray_sym(id);
            }
        }
        /* Temporal/guid atom -> symbol: use ray_fmt for formatting */
        if (ray_is_atom(val) && (is_temporal(val) || val->type == -RAY_GUID)) {
            ray_t* formatted = ray_fmt(val, 0);
            if (formatted && !RAY_IS_ERR(formatted)) {
                const char* sp = ray_str_ptr(formatted);
                size_t slen = ray_str_len(formatted);
                int64_t id = ray_sym_intern(sp, slen);
                ray_release(formatted);
                return ray_sym(id);
            }
            if (formatted) ray_release(formatted);
        }
        /* Vector cast: SYMBOL vec from other vecs */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_SYM);
        return ray_error("type", "as: cannot cast %s to sym", ray_type_name(val->type));
    }
    /* Cast to DATE/date */
    if (cast_match(tname, tlen, "DATE") || cast_match(tname, tlen, "date")) {
        ray_release(s);
        if (val->type == -RAY_DATE) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return ray_date((int64_t)val->b8);
        if (val->type == -RAY_U8)  return ray_date((int64_t)val->u8);
        if (val->type == -RAY_I16) return ray_date((int64_t)val->i16);
        if (val->type == -RAY_I32) return ray_date((int64_t)val->i32);
        if (val->type == -RAY_I64) return ray_date(val->i64);
        if (val->type == -RAY_F64) return ray_date(ray_cast_f64_to_i32_null(val->f64));
        if (val->type == -RAY_TIME) return ray_date((int64_t)val->i32);
        if (val->type == -RAY_TIMESTAMP) return ray_date(ts_days_floor(val->i64));
        if (val->type == -RAY_STR) {
            /* Parse "YYYY.MM.DD" with a bounded cursor that must consume the
             * whole string.  The old sscanf path accepted trailing garbage
             * ("2024.01.02junk") and, worse, never range-checked the month:
             * a month > 13 drove the days-in-month table index out of bounds
             * — e.g. (as 'DATE "9999.99.99") read md[99] (ASan OOB). Validate
             * the month BEFORE indexing the table, and the day against the
             * actual days in that (possibly leap) month. Mirrors the strict
             * TIMESTAMP string cast. */
            const char* sp = ray_str_ptr(val);
            size_t sl = ray_str_len(val);
            size_t i = 0;
            int y = 0, m = 0, d2 = 0;
            #define DT_DIGITS(w, out) do {                                                 \
                (out) = 0;                                                                 \
                for (int _k = 0; _k < (w); _k++) {                                         \
                    if (i >= sl || sp[i] < '0' || sp[i] > '9')                             \
                        return ray_error("domain", "as: cannot parse str as date, expected YYYY.MM.DD"); \
                    (out) = (out) * 10 + (sp[i] - '0'); i++;                               \
                }                                                                         \
            } while (0)
            DT_DIGITS(4, y);
            if (i >= sl || sp[i] != '.') return ray_error("domain", "as: cannot parse str as date, expected YYYY.MM.DD");
            i++;
            DT_DIGITS(2, m);
            if (i >= sl || sp[i] != '.') return ray_error("domain", "as: cannot parse str as date, expected YYYY.MM.DD");
            i++;
            DT_DIGITS(2, d2);
            if (i != sl) return ray_error("domain", "as: cannot parse str as date, unexpected trailing characters");
            #undef DT_DIGITS
            static const int dt_md[] = {0,31,28,31,30,31,30,31,31,30,31,30,31};
            int leap = (y % 4 == 0 && (y % 100 != 0 || y % 400 == 0));
            if (m < 1 || m > 12)
                return ray_error("domain", "as: cannot parse str as date, month out of range");
            int mdays = dt_md[m] + (m == 2 && leap ? 1 : 0);
            if (d2 < 1 || d2 > mdays)
                return ray_error("domain", "as: cannot parse str as date, day out of range for month");
            int64_t days = 0;
            { int ty;
              for (ty = 2000; ty < y; ty++) days += (ty % 4 == 0 && (ty % 100 != 0 || ty % 400 == 0)) ? 366 : 365;
              for (ty = y; ty < 2000; ty++) days -= (ty % 4 == 0 && (ty % 100 != 0 || ty % 400 == 0)) ? 366 : 365;
            }
            { for (int mi = 1; mi < m; mi++) days += dt_md[mi] + (mi == 2 && leap ? 1 : 0);
              days += d2 - 1;
            }
            return ray_date(days);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_DATE);
        return ray_error("type", "as: cannot cast %s to date", ray_type_name(val->type));
    }
    /* Cast to TIME/time */
    if (cast_match(tname, tlen, "TIME") || cast_match(tname, tlen, "time")) {
        ray_release(s);
        if (val->type == -RAY_TIME) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return ray_time((int64_t)val->b8);
        if (val->type == -RAY_U8)  return ray_time((int64_t)val->u8);
        if (val->type == -RAY_I16) return ray_time((int64_t)val->i16);
        if (val->type == -RAY_I32) return ray_time((int64_t)val->i32);
        if (val->type == -RAY_I64) return ray_time(val->i64);
        if (val->type == -RAY_F64) return ray_time(ray_cast_f64_to_i32_null(val->f64));
        if (val->type == -RAY_DATE) return ray_time((int64_t)val->i32);
        if (val->type == -RAY_TIMESTAMP)
            /* TIMESTAMP is ns since epoch; TIME stores ms-of-day.  Use
             * floor-mod (not C-style truncate-toward-zero %) so pre-
             * 2000 timestamps give time-of-day in [0, 86_400_000) ms,
             * matching wall-clock semantics. */
            return ray_time((int64_t)(ts_ns_in_day(val->i64) / 1000000LL));
        if (val->type == -RAY_STR) {
            /* Parse "[-]HH:MM[:SS][.fff]" with a bounded cursor that must
             * consume the whole string.  TIME is a signed duration — ms-of-day
             * can exceed a day and go negative (see .csv.read round-tripping) —
             * so the hour field is variable width and unbounded, but minutes
             * and seconds are 0-59.  The old sscanf path accepted trailing
             * garbage ("12:34:56junk") and out-of-range fields ("25:99:99" →
             * 26:40:39), silently corrupting the value; mirror the strict DATE
             * / TIMESTAMP string casts. */
            const char* sp = ray_str_ptr(val);
            size_t sl = ray_str_len(val);
            size_t i = 0;
            int64_t sign = 1;
            if (i < sl && sp[i] == '-') { sign = -1; i++; }
            /* Variable-width hour, >= 1 digit, capped so ms cannot overflow. */
            if (i >= sl || sp[i] < '0' || sp[i] > '9')
                return ray_error("domain", "as: cannot parse str as time, expected [-]HH:MM[:SS][.fff]");
            int64_t hh = 0;
            int hd = 0;
            while (i < sl && sp[i] >= '0' && sp[i] <= '9') {
                hh = hh * 10 + (sp[i] - '0'); i++;
                if (++hd > 7) return ray_error("domain", "as: cannot parse str as time, hour field too long");
            }
            /* Read exactly 2 digits into `out`, validate 0..59. */
            #define TM_2DIGIT(out) do {                                                       \
                if (i + 1 >= sl || sp[i] < '0' || sp[i] > '9' || sp[i+1] < '0' || sp[i+1] > '9') \
                    return ray_error("domain", "as: cannot parse str as time, expected 2-digit field"); \
                (out) = (sp[i]-'0')*10 + (sp[i+1]-'0'); i += 2;                              \
                if ((out) > 59) return ray_error("domain", "as: cannot parse str as time, minute/second out of range"); \
            } while (0)
            if (i >= sl || sp[i] != ':')
                return ray_error("domain", "as: cannot parse str as time, expected ':' after hour");
            i++;
            int mm = 0, ss = 0, tms = 0;
            TM_2DIGIT(mm);
            if (i < sl && sp[i] == ':') { i++; TM_2DIGIT(ss); }
            #undef TM_2DIGIT
            /* Optional fractional seconds → milliseconds; a bare trailing '.'
             * is accepted as .000. */
            if (i < sl && sp[i] == '.') {
                i++;
                char mbuf[4] = "000";
                int mi = 0;
                while (i < sl && sp[i] >= '0' && sp[i] <= '9') {
                    if (mi < 3) mbuf[mi++] = sp[i];
                    i++;
                }
                tms = (int)strtol(mbuf, NULL, 10);
            }
            if (i != sl)
                return ray_error("domain", "as: cannot parse str as time, unexpected trailing characters");
            int64_t ms = sign * (hh * 3600000LL + (int64_t)mm * 60000LL + (int64_t)ss * 1000LL + tms);
            if (ms > INT32_MAX || ms < INT32_MIN)
                return ray_error("domain", "as: cannot parse str as time, out of int32 millisecond range");
            return ray_time(ms);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_TIME);
        return ray_error("type", "as: cannot cast %s to time", ray_type_name(val->type));
    }
    /* Cast to TIMESTAMP/timestamp */
    if (cast_match(tname, tlen, "TIMESTAMP") || cast_match(tname, tlen, "timestamp")) {
        ray_release(s);
        if (val->type == -RAY_TIMESTAMP) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return ray_timestamp((int64_t)val->b8);
        if (val->type == -RAY_U8)  return ray_timestamp((int64_t)val->u8);
        if (val->type == -RAY_I16) return ray_timestamp((int64_t)val->i16);
        if (val->type == -RAY_I32) return ray_timestamp((int64_t)val->i32);
        if (val->type == -RAY_I64) return ray_timestamp(val->i64);
        if (val->type == -RAY_F64) return ray_timestamp(ray_cast_f64_to_i64_null(val->f64));
        if (val->type == -RAY_TIME) return ray_timestamp((int64_t)val->i32);
        if (val->type == -RAY_DATE) {
            int64_t days = val->i32;
            return ray_timestamp(days * 24LL * 60 * 60 * 1000000000LL);
        }
        /* ISO string -> timestamp.  Parse the whole value with a bounded
         * cursor and require it to reach the end of the string.  A partial
         * or trailing-garbage match must be REJECTED — otherwise malformed
         * input is silently cast to a valid-but-wrong timestamp, the exact
         * data-corruption class this parser guards against.  Grammar:
         *   YYYY<sep>MM<sep>DD [ (T|' '|D) HH:MM[:SS][.frac] [Z|(+|-)HH[:]?MM] ]
         * where <sep> is '-' or '.' and is consistent within the date. */
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val);
            size_t sl = ray_str_len(val);
            size_t i = 0;
            int y = 0, m = 0, d = 0, hh = 0, mm = 0, ss = 0;
            long long frac = 0, tz_ns = 0;

            /* Read exactly `w` digits (0-9) into `out`, or fail. */
            #define TS_DIGITS(w, out) do {                                                     \
                (out) = 0;                                                                     \
                for (int _k = 0; _k < (w); _k++) {                                             \
                    if (i >= sl || sp[i] < '0' || sp[i] > '9')                                 \
                        return ray_error("domain", "as: cannot parse str as timestamp, expected %d-digit field at offset %lld", (w), (long long)i); \
                    (out) = (out) * 10 + (sp[i] - '0'); i++;                                   \
                }                                                                             \
            } while (0)

            TS_DIGITS(4, y);
            if (i >= sl || (sp[i] != '-' && sp[i] != '.'))
                return ray_error("domain", "as: cannot parse str as timestamp, expected '-' or '.' date separator");
            char dsep = sp[i]; i++;
            TS_DIGITS(2, m);
            if (i >= sl || sp[i] != dsep)
                return ray_error("domain", "as: cannot parse str as timestamp, inconsistent date separator");
            i++;
            TS_DIGITS(2, d);
            if (m < 1 || m > 12 || d < 1)
                return ray_error("domain", "as: cannot parse str as timestamp, date component out of range");
            static const int ts_md[] = {0,31,28,31,30,31,30,31,31,30,31,30,31};
            int leap = (y % 4 == 0 && (y % 100 != 0 || y % 400 == 0));
            int mdays = ts_md[m] + (m == 2 && leap ? 1 : 0);
            if (d > mdays)
                return ray_error("domain", "as: cannot parse str as timestamp, date component out of range");

            if (i < sl) {
                /* date/time separator: T, space, or D (kdb literal form) */
                if (sp[i] != 'T' && sp[i] != ' ' && sp[i] != 'D')
                    return ray_error("domain", "as: cannot parse str as timestamp, expected 'T', ' ', or 'D' between date and time");
                i++;
                TS_DIGITS(2, hh);
                if (i >= sl || sp[i] != ':')
                    return ray_error("domain", "as: cannot parse str as timestamp, expected ':' after hour");
                i++;
                TS_DIGITS(2, mm);
                if (i < sl && sp[i] == ':') { i++; TS_DIGITS(2, ss); }
                if (hh > 23 || mm > 59 || ss > 59)
                    return ray_error("domain", "as: cannot parse str as timestamp, time component out of range");
                /* optional fractional seconds: at least one digit; up to 9
                 * are kept (right-padded), any beyond are consumed. */
                if (i < sl && sp[i] == '.') {
                    i++;
                    if (i >= sl || sp[i] < '0' || sp[i] > '9')
                        return ray_error("domain", "as: cannot parse str as timestamp, expected fractional digits after '.'");
                    char fbuf[10] = "000000000";
                    int fi = 0;
                    while (i < sl && sp[i] >= '0' && sp[i] <= '9') {
                        if (fi < 9) fbuf[fi++] = sp[i];
                        i++;
                    }
                    frac = strtoll(fbuf, NULL, 10);
                }
                /* optional timezone: Z | (+|-)HH[:]?MM */
                if (i < sl) {
                    if (sp[i] == 'Z' || sp[i] == 'z') {
                        i++;
                    } else if (sp[i] == '+' || sp[i] == '-') {
                        int tz_sign = (sp[i] == '+') ? 1 : -1; i++;
                        int tz_hh = 0, tz_mm = 0;
                        TS_DIGITS(2, tz_hh);
                        if (i < sl && sp[i] == ':') i++;   /* optional ':' */
                        TS_DIGITS(2, tz_mm);
                        if (tz_hh > 23 || tz_mm > 59)
                            return ray_error("domain", "as: cannot parse str as timestamp, timezone offset out of range");
                        tz_ns = (long long)tz_sign *
                                ((long long)tz_hh * 3600 + (long long)tz_mm * 60) * 1000000000LL;
                    } else {
                        return ray_error("domain", "as: cannot parse str as timestamp, unexpected trailing characters");
                    }
                }
            }
            if (i != sl)
                return ray_error("domain", "as: cannot parse str as timestamp, unexpected trailing characters");
            #undef TS_DIGITS

            /* Convert to days since 2000-01-01 */
            int64_t days = 0;
            { int ty;
              for (ty = 2000; ty < y; ty++) days += (ty % 4 == 0 && (ty % 100 != 0 || ty % 400 == 0)) ? 366 : 365;
              for (ty = y; ty < 2000; ty++) days -= (ty % 4 == 0 && (ty % 100 != 0 || ty % 400 == 0)) ? 366 : 365;
            }
            {
              for (int mi = 1; mi < m; mi++) days += ts_md[mi] + (mi == 2 && leap ? 1 : 0);
              days += d - 1;
            }
            const int64_t ns_per_day = 86400000000000LL;
            if (days > INT64_MAX / ns_per_day || days < INT64_MIN / ns_per_day)
                return ray_error("domain", "as: timestamp out of int64 nanosecond range");
            int64_t day_ns = days * ns_per_day;
            int64_t tod_ns = (int64_t)hh * 3600000000000LL +
                             (int64_t)mm * 60000000000LL + (int64_t)ss * 1000000000LL + frac;
            if (day_ns > INT64_MAX - tod_ns)
                return ray_error("domain", "as: timestamp out of int64 nanosecond range");
            int64_t ns = day_ns + tod_ns;
            /* Timezone offset converts local wall-clock to UTC. */
            if (tz_ns > 0 && ns < INT64_MIN + tz_ns)
                return ray_error("domain", "as: timestamp out of int64 nanosecond range");
            if (tz_ns < 0 && ns > INT64_MAX + tz_ns)
                return ray_error("domain", "as: timestamp out of int64 nanosecond range");
            ns -= tz_ns;
            return ray_timestamp(ns);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_TIMESTAMP);
        return ray_error("type", "as: cannot cast %s to timestamp", ray_type_name(val->type));
    }
    /* Cast to GUID/guid */
    if (cast_match(tname, tlen, "GUID") || cast_match(tname, tlen, "guid")) {
        ray_release(s);
        if (val->type == -RAY_GUID) { ray_retain(val); return val; }
        if (val->type == -RAY_STR) {
            /* Parse UUID string: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx".
             * Require the canonical hyphenated form exactly: 32 hex
             * nibbles with the four dashes at offsets 8/13/18/23.  The old
             * parser skipped '-' anywhere and decoded any character as a
             * nibble, so non-hex garbage (e.g. all-'z') silently produced a
             * wrong-but-valid GUID. */
            const char* sp = ray_str_ptr(val);
            size_t sl = ray_str_len(val);
            if (sl != 36) return ray_error("domain", "as: cannot parse str as guid, expected 36 chars, got %lld", (long long)sl);
            uint8_t bytes[16];
            int nib = 0;
            for (size_t j = 0; j < sl; j++) {
                char c = sp[j];
                /* Offsets 8/13/18/23 MUST be the dash; every other position
                 * MUST be a hex digit.  This is position-driven (rather than
                 * "reject a misplaced dash") so the 32 non-dash positions are
                 * exactly 32 nibbles — nib never exceeds 31 and bytes[nib>>1]
                 * cannot run past bytes[16]. */
                if (j == 8 || j == 13 || j == 18 || j == 23) {
                    if (c != '-')
                        return ray_error("domain", "as: cannot parse str as guid, expected '-' at offset %lld", (long long)j);
                    continue;
                }
                int v;
                if (c >= '0' && c <= '9')        v = c - '0';
                else if (c >= 'a' && c <= 'f')   v = c - 'a' + 10;
                else if (c >= 'A' && c <= 'F')   v = c - 'A' + 10;
                else return ray_error("domain", "as: cannot parse str as guid, non-hex character '%c'", c);
                if ((nib & 1) == 0) bytes[nib >> 1] = (uint8_t)(v << 4);
                else                bytes[nib >> 1] |= (uint8_t)v;
                nib++;
            }
            return ray_guid(bytes);
        }
        /* Vector of GUIDs: empty vector cast */
        if (ray_is_vec(val) && val->len == 0)
            return ray_vec_new(RAY_GUID, 0);
        /* List of strings -> GUID vector */
        if (val->type == RAY_LIST) {
            int64_t n2 = val->len;
            ray_t* vec = ray_vec_new(RAY_GUID, n2);
            if (RAY_IS_ERR(vec)) return vec;
            vec->len = n2;
            uint8_t* data = (uint8_t*)ray_data(vec);
            ray_t** items = (ray_t**)ray_data(val);
            for (int64_t i = 0; i < n2; i++) {
                ray_t* cast = ray_cast_fn(type_sym, items[i]);
                if (RAY_IS_ERR(cast)) { ray_release(vec); return cast; }
                if (cast->obj) memcpy(data + i * 16, ray_data(cast->obj), 16);
                else memcpy(data + i * 16, ray_data(cast), 16);
                ray_release(cast);
            }
            ray_t* result = cast_vec_copy_nulls(vec, val);
            if (RAY_IS_ERR(result)) return result;
            return vec;
        }
        return ray_error("type", "as: cannot cast %s to guid", ray_type_name(val->type));
    }
    /* Cast to U8/u8 */
    if (cast_match(tname, tlen, "U8") || cast_match(tname, tlen, "u8")) {
        ray_release(s);
        if (val->type == -RAY_U8) { ray_retain(val); return val; }
        if (val->type == -RAY_BOOL) return ray_u8(val->b8 ? 1 : 0);
        if (val->type == -RAY_I16) return ray_u8((uint8_t)val->i16);
        if (val->type == -RAY_I32) return ray_u8((uint8_t)val->i32);
        if (val->type == -RAY_I64) return ray_u8((uint8_t)val->i64);
        if (val->type == -RAY_F64) return ray_u8(ray_cast_f64_to_u8_null(val->f64));
        if (val->type == -RAY_STR) {
            const char* sp = ray_str_ptr(val);
            char* end; errno = 0;
            int64_t v = strtoll(sp, &end, 10);
            if (end == sp) return ray_error("domain", "as: cannot parse str as u8");
            if (*end != '\0')
                return ray_error("domain", "as: cannot parse str as u8, unexpected trailing characters");
            if (errno == ERANGE)
                return ray_error("domain", "as: cannot parse str as u8, value out of int64 range");
            if (v < 0 || v > UINT8_MAX)
                return ray_error("domain", "as: cannot parse str as u8, value out of u8 range");
            return ray_u8((uint8_t)v);
        }
        /* Vector cast */
        if (ray_is_vec(val) || val->type == RAY_LIST)
            return cast_vec_numeric(type_sym, val, RAY_U8);
        return ray_error("type", "as: cannot cast %s to u8", ray_type_name(val->type));
    }
    /* Cast to DICT */
    if (cast_match(tname, tlen, "DICT")) {
        ray_release(s);
        if (val->type == RAY_DICT) { ray_retain(val); return val; }
        /* Table -> Dict */
        if (val->type == RAY_TABLE) {
            int64_t ncols = ray_table_ncols(val);
            ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, ncols);
            if (RAY_IS_ERR(keys)) return keys;
            ray_t* vals = ray_list_new(ncols);
            if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
            for (int64_t c = 0; c < ncols; c++) {
                int64_t col_name = ray_table_col_name(val, c);
                ray_t* col_val = ray_table_get_col_idx(val, c);
                keys = ray_vec_append(keys, &col_name);
                if (RAY_IS_ERR(keys)) { ray_release(vals); return keys; }
                vals = ray_list_append(vals, col_val);
                if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
            }
            return ray_dict_new(keys, vals);
        }
        return ray_error("type", "as: cannot cast %s to dict", ray_type_name(val->type));
    }
    /* Cast to TABLE */
    if (cast_match(tname, tlen, "TABLE")) {
        ray_release(s);
        if (val->type == RAY_TABLE) { ray_retain(val); return val; }
        /* Dict -> Table */
        if (val->type == RAY_DICT) {
            ray_t* dkeys = ray_dict_keys(val);
            ray_t* dvals = ray_dict_vals(val);
            int64_t ncols = dkeys ? dkeys->len : 0;
            if (!dkeys || dkeys->type != RAY_SYM || !dvals || dvals->type != RAY_LIST)
                return ray_error("type", "as: cannot cast dict to table, keys must be sym and values must be list");
            ray_t** col_ptrs = (ray_t**)ray_data(dvals);
            ray_t* tbl = ray_table_new(ncols);
            if (RAY_IS_ERR(tbl)) return tbl;
            for (int64_t c = 0; c < ncols; c++) {
                int64_t col_name = ray_read_sym(ray_data(dkeys), c, RAY_SYM, dkeys->attrs);
                ray_t* col_val = col_ptrs[c];
                ray_retain(col_val);
                tbl = ray_table_add_col(tbl, col_name, col_val);
                ray_release(col_val);
                if (RAY_IS_ERR(tbl)) return tbl;
            }
            return tbl;
        }
        return ray_error("type", "as: cannot cast %s to table", ray_type_name(val->type));
    }
    ray_release(s);
    return ray_error("domain", "as: unknown target type name");
}

/* (type val) — return the type code of a value */
/* ray_type_name moved to internal.h */

ray_t* ray_type_fn(ray_t* val) {
    RAY_ASSERT_VALUE(val);
    if (RAY_IS_NULL(val)) return ray_sym(ray_sym_intern("null", 4));
    const char* name = ray_type_name(val->type);
    int64_t id = ray_sym_intern(name, strlen(name));
    return ray_sym(id);
}

/* Ceiling for the growth of an *unsized* read — a stream with no size to
 * go on.  The hinted path is deliberately not subject to it.
 *
 * The heap's anon watermark is NOT a ceiling: crossing it routes the
 * mapping to a spill file rather than failing ("this never rejects work,
 * it just picks disk over RAM" — mem/heap.c).  Bounding by the watermark
 * itself would therefore trade an OOM kill for filling the spill file,
 * which is why this bounds against the *remaining* headroom instead, and
 * takes a quarter of it so an endless stream errors well before the first
 * spill.  A quarter, and the "unknown RAM stays permissive" convention,
 * match the admission gate in ops/cdfuse.c, which rides the same
 * watermark — so ray_heap_set_anon_watermark still steers this and no new
 * knob is introduced.
 *
 * Returns 0 for "no bound known", which callers treat as permissive. */
int64_t ray_unsized_read_budget(void) {
    int64_t wm = ray_heap_anon_watermark();
    if (wm <= 0) return 0;                       /* RAM unknown → permissive */
    int64_t avail = wm - ray_heap_anon_committed();
    if (avail <= 0) return 0;
    return avail / 4;
}

static ray_t* read_file_bytes(ray_t* path_obj, const char* op) {
    if (path_obj->type != -RAY_STR)
        return ray_error("type", "%s: path must be str, got %s", op, ray_type_name(path_obj->type));
    const char* path = ray_str_ptr(path_obj);
    if (!path || ray_str_len(path_obj) == 0)
        return ray_error("domain", "%s: empty path", op);

    /* Reject a directory up front, as `load` does (see the S_ISDIR check in
     * ray_load_file_fn).  Platforms disagree on what fopen/fread do with
     * one — Linux fails the read with EISDIR, others can succeed and
     * return nothing — and "" for a directory would be exactly the silent
     * wrong answer this change exists to remove.  POSIX-only: <sys/stat.h>
     * is guarded out on Windows and MSVC has no S_ISDIR, and unlike
     * ray_load_file_fn this function is compiled on both paths. */
#if !defined(RAY_OS_WINDOWS)
    struct stat dst;
    if (stat(path, &dst) == 0 && S_ISDIR(dst.st_mode))
        return ray_error("io", "%s \"%s\": is a directory", op, path);
#endif

    FILE* fp = fopen(path, "rb");
    if (!fp) return ray_error("io", NULL);

    /* Read until EOF rather than to the size the file reports (#572).
     * Everything under /proc and /sys reports 0 while yielding data, so
     * sizing the read made those files come back as a well-formed empty
     * value — no error, no short-read signal, and a caller branching on
     * the content silently took the wrong branch.  Seeking also fails
     * outright on a FIFO, and an ordinary file that changed size between
     * the measure and the read became a spurious io error.
     *
     * The reported size is still worth having: where it is positive it is
     * an exact capacity hint, so the common case does one allocation and
     * one read with no growth. */
    long hint = 0;
    if (fseek(fp, 0, SEEK_END) == 0) {
        hint = ftell(fp);
        /* An implausible hint is treated as no hint rather than trusted
         * into an overflow — a directory, for instance, seeks fine and
         * reports LONG_MAX.  (Directories are already rejected by the
         * S_ISDIR check above; this guard is for anything else that
         * reports a size it cannot back.) */
        if (hint < 0 || (int64_t)hint > INT64_MAX / 2 || fseek(fp, 0, SEEK_SET) != 0)
            hint = 0;
    }
    /* Unseekable (pipe, character device) or size-0-but-readable: start at
     * a page and grow.  The extra byte on a hinted read is what lets a
     * single fread distinguish "exactly the hinted size" from "more than
     * the hint", so a file that grew is read whole instead of truncated. */
    int64_t cap = hint > 0 ? (int64_t)hint + 1 : 4096;

    /* No ceiling on the hinted path.  A reported size is a finite claim the
     * filesystem backs, and a large file allocating through the heap and
     * spilling to disk is the behaviour that was already correct — the
     * budget below exists to terminate an *endless* stream, which is not a
     * situation a hint can describe. */
    const int64_t budget = ray_unsized_read_budget();

    ray_t* result = ray_vec_new(RAY_U8, cap);
    if (!result || RAY_IS_ERR(result)) {
        fclose(fp);
        return result ? result : ray_error("oom", NULL);
    }

    int64_t len = 0;
    for (;;) {
        if (len == cap) {
            /* An unsized stream — /dev/zero, a live FIFO — has no natural
             * end.  Sizing from the file used to return "" for one
             * instantly; growing without a bound instead walks the process
             * into OOM or into the spill file.  Check before doubling, so
             * the multiply itself cannot overflow. */
            /* Peak during the grow is the old buffer plus the new one — 3*cap,
             * since the old stays alive across the memcpy — so bound the
             * peak rather than the result.  Checked before the multiply, so
             * the multiply itself cannot overflow. */
            if (cap > INT64_MAX / 3) {
                ray_release(result); fclose(fp);
                return ray_error("io", "%s \"%s\": input too large to read", op, path);
            }
            if (budget > 0 && cap * 3 > budget) {
                ray_release(result); fclose(fp);
                return ray_error("io", "%s \"%s\": unsized input exceeds the readable budget of %lld bytes",
                                 op, path, (long long)budget);
            }
            int64_t ncap = cap * 2;
            ray_t* grown = ray_vec_new(RAY_U8, ncap);
            if (!grown || RAY_IS_ERR(grown)) {
                ray_release(result); fclose(fp);
                return grown ? grown : ray_error("oom", NULL);
            }
            memcpy(ray_data(grown), ray_data(result), (size_t)len);
            ray_release(result);
            result = grown; cap = ncap;
        }
        size_t got = fread((uint8_t*)ray_data(result) + len, 1, (size_t)(cap - len), fp);
        len += (int64_t)got;
        if (got == 0) {
            if (ferror(fp)) { ray_release(result); fclose(fp); return ray_error("io", NULL); }
            break;  /* EOF */
        }
    }

    if (fclose(fp) != 0) { ray_release(result); return ray_error("io", NULL); }
    result->len = len;
    return result;
}

/* (read path) — read a file's contents as a string */
ray_t* ray_read_file_fn(ray_t* path_obj) {
    ray_t* bytes = read_file_bytes(path_obj, "read");
    if (RAY_IS_ERR(bytes)) return bytes;

    ray_t* result = ray_str((const char*)ray_data(bytes), (size_t)bytes->len);
    ray_release(bytes);
    return result;
}

/* (read-bytes path) — read a file's contents as a U8 byte vector */
ray_t* ray_read_bytes_fn(ray_t* path_obj) {
    return read_file_bytes(path_obj, "read-bytes");
}

static bool load_path_is_absolute(const char* p) {
    if (p[0] == '/' || p[0] == '\\') return true;
#if defined(RAY_OS_WINDOWS)
    if (((p[0] >= 'A' && p[0] <= 'Z') || (p[0] >= 'a' && p[0] <= 'z')) && p[1] == ':') return true;
#endif
    return false;
}

/* Where a relative load path is looked for after the working directory:
 * below $RAYFORCE_HOME, q's QHOME fallback (#506).  Fills `alt` with the
 * candidate and returns true when a home is set and the path is relative. */
static bool load_home_candidate(const char* path, char* alt, size_t cap) {
    if (load_path_is_absolute(path)) return false;
    const char* home = getenv("RAYFORCE_HOME");
    if (!home || !*home) return false;
    size_t hl = strlen(home);
    while (hl > 1 && (home[hl - 1] == '/' || home[hl - 1] == '\\')) hl--;
    int n = snprintf(alt, cap, "%.*s/%s", (int)hl, home, path);
    return n > 0 && (size_t)n < cap;
}

/* (load path) — read and evaluate a Rayfall script file via mmap.
 *
 * A relative path is resolved against the working directory first and,
 * when that does not exist and RAYFORCE_HOME is set, below the home;
 * an absolute path is used as given.  Nested loads follow the same rule
 * (relative to the working directory, not to the loading file, as q
 * does).  The path that was actually opened is what (.sys.args)
 * reports as `source` while the file runs. */
ray_t* ray_load_file_fn(ray_t* path_obj) {
    if (path_obj->type != -RAY_STR) return ray_error("type", "load: path must be str, got %s", ray_type_name(path_obj->type));
    const char* path = ray_str_ptr(path_obj);
    if (!path) return ray_error("domain", "load: empty path");
    size_t path_len = ray_str_len(path_obj);
    char alt[4096];

#if defined(RAY_OS_WINDOWS)
    /* Windows: fall back to fread */
    FILE* fp = fopen(path, "r");
    if (!fp && errno == ENOENT && load_home_candidate(path, alt, sizeof(alt))) {
        fp = fopen(alt, "r");
        if (!fp) return ray_error("io", "load \"%s\": %s (also tried \"%s\")", path, strerror(ENOENT), alt);
        path = alt; path_len = strlen(alt);
    }
    if (!fp) return ray_error("io", "load \"%s\": %s", path, strerror(errno));
    fseek(fp, 0, SEEK_END);
    long sz = ftell(fp);
    fseek(fp, 0, SEEK_SET);
    if (sz < 0) { int e = errno; fclose(fp); return ray_error("io", "load \"%s\": cannot determine size: %s", path, strerror(e)); }
    if (sz == 0) { fclose(fp); return ray_i64(0); }
    char* buf = (char*)ray_alloc_raw((size_t)sz + 1);
    if (!buf) { fclose(fp); return ray_error("oom", NULL); }
    size_t rd = fread(buf, 1, (size_t)sz, fp);
    fclose(fp);
    buf[rd] = '\0';

    ray_t* nfo = ray_nfo_create(path, path_len, buf, rd);
    ray_t* parsed = ray_parse_with_nfo(buf, nfo);
    if (RAY_IS_ERR(parsed)) { ray_release(nfo); ray_free_raw(buf); return parsed; }

    ray_t* prev_nfo = ray_eval_get_nfo();
    ray_eval_set_nfo(nfo);
    ray_t* result = ray_eval(parsed);
    ray_eval_set_nfo(prev_nfo);

    ray_release(parsed);
    ray_release(nfo);
    ray_free_raw(buf);
    return result;
#else
    /* Every failure names the file and the OS cause: a bare `io` from a
     * script's load is indistinguishable from an IPC, journal or storage
     * failure and sends diagnosis the wrong way (#505). */
    int fd = open(path, O_RDONLY);
    if (fd < 0 && errno == ENOENT && load_home_candidate(path, alt, sizeof(alt))) {
        fd = open(alt, O_RDONLY);
        if (fd < 0) return ray_error("io", "load \"%s\": %s (also tried \"%s\")", path, strerror(ENOENT), alt);
        path = alt; path_len = strlen(alt);
    }
    if (fd < 0) return ray_error("io", "load \"%s\": %s", path, strerror(errno));
    struct stat st;
    if (fstat(fd, &st) < 0 || st.st_size < 0) {
        int e = errno; close(fd);
        return ray_error("io", "load \"%s\": cannot stat: %s", path, strerror(e));
    }
    if (S_ISDIR(st.st_mode)) { close(fd); return ray_error("io", "load \"%s\": is a directory", path); }
    size_t sz = (size_t)st.st_size;
    if (sz == 0) { close(fd); return ray_i64(0); }
    char* map = (char*)ray_vm_map_fd_ro(fd, sz);
    { int e = errno; close(fd); errno = e; }
    if (!map) return ray_error("io", "load \"%s\": cannot map %zu bytes: %s", path, sz, strerror(errno));
    /* Copy to NUL-terminated buffer -- mmap region may not have a trailing NUL */
    char* buf = (char*)ray_alloc_raw(sz + 1);
    if (!buf) { ray_vm_unmap_file(map, sz); return ray_error("oom", NULL); }
    memcpy(buf, map, sz);
    buf[sz] = '\0';
    ray_vm_unmap_file(map, sz);

    ray_t* nfo = ray_nfo_create(path, path_len, buf, sz);
    ray_t* parsed = ray_parse_with_nfo(buf, nfo);
    if (RAY_IS_ERR(parsed)) { ray_release(nfo); ray_free_raw(buf); return parsed; }

    ray_t* prev_nfo = ray_eval_get_nfo();
    ray_eval_set_nfo(nfo);
    ray_t* result = ray_eval(parsed);
    ray_eval_set_nfo(prev_nfo);

    ray_release(parsed);
    ray_release(nfo);
    ray_free_raw(buf);
    return result;
#endif
}

static ray_t* write_file_data(ray_t* path_obj, const void* data, size_t len,
                              const char* op) {
    if (path_obj->type != -RAY_STR)
        return ray_error("type", "%s: path must be str, got %s", op, ray_type_name(path_obj->type));
    const char* path = ray_str_ptr(path_obj);
    if (!path || ray_str_len(path_obj) == 0)
        return ray_error("domain", "%s: empty path", op);
    if (len > 0 && !data) return ray_error("domain", "%s: invalid content", op);

    FILE* fp = fopen(path, "wb");
    if (!fp) return ray_error("io", NULL);
    size_t written = len > 0 ? fwrite(data, 1, len, fp) : 0;
    int close_rc = fclose(fp);
    if (written != len || close_rc != 0) return ray_error("io", NULL);
    return make_i64(0);
}

/* (write path content) — write a string to a file */
ray_t* ray_write_file_fn(ray_t* path_obj, ray_t* content) {
    if (content->type != -RAY_STR)
        return ray_error("type", "write: content must be str, got %s", ray_type_name(content->type));
    return write_file_data(path_obj, ray_str_ptr(content), ray_str_len(content),
                           "write");
}

/* (write-bytes path content) — write a U8 byte vector to a file */
ray_t* ray_write_bytes_fn(ray_t* path_obj, ray_t* content) {
    if (content->type != RAY_U8)
        return ray_error("type", "write-bytes: content must be U8, got %s", ray_type_name(content->type));
    return write_file_data(path_obj, ray_data(content), (size_t)content->len,
                           "write-bytes");
}

/* ══════════════════════════════════════════
 * Additional builtins (ported from rayforce)
 * ══════════════════════════════════════════ */

/* (enlist a b c ...) -> typed vector from atoms */
ray_t* ray_enlist_fn(ray_t** args, int64_t n) {
    if (n == 0) return ray_vec_new(RAY_I64, 0);
    /* Determine type from first arg */
    int8_t atype = args[0]->type;
    bool homogeneous = true;
    bool has_float = (atype == -RAY_F64);
    bool has_int = (atype == -RAY_I64);
    for (int64_t i = 1; i < n; i++) {
        if (args[i]->type != atype) homogeneous = false;
        if (args[i]->type == -RAY_F64) has_float = true;
        if (args[i]->type == -RAY_I64) has_int = true;
    }
    /* Mixed int/float -> promote to f64 */
    if (!homogeneous && has_float && has_int) {
        ray_t* vec = ray_vec_new(RAY_F64, n);
        if (RAY_IS_ERR(vec)) return vec;
        double* d = (double*)ray_data(vec);
        for (int64_t i = 0; i < n; i++)
            d[i] = (args[i]->type == -RAY_F64) ? args[i]->f64 : (double)args[i]->i64;
        vec->len = n;
        for (int64_t i = 0; i < n; i++) {
            if (RAY_ATOM_IS_NULL(args[i])) {
                /* Dual-encoding contract: the (double)NULL_I64 cast above
                 * produces a large finite value (~-9.22e18), not NaN.  Stamp
                 * the F64 sentinel so the payload matches the bitmap bit. */
                d[i] = NULL_F64;
                ray_vec_set_null(vec, i, true);
            }
        }
        return vec;
    }
    if (homogeneous && atype < 0) {
        int8_t vtype = -atype;
        ray_t* vec = ray_vec_new(vtype, n);
        if (RAY_IS_ERR(vec)) return vec;
        switch (vtype) {
        case RAY_I64: case RAY_TIMESTAMP: {
            int64_t* d = (int64_t*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->i64;
            break;
        }
        case RAY_F64: {
            double* d = (double*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->f64;
            break;
        }
        case RAY_I32: case RAY_DATE: case RAY_TIME: {
            int32_t* d = (int32_t*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->i32;
            break;
        }
        case RAY_I16: {
            int16_t* d = (int16_t*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->i16;
            break;
        }
        case RAY_BOOL: {
            bool* d = (bool*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->b8;
            break;
        }
        case RAY_SYM: {
            int64_t* d = (int64_t*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->i64;
            break;
        }
        case RAY_U8: {
            uint8_t* d = (uint8_t*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) d[i] = args[i]->u8;
            break;
        }
        case RAY_STR: {
            ray_t* svec = ray_vec_new(RAY_STR, n);
            if (RAY_IS_ERR(svec)) { ray_free(vec); return svec; }
            for (int64_t i = 0; i < n; i++) {
                svec = ray_str_vec_append(svec, ray_str_ptr(args[i]), ray_str_len(args[i]));
                if (RAY_IS_ERR(svec)) return svec;
            }
            ray_free(vec);
            return svec;
        }
        case RAY_GUID: {
            uint8_t* d = (uint8_t*)ray_data(vec);
            for (int64_t i = 0; i < n; i++) {
                const uint8_t* gd = args[i]->obj ? (const uint8_t*)ray_data(args[i]->obj) : (const uint8_t*)ray_data(args[i]);
                memcpy(d + i * 16, gd, 16);
            }
            break;
        }
        default: goto as_list;
        }
        vec->len = n;
        for (int64_t i = 0; i < n; i++) {
            if (RAY_ATOM_IS_NULL(args[i]))
                ray_vec_set_null(vec, i, true);
        }
        return vec;
    }
as_list:;
    /* Heterogeneous -> list */
    ray_t* lst = ray_list_new((int32_t)n);
    if (RAY_IS_ERR(lst)) return lst;
    for (int64_t i = 0; i < n; i++) {
        ray_retain(args[i]);
        lst = ray_list_append(lst, args[i]);
        ray_release(args[i]);
        if (RAY_IS_ERR(lst)) return lst;
    }
    return lst;
}

/* (dict keys vals) -> dict.  Wraps two parallel containers as a [keys,
 * vals] block.  When vals is shorter than keys, the tail is filled with
 * typed null I64.  Both inputs are copied (refs retained) — caller keeps
 * ownership of the originals. */
ray_t* ray_dict_fn(ray_t* keys, ray_t* vals) {
    if (!ray_is_vec(keys)) return ray_error("type", "dict: keys must be a vector, got %s", ray_type_name(keys->type));
    int64_t n = keys->len;

    /* Hold a fresh ref to keys so ownership is transferred into the dict. */
    ray_retain(keys);

    /* Materialize vals as RAY_LIST of length n. */
    ray_t* vlist = ray_list_new(n);
    if (RAY_IS_ERR(vlist)) { ray_release(keys); return vlist; }
    for (int64_t i = 0; i < n; i++) {
        ray_t* v;
        int alloc = 0;
        if (vals->type == RAY_LIST) {
            v = (i < vals->len) ? ((ray_t**)ray_data(vals))[i] : NULL;
        } else if (ray_is_vec(vals)) {
            v = collection_elem(vals, i, &alloc);
        } else {
            v = vals;
        }
        if (v && !RAY_IS_ERR(v)) {
            vlist = ray_list_append(vlist, v);
            if (alloc) ray_release(v);
        } else {
            ray_t* null_val = ray_typed_null(-RAY_I64);
            vlist = ray_list_append(vlist, null_val);
            ray_release(null_val);
        }
        if (RAY_IS_ERR(vlist)) { ray_release(keys); return vlist; }
    }
    return ray_dict_new(keys, vlist);
}

/* (nil? x) -> true if x is null.
 *
 * Element-wise over a vector or a boxed list: one BOOL per element, the
 * same answer the query path's OP_ISNULL gives per row, so `nil?` means
 * the same thing inside and outside a query (#485).  Nulls are in-band
 * (sentinels, the empty symbol, the empty string), so ray_vec_is_null
 * decides per element regardless of the HAS_NULLS hint.  A list element
 * that is not an atom is not null.  Atoms, the null object, and any
 * other container (a table, a dict) answer for the object itself. */
ray_t* ray_nil_fn(ray_t* x) {
    RAY_ASSERT_VALUE(x);
    if (RAY_IS_NULL(x)) return ray_bool(true);
    if (ray_is_atom(x)) return ray_bool(RAY_ATOM_IS_NULL(x));
    if (ray_is_vec(x) || x->type == RAY_LIST) {
        int64_t n = ray_len(x);
        ray_t* out = ray_vec_new(RAY_BOOL, n);
        if (!out || RAY_IS_ERR(out)) return out ? out : ray_error("oom", NULL);
        uint8_t* o = (uint8_t*)ray_data(out);
        if (x->type == RAY_LIST) {
            ray_t** e = (ray_t**)ray_data(x);
            for (int64_t i = 0; i < n; i++)
                o[i] = (uint8_t)(!e[i] || RAY_IS_NULL(e[i]) ||
                                 (ray_is_atom(e[i]) && RAY_ATOM_IS_NULL(e[i])));
        } else {
            for (int64_t i = 0; i < n; i++)
                o[i] = (uint8_t)ray_vec_is_null(x, i);
        }
        out->len = n;
        return out;
    }
    return ray_bool(false);
}

/* (where bool-vec) -> indices of true values */

/* Parallel where: 3-phase chunk compaction (parallel per-chunk popcount →
 * tiny serial prefix → parallel per-chunk emit at disjoint offsets). */
#define WHERE_CHUNK_ROWS 65536

typedef struct {
    const uint8_t* data;
    int64_t  n;
    int64_t  chunk_rows;
    int64_t* chunk_counts;   /* popcount, then exclusive-prefix offsets */
    int64_t* out;            /* emit-phase output (NULL during count) */
} where_ctx_t;

/* Dispatched via ray_pool_dispatch_n — each task is ONE chunk ([i,i+1)). */
static void where_count_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) {
    (void)wid;
    where_ctx_t* c = (where_ctx_t*)ctxp;
    for (int64_t ch = start; ch < end; ch++) {
        int64_t lo = ch * c->chunk_rows;
        int64_t hi = lo + c->chunk_rows;
        if (hi > c->n) hi = c->n;
        int64_t cnt = 0;
        for (int64_t i = lo; i < hi; i++) cnt += (c->data[i] != 0);
        c->chunk_counts[ch] = cnt;
    }
}

/* Word-at-a-time zero skip: sparse masks — the common WHERE shape —
 * run at memory speed, falling to per-byte only inside a live word. */
static void where_emit_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) {
    (void)wid;
    where_ctx_t* c = (where_ctx_t*)ctxp;
    for (int64_t ch = start; ch < end; ch++) {
        int64_t lo = ch * c->chunk_rows;
        int64_t hi = lo + c->chunk_rows;
        if (hi > c->n) hi = c->n;
        int64_t j = c->chunk_counts[ch];
        int64_t i = lo;
        int64_t h8 = lo + ((hi - lo) & ~7LL);
        for (; i < h8; i += 8) {
            uint64_t w;
            memcpy(&w, c->data + i, 8);
            if (!w) continue;
            for (int64_t k = 0; k < 8; k++)
                if (c->data[i + k]) c->out[j++] = i + k;
        }
        for (; i < hi; i++)
            if (c->data[i]) c->out[j++] = i;
    }
}

ray_t* ray_where_fn(ray_t* x) {
    if (!ray_is_vec(x) || x->type != RAY_BOOL)
        return ray_error("type", "where: argument must be a b8 vector, got %s", ray_type_name(x->type));
    const uint8_t* data = (const uint8_t*)ray_data(x);
    int64_t n = x->len;

    ray_pool_t* pool = ray_pool_get();
    if (ray_pool_par_dispatch_ok(pool, n, RAY_PARALLEL_THRESHOLD)) {
        /* One task per chunk (ray_pool_dispatch_n).  Chunk count is capped
         * at 1024 — the pool's INITIAL ring capacity — so dispatch_n never
         * needs to grow the ring (its clamp-on-grow-failure silently DROPS
         * tasks, which here would mean uninitialized prefix entries and
         * out-of-bounds emit offsets).  1024 tasks is ample granularity. */
        int64_t chunk_rows = WHERE_CHUNK_ROWS;
        while ((n + chunk_rows - 1) / chunk_rows > RAY_POOL_INIT_TASKS) chunk_rows *= 2;
        int64_t n_chunks = (n + chunk_rows - 1) / chunk_rows;
        /* ops/internal.h's scratch_alloc idiom is unavailable here (that
         * header clashes with builtins.c-local helpers); ray_alloc +
         * ray_release below is its exact expansion. */
        ray_t* cc_hdr = ray_alloc((size_t)n_chunks * sizeof(int64_t));
        int64_t* chunk_counts = cc_hdr ? (int64_t*)ray_data(cc_hdr) : NULL;
        if (chunk_counts) {
            where_ctx_t wc = { .data = data, .n = n, .chunk_rows = chunk_rows,
                               .chunk_counts = chunk_counts, .out = NULL };
            ray_pool_dispatch_n(pool, where_count_fn, &wc, (uint32_t)n_chunks);
            int64_t cum = 0;
            for (int64_t ch = 0; ch < n_chunks; ch++) {
                int64_t v = chunk_counts[ch];
                chunk_counts[ch] = cum;
                cum += v;
            }
            ray_t* result = ray_vec_new(RAY_I64, cum);
            if (RAY_IS_ERR(result)) { ray_release(cc_hdr); return result; }
            wc.out = (int64_t*)ray_data(result);
            ray_pool_dispatch_n(pool, where_emit_fn, &wc, (uint32_t)n_chunks);
            result->len = cum;
            ray_release(cc_hdr);
            return result;
        }
        /* scratch OOM → sequential fallback below */
    }

    /* Count pass: `cnt += byte != 0` vectorizes; the branchy form costs
     * a mispredict-prone compare per element. */
    int64_t cnt = 0;
    for (int64_t i = 0; i < n; i++) cnt += (data[i] != 0);
    ray_t* result = ray_vec_new(RAY_I64, cnt);
    if (RAY_IS_ERR(result)) return result;
    int64_t* out = (int64_t*)ray_data(result);
    int64_t j = 0;
    /* Emit pass: skip zero 8-byte lanes word-at-a-time (sparse masks —
     * the common WHERE shape — run at memory speed), fall to per-byte
     * only inside a live word. */
    int64_t i = 0;
    int64_t n8 = n & ~7LL;
    for (; i < n8; i += 8) {
        uint64_t w;
        memcpy(&w, data + i, 8);
        if (!w) continue;
        for (int64_t k = 0; k < 8; k++)
            if (data[i + k]) out[j++] = i + k;
    }
    for (; i < n; i++)
        if (data[i]) out[j++] = i;
    result->len = cnt;
    return result;
}

/* (group vec) -> dict mapping each unique value to its indices */
/* ---------------------------------------------------------------------------
 * Open-address hash set for ray_group_indices_fn's scalar / GUID fast paths.
 *
 * Each slot holds either GHT_EMPTY or an already-allocated group index.
 * Lookups compare keys by calling back into the caller with the stored
 * group index — the caller already knows whether the key shape is a
 * plain int64 (scalar) or 16 bytes of guid material in the source
 * column.  Load factor is capped at 0.5; grow on overflow.
 *
 * The table is ref-counted via ray_alloc so the main bookkeeping code
 * can free it in one place on every exit path.
 * ------------------------------------------------------------------------- */

#define GHT_EMPTY 0xFFFFFFFFu

typedef struct group_ht_t {
    ray_t*     block;   /* backing ray_alloc block */
    uint32_t*  slots;   /* cap entries */
    uint32_t   cap;     /* power of 2 */
    uint32_t   mask;    /* cap - 1 */
    uint32_t   count;   /* live entries */
} group_ht_t;

static bool group_ht_init(group_ht_t* h, uint32_t initial_cap) {
    uint32_t cap = 16;
    while (cap < initial_cap) cap *= 2;
    h->block = ray_alloc((size_t)cap * sizeof(uint32_t));
    if (!h->block || RAY_IS_ERR(h->block)) { h->block = NULL; return false; }
    h->slots = (uint32_t*)ray_data(h->block);
    h->cap   = cap;
    h->mask  = cap - 1;
    h->count = 0;
    for (uint32_t i = 0; i < cap; i++) h->slots[i] = GHT_EMPTY;
    return true;
}

static void group_ht_free(group_ht_t* h) {
    if (h->block) ray_free(h->block);
    h->block = NULL;
    h->slots = NULL;
    h->cap = h->mask = h->count = 0;
}

/* Rehash callback: given the stored group index, return the hash for
 * it.  This lets us grow without recomputing raw keys — caller knows
 * how to translate gi back to a key. */
typedef uint64_t (*group_ht_gi_hash_fn)(uint32_t gi, void* ctx);

static bool group_ht_grow(group_ht_t* h, group_ht_gi_hash_fn hash_gi, void* ctx) {
    uint32_t new_cap = h->cap * 2;
    if (new_cap < h->cap) return false;  /* overflow */
    ray_t* new_block = ray_alloc((size_t)new_cap * sizeof(uint32_t));
    if (!new_block || RAY_IS_ERR(new_block)) return false;
    uint32_t* new_slots = (uint32_t*)ray_data(new_block);
    uint32_t new_mask = new_cap - 1;
    for (uint32_t i = 0; i < new_cap; i++) new_slots[i] = GHT_EMPTY;
    for (uint32_t i = 0; i < h->cap; i++) {
        uint32_t gi = h->slots[i];
        if (gi == GHT_EMPTY) continue;
        uint64_t hh = hash_gi(gi, ctx);
        uint32_t slot = (uint32_t)(hh & new_mask);
        while (new_slots[slot] != GHT_EMPTY) slot = (slot + 1) & new_mask;
        new_slots[slot] = gi;
    }
    ray_free(h->block);
    h->block = new_block;
    h->slots = new_slots;
    h->cap   = new_cap;
    h->mask  = new_mask;
    return true;
}

static inline uint64_t mix64(uint64_t h) {
    /* Murmur3 fmix64 */
    h ^= h >> 33; h *= 0xFF51AFD7ED558CCDULL;
    h ^= h >> 33; h *= 0xC4CEB9FE1A85EC53ULL;
    h ^= h >> 33;
    return h;
}

static inline uint64_t hash_guid(const uint8_t* g) {
    uint64_t a, b;
    memcpy(&a, g,     8);
    memcpy(&b, g + 8, 8);
    return mix64(a ^ (b * 0x9E3779B97F4A7C15ULL));
}

static inline uint64_t hash_i64(int64_t v) {
    return mix64((uint64_t)v);
}

/* Hash a generic atom or list, mirroring atom_eq's structural compare.
 * Used by the ray_group_indices_fn LIST path to replace the historical O(N²)
 * linear scan with an open-addressed hash table.  Cross-type numeric
 * coercion goes through f64 so an I64 atom and an F64 atom holding the
 * same value collide (matches atom_eq's `is_numeric → as_f64`).
 *
 * Mirrors the type dispatch in collection.c::hs_hash_row's RAY_LIST
 * branch but recurses into nested lists — hs_hash_row's default tag-
 * only fallback collapses every nested-list row to the same hash, so
 * the existing path is degenerate for composite multi-key composites.
 * Uses the canonical wyhash helpers from ops/hash.h, same as the
 * pivot / datalog / join hashers. */
uint64_t ray_atom_hash(ray_t* a) {
    /* List-element position: elements may be a bare C NULL (ray_list_set stores
     * NULL unretained; ray_list_get is out-of-range), mirroring atom_eq's
     * C-NULL-tolerant element handling — so keep the !a guard, not RAY_ASSERT_VALUE. */
    if (!a || RAY_ATOM_IS_NULL(a)) return 0;
    if (is_numeric(a)) return ray_hash_f64(as_f64(a));
    switch (a->type) {
        case -RAY_SYM:       return ray_hash_i64(a->i64);
        case -RAY_DATE:
        case -RAY_TIME:      return ray_hash_i64((int64_t)a->i32);
        case -RAY_TIMESTAMP: return ray_hash_i64(a->i64);
        case -RAY_GUID: {
            const uint8_t* g = a->obj
                ? (const uint8_t*)ray_data(a->obj)
                : (const uint8_t*)ray_data((ray_t*)a);
            return ray_hash_bytes(g, 16);
        }
        case -RAY_STR:
            return ray_hash_bytes(ray_str_ptr(a), ray_str_len(a));
        case RAY_LIST: {
            int64_t n = a->len;
            ray_t** elems = (ray_t**)ray_data(a);
            /* Seed with len so [] and a list of zeros differ. */
            uint64_t h = ray_hash_i64(n);
            for (int64_t i = 0; i < n; i++)
                h = ray_hash_combine(h, ray_atom_hash(elems[i]));
            return h;
        }
        default:
            /* Vec or unknown atom kind: fold type tag and length.  Two
             * structurally-equal lists never reach here (RAY_LIST branch
             * above) so we can't accidentally produce different hashes
             * for atom_eq-equal pairs. */
            return ray_hash_i64(((int64_t)a->type << 32) ^ (int64_t)a->len);
    }
}

/* Context for GUID rehash: the 16-byte source base and, indirectly,
 * gvals — which stores the row_idx of the first occurrence per group. */
typedef struct {
    const uint8_t* base;
    const int64_t* gvals;
} ght_guid_ctx_t;

static uint64_t ght_guid_hash_gi(uint32_t gi, void* ctx) {
    ght_guid_ctx_t* c = (ght_guid_ctx_t*)ctx;
    return hash_guid(c->base + c->gvals[gi] * 16);
}

typedef struct { const int64_t* gvals; } ght_i64_ctx_t;
static uint64_t ght_i64_hash_gi(uint32_t gi, void* ctx) {
    ght_i64_ctx_t* c = (ght_i64_ctx_t*)ctx;
    return hash_i64(c->gvals[gi]);
}

typedef struct {
    const ray_str_t* desc;
    const char* pool;
    const int64_t* gvals;
} ght_str_ctx_t;

static uint64_t ght_str_hash_gi(uint32_t gi, void* ctx) {
    ght_str_ctx_t* c = (ght_str_ctx_t*)ctx;
    return ray_str_t_hash32(&c->desc[c->gvals[gi]], c->pool);
}

/* Context for the LIST-path rehash: gkeys holds atom pointers for each
 * unique group (one slot per gi), recomputed on grow via atom_hash. */
typedef struct { ray_t** gkeys; } ght_list_ctx_t;
static uint64_t ght_list_hash_gi(uint32_t gi, void* ctx) {
    ght_list_ctx_t* c = (ght_list_ctx_t*)ctx;
    return ray_atom_hash(c->gkeys[gi]);
}

/* Grow the per-group bookkeeping arrays used by ray_group_indices_fn.
 * Doubles capacity; copies existing entries; returns false on OOM.
 * Caller is responsible for cleaning up and returning an error if this fails. */
static bool group_grow(ray_t** val_block, ray_t** ivblock,
                       int64_t** gvals, ray_t*** idx_vecs,
                       int64_t cur_count, int64_t* max_groups) {
    int64_t new_max = *max_groups * 2;
    if (new_max <= *max_groups) return false;  /* overflow */
    ray_t* new_val = ray_alloc((size_t)new_max * sizeof(int64_t));
    if (!new_val || RAY_IS_ERR(new_val)) return false;
    ray_t* new_iv = ray_alloc((size_t)new_max * sizeof(ray_t*));
    if (!new_iv || RAY_IS_ERR(new_iv)) { ray_free(new_val); return false; }
    memcpy(ray_data(new_val), *gvals, (size_t)cur_count * sizeof(int64_t));
    memcpy(ray_data(new_iv), *idx_vecs, (size_t)cur_count * sizeof(ray_t*));
    ray_free(*val_block);
    ray_free(*ivblock);
    *val_block = new_val;
    *ivblock = new_iv;
    *gvals = (int64_t*)ray_data(new_val);
    *idx_vecs = (ray_t**)ray_data(new_iv);
    *max_groups = new_max;
    return true;
}

/* Same as group_grow but also resizes the LIST-path's keys block
 * (gkeys — ray_t* atom pointers, one per unique group).  Multi-key
 * non-agg select-by lands here via composite-key LISTs and can
 * exceed the initial 1024-slot cap on real workloads. */
static bool group_grow_listkeys(ray_t** val_block, ray_t** ivblock, ray_t** kblock,
                                int64_t** gvals, ray_t*** idx_vecs, ray_t*** gkeys,
                                int64_t cur_count, int64_t* max_groups) {
    int64_t new_max = *max_groups * 2;
    if (new_max <= *max_groups) return false;
    ray_t* new_val = ray_alloc((size_t)new_max * sizeof(int64_t));
    if (!new_val || RAY_IS_ERR(new_val)) return false;
    ray_t* new_iv = ray_alloc((size_t)new_max * sizeof(ray_t*));
    if (!new_iv || RAY_IS_ERR(new_iv)) { ray_free(new_val); return false; }
    ray_t* new_k = ray_alloc((size_t)new_max * sizeof(ray_t*));
    if (!new_k || RAY_IS_ERR(new_k)) { ray_free(new_val); ray_free(new_iv); return false; }
    memcpy(ray_data(new_val), *gvals,    (size_t)cur_count * sizeof(int64_t));
    memcpy(ray_data(new_iv),  *idx_vecs, (size_t)cur_count * sizeof(ray_t*));
    memcpy(ray_data(new_k),   *gkeys,    (size_t)cur_count * sizeof(ray_t*));
    ray_free(*val_block);
    ray_free(*ivblock);
    ray_free(*kblock);
    *val_block = new_val;
    *ivblock   = new_iv;
    *kblock    = new_k;
    *gvals     = (int64_t*)ray_data(new_val);
    *idx_vecs  = (ray_t**)ray_data(new_iv);
    *gkeys     = (ray_t**)ray_data(new_k);
    *max_groups = new_max;
    return true;
}

ray_t* ray_group_indices_fn(ray_t* x) {
    if (!ray_is_vec(x) && x->type != RAY_LIST)
        return ray_error("type", "group: argument must be a vector or list, got %s", ray_type_name(x->type));
    int64_t n = x->len;
    ray_t* parallel = agg_group_indices(x);
    if (parallel) return parallel;
    if (n == 0) {
        ray_t* keys = ray_list_new(0);
        if (RAY_IS_ERR(keys)) return keys;
        ray_t* vals = ray_list_new(0);
        if (RAY_IS_ERR(vals)) { ray_release(keys); return vals; }
        return ray_dict_new(keys, vals);
    }

    /* Collect unique values.  Every path grows these arrays on demand;
     * starting at 1024 keeps the initial allocation cheap. */
    int64_t max_groups = n < 1024 ? n : 1024;
    ray_t* val_block = ray_alloc((size_t)(max_groups * sizeof(int64_t)));
    if (RAY_IS_ERR(val_block)) return val_block;
    int64_t* gvals = (int64_t*)ray_data(val_block);

    /* For each group, store indices in a separate i64 vector */
    ray_t** idx_vecs = NULL;
    ray_t* ivblock = ray_alloc((size_t)(max_groups * sizeof(ray_t*)));
    if (RAY_IS_ERR(ivblock)) { ray_free(val_block); return ivblock; }
    idx_vecs = (ray_t**)ray_data(ivblock);
    int64_t ngroups = 0;

    /* For LIST type, use atom_eq-based grouping with stored keys.
     * Open-address hash table on atom_hash replaces the historical
     * O(N²) linear scan over gkeys — multi-key non-agg select-by on
     * Large tables with many unique keys now run in linear time. */
    if (x->type == RAY_LIST) {
        ray_t** elems = (ray_t**)ray_data(x);
        ray_t* kblock = ray_alloc((size_t)(max_groups * sizeof(ray_t*)));
        if (RAY_IS_ERR(kblock)) { ray_free(val_block); ray_free(ivblock); return kblock; }
        ray_t** gkeys = (ray_t**)ray_data(kblock);

        group_ht_t ht;
        uint32_t seed_cap = (uint32_t)(n < 64 ? 64 : (n < 1048576 ? (n * 2) : 2097152));
        if (!group_ht_init(&ht, seed_cap)) {
            ray_free(val_block); ray_free(ivblock); ray_free(kblock);
            return ray_error("oom", NULL);
        }
        ght_list_ctx_t lctx = { .gkeys = gkeys };

        for (int64_t i = 0; i < n; i++) {
            ray_t* elem = elems[i];
            uint64_t h = ray_atom_hash(elem);
            uint32_t slot = (uint32_t)(h & ht.mask);
            uint32_t gi_found = GHT_EMPTY;
            while (ht.slots[slot] != GHT_EMPTY) {
                uint32_t gi_p = ht.slots[slot];
                if (atom_eq(gkeys[gi_p], elem)) { gi_found = gi_p; break; }
                slot = (slot + 1) & ht.mask;
            }
            int64_t gi;
            if (gi_found != GHT_EMPTY) {
                gi = gi_found;
            } else {
                if (ngroups >= max_groups) {
                    if (!group_grow_listkeys(&val_block, &ivblock, &kblock,
                                             &gvals, &idx_vecs, &gkeys,
                                             ngroups, &max_groups)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock); ray_free(kblock);
                        return ray_error("oom", NULL);
                    }
                    lctx.gkeys = gkeys;
                }
                gi = ngroups++;
                gkeys[gi] = elem;
                idx_vecs[gi] = ray_vec_new(RAY_I64, 0);
                ht.slots[slot] = (uint32_t)gi;
                ht.count++;
                if (ht.count * 2 > ht.cap) {
                    if (!group_ht_grow(&ht, ght_list_hash_gi, &lctx)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock); ray_free(kblock);
                        return ray_error("oom", NULL);
                    }
                }
            }
            idx_vecs[gi] = ray_vec_append(idx_vecs[gi], &i);
        }
        group_ht_free(&ht);
        /* Build dict: keys as RAY_LIST (heterogeneous atoms), vals as
         * RAY_LIST of I64 idx vectors. */
        ray_t* keys_lst = ray_list_new(ngroups);
        if (RAY_IS_ERR(keys_lst)) { ray_free(kblock); goto gfail; }
        ray_t* vals_lst = ray_list_new(ngroups);
        if (RAY_IS_ERR(vals_lst)) { ray_release(keys_lst); ray_free(kblock); goto gfail; }
        for (int64_t g = 0; g < ngroups; g++) {
            keys_lst = ray_list_append(keys_lst, gkeys[g]);
            if (RAY_IS_ERR(keys_lst)) { ray_release(vals_lst); ray_free(kblock); goto gfail; }
            vals_lst = ray_list_append(vals_lst, idx_vecs[g]);
            ray_release(idx_vecs[g]);
            idx_vecs[g] = NULL;
            if (RAY_IS_ERR(vals_lst)) { ray_release(keys_lst); ray_free(kblock); goto gfail; }
        }
        ray_free(val_block); ray_free(ivblock); ray_free(kblock);
        return ray_dict_new(keys_lst, vals_lst);
    }

    /* RAY_GUID: 16-byte fixed-width grouping via open-address hash set
     * keyed on the guid bytes.  Previously this was an O(N²) linear
     * scan against every existing group, which made (group guid_col)
     * and (select ... by: OrderId) on a 10M row table effectively
     * infinite. */
    if (x->type == RAY_GUID) {
        const uint8_t* base = (const uint8_t*)ray_data(x);
        group_ht_t ht;
        uint32_t seed_cap = (uint32_t)(n < 64 ? 64 : (n < 1048576 ? (n * 2) : 2097152));
        if (!group_ht_init(&ht, seed_cap)) {
            ray_free(val_block); ray_free(ivblock);
            return ray_error("oom", NULL);
        }
        ght_guid_ctx_t gctx = { .base = base, .gvals = gvals };
        ray_progress_update("group", "guid-scan", 0, (uint64_t)n);
        for (int64_t i = 0; i < n; i++) {
            if (((i) & 65535) == 0) {
                if (ray_interrupted()) {
                    for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                    group_ht_free(&ht);
                    ray_free(val_block); ray_free(ivblock);
                    return ray_error("cancel", "interrupted");
                }
                ray_progress_update(NULL, NULL, (uint64_t)i, (uint64_t)n);
            }
            const uint8_t* cur = base + i * 16;
            uint64_t h = hash_guid(cur);
            uint32_t slot = (uint32_t)(h & ht.mask);
            uint32_t gi_found = GHT_EMPTY;
            while (ht.slots[slot] != GHT_EMPTY) {
                uint32_t gi = ht.slots[slot];
                if (memcmp(base + gvals[gi] * 16, cur, 16) == 0) {
                    gi_found = gi;
                    break;
                }
                slot = (slot + 1) & ht.mask;
            }
            int64_t gi;
            if (gi_found != GHT_EMPTY) {
                gi = gi_found;
            } else {
                if (ngroups >= max_groups) {
                    if (!group_grow(&val_block, &ivblock, &gvals, &idx_vecs,
                                    ngroups, &max_groups)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock);
                        return ray_error("oom", NULL);
                    }
                    gctx.gvals = gvals;
                }
                gi = ngroups++;
                gvals[gi] = i;  /* store row index of first occurrence */
                idx_vecs[gi] = ray_vec_new(RAY_I64, 0);
                ht.slots[slot] = (uint32_t)gi;
                ht.count++;
                /* Grow at load factor 0.5 */
                if (ht.count * 2 > ht.cap) {
                    if (!group_ht_grow(&ht, ght_guid_hash_gi, &gctx)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock);
                        return ray_error("oom", NULL);
                    }
                }
            }
            idx_vecs[gi] = ray_vec_append(idx_vecs[gi], &i);
        }
        group_ht_free(&ht);
        /* Keys: dense GUID vector built from collected gvals; vals: LIST of idx vecs. */
        ray_t* keys_vec = ray_vec_new(RAY_GUID, ngroups);
        if (RAY_IS_ERR(keys_vec)) goto gfail;
        for (int64_t g = 0; g < ngroups; g++)
            keys_vec = ray_vec_append(keys_vec, base + gvals[g] * 16);
        if (RAY_IS_ERR(keys_vec)) goto gfail;
        ray_t* vals_lst = ray_list_new(ngroups);
        if (RAY_IS_ERR(vals_lst)) { ray_release(keys_vec); goto gfail; }
        for (int64_t g = 0; g < ngroups; g++) {
            vals_lst = ray_list_append(vals_lst, idx_vecs[g]);
            ray_release(idx_vecs[g]);
            idx_vecs[g] = NULL;
            if (RAY_IS_ERR(vals_lst)) { ray_release(keys_vec); goto gfail; }
        }
        ray_free(val_block); ray_free(ivblock);
        return ray_dict_new(keys_vec, vals_lst);
    }

    /* RAY_STR: descriptor hashing plus collision-safe content equality.
     * Pooled descriptors reuse their cached hash, so grouping digest keys
     * does not reread 32/64-byte payloads merely to locate the hash slot. */
    if (x->type == RAY_STR) {
        ray_t* owner = x;
        int64_t off = 0;
        if (x->attrs & RAY_ATTR_SLICE) {
            owner = x->slice_parent;
            off = x->slice_offset;
        }
        const ray_str_t* desc = (const ray_str_t*)ray_data(owner) + off;
        const char* pool = (owner->str_pool && !RAY_IS_ERR(owner->str_pool))
                         ? (const char*)ray_data(owner->str_pool) : NULL;

        group_ht_t ht;
        uint32_t seed_cap = (uint32_t)(n < 64 ? 64 : (n < 1048576 ? (n * 2) : 2097152));
        if (!group_ht_init(&ht, seed_cap)) {
            ray_free(val_block); ray_free(ivblock);
            return ray_error("oom", NULL);
        }
        ght_str_ctx_t sctx = { .desc = desc, .pool = pool, .gvals = gvals };

        for (int64_t i = 0; i < n; i++) {
            const ray_str_t* cur = &desc[i];
            uint64_t h = ray_str_t_hash32(cur, pool);
            uint32_t slot = (uint32_t)(h & ht.mask);
            uint32_t gi_found = GHT_EMPTY;
            while (ht.slots[slot] != GHT_EMPTY) {
                uint32_t gi = ht.slots[slot];
                if (ray_str_t_eq(&desc[gvals[gi]], pool, cur, pool)) {
                    gi_found = gi;
                    break;
                }
                slot = (slot + 1) & ht.mask;
            }

            int64_t gi;
            if (gi_found != GHT_EMPTY) {
                gi = gi_found;
            } else {
                if (ngroups >= max_groups) {
                    if (!group_grow(&val_block, &ivblock, &gvals, &idx_vecs,
                                    ngroups, &max_groups)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock);
                        return ray_error("oom", NULL);
                    }
                    sctx.gvals = gvals;
                }
                gi = ngroups++;
                gvals[gi] = i;
                idx_vecs[gi] = ray_vec_new(RAY_I64, 0);
                ht.slots[slot] = (uint32_t)gi;
                ht.count++;
                if (ht.count * 2 > ht.cap) {
                    if (!group_ht_grow(&ht, ght_str_hash_gi, &sctx)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock);
                        return ray_error("oom", NULL);
                    }
                }
            }
            idx_vecs[gi] = ray_vec_append(idx_vecs[gi], &i);
        }
        group_ht_free(&ht);

        /* Build keys from each group's first source row, preserving encounter order. */
        ray_t* keys_vec = ray_vec_new(RAY_STR, ngroups);
        if (RAY_IS_ERR(keys_vec)) {
            for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
            ray_free(val_block); ray_free(ivblock);
            return ray_error("oom", NULL);
        }
        for (int64_t g = 0; g < ngroups; g++) {
            const ray_str_t* key = &desc[gvals[g]];
            keys_vec = ray_str_vec_append(keys_vec,
                                          ray_str_t_ptr(key, pool), key->len);
        }
        ray_t* vals_lst = ray_list_new(ngroups);
        if (RAY_IS_ERR(vals_lst)) {
            ray_release(keys_vec); goto gfail;
        }
        for (int64_t g = 0; g < ngroups; g++) {
            vals_lst = ray_list_append(vals_lst, idx_vecs[g]);
            ray_release(idx_vecs[g]);
            idx_vecs[g] = NULL;
            if (RAY_IS_ERR(vals_lst)) { ray_release(keys_vec); goto gfail; }
        }
        ray_free(val_block); ray_free(ivblock);
        return ray_dict_new(keys_vec, vals_lst);
    }

    /* Scalar fast path: every primitive-typed vector packs its group
     * key into an int64 (sym id, raw integer, date/time/timestamp, bool).
     * Use an open-address hash set so high-cardinality group-by stays
     * linear in n rather than the historical O(N²) per-row linear scan. */
    group_ht_t ht;
    uint32_t seed_cap = (uint32_t)(n < 64 ? 64 : (n < 1048576 ? (n * 2) : 2097152));
    if (!group_ht_init(&ht, seed_cap)) {
        ray_free(val_block); ray_free(ivblock);
        return ray_error("oom", NULL);
    }
    ght_i64_ctx_t sctx = { .gvals = gvals };
    /* Null routing: null inputs share the same storage value as a legitimate
     * zero/sentinel (e.g. NULL_I64's atom stores i64=0, NULL_I32 stores
     * i32=0).  Without a separate null bucket the hash table would conflate
     * `0Nl` with a real `0`, silently merging two semantically distinct
     * groups.  Track a single `null_gi` and route every null row there;
     * non-null rows continue to use the value-keyed hash table. */
    int64_t null_gi = -1;
    for (int64_t i = 0; i < n; i++) {
        if (ray_vec_is_null(x, i)) {
            if (null_gi < 0) {
                if (ngroups >= max_groups) {
                    if (!group_grow(&val_block, &ivblock, &gvals, &idx_vecs,
                                    ngroups, &max_groups)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock);
                        return ray_error("oom", NULL);
                    }
                    sctx.gvals = gvals;
                }
                null_gi = ngroups++;
                gvals[null_gi] = 0;          /* placeholder; key value set later */
                idx_vecs[null_gi] = ray_vec_new(RAY_I64, 0);
            }
            idx_vecs[null_gi] = ray_vec_append(idx_vecs[null_gi], &i);
            continue;
        }
        int64_t v;
        /* SYM columns have an adaptive dictionary-index width (W8/W16/W32/W64
         * in attrs); read through ray_read_sym rather than assuming W64, or a
         * narrow-width column (e.g. a few distinct exchanges => W8) is read at
         * an 8-byte stride and walks off the end of the buffer. */
        if (x->type == RAY_SYM)
            v = ray_read_sym(ray_data(x), i, x->type, x->attrs);
        else if (x->type == RAY_I64 || x->type == RAY_TIMESTAMP)
            v = ((int64_t*)ray_data(x))[i];
        else if (x->type == RAY_I32 || x->type == RAY_DATE || x->type == RAY_TIME)
            v = ((int32_t*)ray_data(x))[i];
        else if (x->type == RAY_I16)
            v = ((int16_t*)ray_data(x))[i];
        else if (x->type == RAY_BOOL || x->type == RAY_U8)
            v = ((uint8_t*)ray_data(x))[i];
        else if (x->type == RAY_F64 || x->type == RAY_F32) {
            /* Hash by IEEE-754 bit pattern, not row index — the previous
             * `v = i` fallback put every float row in its own group and
             * the keys_vec build path then reinterpreted those row
             * indices as floats.  Two adjustments keep the bit-pattern
             * approach consistent with atom_eq's IEEE semantics
             * (`a->f64 == b->f64`):
             *   - +0.0 and -0.0 hash equal: canonicalise -0.0 to 0.0.
             *   - Each NaN is its own group (NaN != NaN under IEEE).
             *     Route NaN rows through the dedicated nan-group path
             *     below so the hash table never matches them. */
            double f = (x->type == RAY_F64)
                ? ((double*)ray_data(x))[i]
                : (double)((float*)ray_data(x))[i];
            if (f != f) {
                /* NaN — own bucket per row, just like the null routing. */
                if (ngroups >= max_groups) {
                    if (!group_grow(&val_block, &ivblock, &gvals, &idx_vecs,
                                    ngroups, &max_groups)) {
                        for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                        group_ht_free(&ht);
                        ray_free(val_block); ray_free(ivblock);
                        return ray_error("oom", NULL);
                    }
                    sctx.gvals = gvals;
                }
                int64_t gi_nan = ngroups++;
                memcpy(&gvals[gi_nan], &f, sizeof(f));
                idx_vecs[gi_nan] = ray_vec_new(RAY_I64, 0);
                idx_vecs[gi_nan] = ray_vec_append(idx_vecs[gi_nan], &i);
                continue;
            }
            f = clear_neg_zero(f);
            memcpy(&v, &f, sizeof(v));
        } else
            v = i;

        uint64_t h = hash_i64(v);
        uint32_t slot = (uint32_t)(h & ht.mask);
        uint32_t gi_found = GHT_EMPTY;
        while (ht.slots[slot] != GHT_EMPTY) {
            uint32_t gi = ht.slots[slot];
            if (gvals[gi] == v) { gi_found = gi; break; }
            slot = (slot + 1) & ht.mask;
        }
        int64_t gi;
        if (gi_found != GHT_EMPTY) {
            gi = gi_found;
        } else {
            if (ngroups >= max_groups) {
                if (!group_grow(&val_block, &ivblock, &gvals, &idx_vecs,
                                ngroups, &max_groups)) {
                    for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                    group_ht_free(&ht);
                    ray_free(val_block); ray_free(ivblock);
                    return ray_error("oom", NULL);
                }
                sctx.gvals = gvals;
            }
            gi = ngroups++;
            gvals[gi] = v;
            idx_vecs[gi] = ray_vec_new(RAY_I64, 0);
            ht.slots[slot] = (uint32_t)gi;
            ht.count++;
            if (ht.count * 2 > ht.cap) {
                if (!group_ht_grow(&ht, ght_i64_hash_gi, &sctx)) {
                    for (int64_t g = 0; g < ngroups; g++) ray_release(idx_vecs[g]);
                    group_ht_free(&ht);
                    ray_free(val_block); ray_free(ivblock);
                    return ray_error("oom", NULL);
                }
            }
        }
        idx_vecs[gi] = ray_vec_append(idx_vecs[gi], &i);
    }
    group_ht_free(&ht);

    /* Build dict: keys vec mirrors x's element type; vals LIST of idx vecs. */
    int8_t key_type = x->type;
    ray_t* keys_vec;
    if (key_type == RAY_SYM) keys_vec = ray_sym_vec_new(RAY_SYM_W64, ngroups);
    else                     keys_vec = ray_vec_new(key_type, ngroups);
    if (RAY_IS_ERR(keys_vec)) goto gfail;

    for (int64_t g = 0; g < ngroups; g++) {
        switch (key_type) {
            case RAY_SYM:
            case RAY_I64:
            case RAY_TIMESTAMP: {
                int64_t v = gvals[g];
                keys_vec = ray_vec_append(keys_vec, &v); break;
            }
            case RAY_I32:
            case RAY_DATE:
            case RAY_TIME: {
                int32_t v = (int32_t)gvals[g];
                keys_vec = ray_vec_append(keys_vec, &v); break;
            }
            case RAY_I16: { int16_t v = (int16_t)gvals[g]; keys_vec = ray_vec_append(keys_vec, &v); break; }
            case RAY_BOOL:
            case RAY_U8:  { uint8_t v = (uint8_t)gvals[g]; keys_vec = ray_vec_append(keys_vec, &v); break; }
            case RAY_F64: {
                /* gvals[g] holds the IEEE-754 bit pattern packed by the
                 * row-loop above; reinterpret rather than int->double
                 * cast (which would produce 0.0/1.0/2.0… instead of the
                 * actual float values). */
                double v;
                memcpy(&v, &gvals[g], sizeof(v));
                keys_vec = ray_vec_append(keys_vec, &v);
                break;
            }
            case RAY_F32: {
                double f;
                memcpy(&f, &gvals[g], sizeof(f));
                float  v = (float)f;
                keys_vec = ray_vec_append(keys_vec, &v);
                break;
            }
            default:      keys_vec = ray_vec_append(keys_vec, &gvals[g]); break;
        }
        if (RAY_IS_ERR(keys_vec)) goto gfail;
        /* If the source column had a null at any row in this group, mark
         * the group's key as null so dict rendering / lookup can recover
         * the null semantics (the integer-value key alone collides with a
         * legitimate zero/sentinel value).  All rows in a value-equality
         * group share the same null-or-not status, so a single probe of
         * the first row index suffices. */
        if (idx_vecs[g] && idx_vecs[g]->len > 0) {
            int64_t first_row = ((int64_t*)ray_data(idx_vecs[g]))[0];
            if (ray_vec_is_null(x, first_row)) {
                ray_vec_set_null(keys_vec, g, true);
                /* Dual-encoding contract: the payload at slot g must hold
                 * the width-correct null sentinel so sentinel-aware readers
                 * agree with the bitmap. */
                void* base = ray_data(keys_vec);
                switch (key_type) {
                    case RAY_I64: case RAY_TIMESTAMP:
                        ((int64_t*)base)[g] = NULL_I64; break;
                    case RAY_I32: case RAY_DATE: case RAY_TIME:
                        ((int32_t*)base)[g] = NULL_I32; break;
                    case RAY_I16:
                        ((int16_t*)base)[g] = NULL_I16; break;
                    case RAY_F64:
                        ((double*)base)[g] = NULL_F64; break;
                    case RAY_F32:
                        ((float*)base)[g] = (float)NULL_F64; break;
                    default: break; /* SYM/BOOL/U8: no sentinel slot */
                }
            }
        }
    }

    /* SYM group keys are raw cell ids of x — resolve over x's domain. */
    if (key_type == RAY_SYM) ray_sym_vec_adopt_domain(keys_vec, x);

    ray_t* vals_lst = ray_list_new(ngroups);
    if (RAY_IS_ERR(vals_lst)) { ray_release(keys_vec); goto gfail; }
    for (int64_t g = 0; g < ngroups; g++) {
        vals_lst = ray_list_append(vals_lst, idx_vecs[g]);
        ray_release(idx_vecs[g]);
        idx_vecs[g] = NULL;
        if (RAY_IS_ERR(vals_lst)) { ray_release(keys_vec); goto gfail; }
    }
    ray_free(val_block);
    ray_free(ivblock);
    return ray_dict_new(keys_vec, vals_lst);

gfail:
    for (int64_t g = 0; g < ngroups; g++)
        if (idx_vecs[g]) ray_release(idx_vecs[g]);
    ray_free(val_block);
    ray_free(ivblock);
    return ray_error("oom", NULL);
}

static ray_t* str_vec_concat_atom(ray_t* vec, ray_t* atom, bool atom_first) {
    const char* ap = ray_str_ptr(atom);
    size_t al = ray_str_len(atom);
    int64_t n = vec->len;
    ray_t* out = ray_vec_new(RAY_STR, n > 0 ? n : 1);
    if (!out || RAY_IS_ERR(out)) return out;
    for (int64_t i = 0; i < n; i++) {
        size_t sl = 0;
        const char* sp = ray_str_vec_get(vec, i, &sl);
        if (!sp) { sp = ""; sl = 0; }
        size_t total = al + sl;
        char sbuf[8192];
        char* buf = sbuf;
        if (total > sizeof(sbuf)) {
            buf = ray_calloc_raw(total);
            if (!buf) {
                ray_release(out);
                return ray_error("oom", NULL);
            }
        }
        if (atom_first) {
            if (al) memcpy(buf, ap, al);
            if (sl) memcpy(buf + al, sp, sl);
        } else {
            if (sl) memcpy(buf, sp, sl);
            if (al) memcpy(buf + sl, ap, al);
        }
        ray_t* prev = out;
        out = ray_str_vec_append(out, buf, total);
        if (buf != sbuf) ray_free_raw(buf);
        if (RAY_IS_ERR(out)) {
            ray_release(prev);
            return out;
        }
    }
    return out;
}

/* (concat a b) -> concatenate vectors/strings/dicts/tables */
ray_t* ray_concat_fn(ray_t* a, ray_t* b) {
    /* Helper: get string content from atom (STR or CHAR), stripping trailing nulls */
    {
        int a_is_str = ray_is_atom(a) && ((-a->type) == RAY_STR);
        int b_is_str = ray_is_atom(b) && ((-b->type) == RAY_STR);
        if (a_is_str && b_is_str) {
            const char *ap, *bp;
            size_t la, lb;
            ap = ray_str_ptr(a); la = ray_str_len(a);
            bp = ray_str_ptr(b); lb = ray_str_len(b);
            /* Strip trailing null bytes */
            while (la > 0 && ap[la - 1] == '\0') la--;
            while (lb > 0 && bp[lb - 1] == '\0') lb--;
            char buf[8192];
            if (la + lb > sizeof(buf)) return ray_error("limit", NULL);
            memcpy(buf, ap, la);
            memcpy(buf + la, bp, lb);
            return ray_str(buf, la + lb);
        }
    }
    if (a->type == RAY_STR && ray_is_atom(b) && (-b->type) == RAY_STR)
        return str_vec_concat_atom(a, b, false);
    if (ray_is_atom(a) && (-a->type) == RAY_STR && b->type == RAY_STR)
        return str_vec_concat_atom(b, a, true);
    /* Vector concat: same type — delegate to ray_vec_concat which handles
     * null bitmap propagation, SYM width promotion, and STR pool merging. */
    if (ray_is_vec(a) && ray_is_vec(b) && a->type == b->type) {
        ray_t* r = ray_vec_concat(a, b);
        /* An append onto an indexed vector keeps the index when every
         * appended key is new (see ray_index_carry_append). */
        if (r && !RAY_IS_ERR(r) && ray_index_has(a)) ray_index_carry_append(a, r);
        return r;
    }
    /* Concat typed vec + boxed list or boxed list + typed vec -> boxed list */
    if ((ray_is_vec(a) && b->type == RAY_LIST) || (a->type == RAY_LIST && ray_is_vec(b))) {
        ray_t* la = (a->type == RAY_LIST) ? a : NULL;
        ray_t* lb = (b->type == RAY_LIST) ? b : NULL;
        ray_t* va = ray_is_vec(a) ? a : NULL;
        ray_t* vb = ray_is_vec(b) ? b : NULL;
        int64_t na = a->len, nb = b->len;
        ray_t* result = ray_alloc((na + nb) * sizeof(ray_t*));
        if (!result) return ray_error("oom", NULL);
        result->type = RAY_LIST;
        result->len = na + nb;
        ray_t** out = (ray_t**)ray_data(result);
        for (int64_t i = 0; i < na; i++) {
            if (va) {
                int alloc = 0;
                out[i] = collection_elem(va, i, &alloc);
            } else {
                out[i] = ((ray_t**)ray_data(la))[i];
                ray_retain(out[i]);
            }
        }
        for (int64_t i = 0; i < nb; i++) {
            if (vb) {
                int alloc = 0;
                out[na + i] = collection_elem(vb, i, &alloc);
            } else {
                out[na + i] = ((ray_t**)ray_data(lb))[i];
                ray_retain(out[na + i]);
            }
        }
        return result;
    }
    /* Boxed list concat */
    if (a->type == RAY_LIST && b->type == RAY_LIST) {
        int64_t na = a->len, nb = b->len;
        ray_t* result = ray_alloc((na + nb) * sizeof(ray_t*));
        if (!result) return ray_error("oom", NULL);
        result->type = RAY_LIST;
        result->len = na + nb;
        ray_t** out = (ray_t**)ray_data(result);
        ray_t** ae = (ray_t**)ray_data(a);
        ray_t** be = (ray_t**)ray_data(b);
        for (int64_t i = 0; i < na; i++) { ray_retain(ae[i]); out[i] = ae[i]; }
        for (int64_t i = 0; i < nb; i++) { ray_retain(be[i]); out[na + i] = be[i]; }
        return result;
    }
    /* Vector concat: mixed types -> boxed list (preserves original element types) */
    if (ray_is_vec(a) && ray_is_vec(b) && a->type != b->type) {
        int64_t na = a->len, nb = b->len;
        ray_t* result = ray_alloc((na + nb) * sizeof(ray_t*));
        if (!result) return ray_error("oom", NULL);
        result->type = RAY_LIST;
        result->len = na + nb;
        ray_t** out = (ray_t**)ray_data(result);
        for (int64_t i = 0; i < na; i++) {
            int alloc = 0;
            out[i] = collection_elem(a, i, &alloc);
            /* collection_elem always allocates for typed vecs, so ownership transfers */
        }
        for (int64_t i = 0; i < nb; i++) {
            int alloc = 0;
            out[na + i] = collection_elem(b, i, &alloc);
        }
        return result;
    }
    /* Atom + vector or vector + atom -> append */
    if (ray_is_atom(a) && ray_is_vec(b) && (-a->type) == b->type) {
        int64_t nb = b->len;
        int esz = ray_elem_size(b->type);
        ray_t* result = ray_vec_new(b->type, 1 + nb);
        if (RAY_IS_ERR(result)) return result;
        /* Copy atom value as first element */
        switch (b->type) {
        case RAY_I64: case RAY_TIMESTAMP: case RAY_SYM:
            ((int64_t*)ray_data(result))[0] = a->i64; break;
        case RAY_F64:
            ((double*)ray_data(result))[0] = a->f64; break;
        case RAY_I32: case RAY_DATE: case RAY_TIME:
            ((int32_t*)ray_data(result))[0] = a->i32; break;
        case RAY_I16:
            ((int16_t*)ray_data(result))[0] = a->i16; break;
        case RAY_BOOL:
            ((bool*)ray_data(result))[0] = a->b8; break;
        case RAY_U8:
            ((uint8_t*)ray_data(result))[0] = a->u8; break;
        case RAY_GUID: {
            const uint8_t* gd = a->obj ? (const uint8_t*)ray_data(a->obj) : (const uint8_t*)ray_data((ray_t*)a);
            memcpy(ray_data(result), gd, 16); break;
        }
        default: ray_free(result); return ray_error("type", "concat: unsupported element type %s", ray_type_name(b->type));
        }
        /* Null propagation: propagate null-ness of the leading atom
         * (a typed-null sentinel written raw at slot 0) and any nulls in the
         * trailing vector b. */
        if (RAY_ATOM_IS_NULL(a) || ray_vec_may_have_nulls(b))
            result->attrs |= RAY_ATTR_HAS_NULLS;
        if (b->type == RAY_SYM) {
            /* Mixed sources: the atom id is runtime-domain by design, b's
             * cells are b-domain — re-express b per cell into the runtime
             * domain (the output's; no-op while b is runtime-domain). */
            int64_t* out = (int64_t*)ray_data(result);
            for (int64_t i = 0; i < nb; i++)
                out[1 + i] = sym_cell_runtime_id(b, i);
        } else {
            memcpy((char*)ray_data(result) + esz, ray_data(b), (size_t)(nb * esz));
        }
        result->len = 1 + nb;
        return result;
    }
    if (ray_is_vec(a) && ray_is_atom(b) && a->type == (-b->type)) {
        int64_t na = a->len;
        int esz = ray_elem_size(a->type);
        ray_t* result = ray_vec_new(a->type, na + 1);
        if (RAY_IS_ERR(result)) return result;
        if (a->type == RAY_SYM) {
            /* Mixed sources (see atom+vec case above): re-express a's
             * cells into the output's runtime domain. */
            int64_t* out = (int64_t*)ray_data(result);
            for (int64_t i = 0; i < na; i++)
                out[i] = sym_cell_runtime_id(a, i);
        } else {
            memcpy(ray_data(result), ray_data(a), (size_t)(na * esz));
        }
        switch (a->type) {
        case RAY_I64: case RAY_TIMESTAMP: case RAY_SYM:
            ((int64_t*)ray_data(result))[na] = b->i64; break;
        case RAY_F64:
            ((double*)ray_data(result))[na] = b->f64; break;
        case RAY_I32: case RAY_DATE: case RAY_TIME:
            ((int32_t*)ray_data(result))[na] = b->i32; break;
        case RAY_I16:
            ((int16_t*)ray_data(result))[na] = b->i16; break;
        case RAY_BOOL:
            ((bool*)ray_data(result))[na] = b->b8; break;
        case RAY_U8:
            ((uint8_t*)ray_data(result))[na] = b->u8; break;
        case RAY_GUID: {
            const uint8_t* gd = b->obj ? (const uint8_t*)ray_data(b->obj) : (const uint8_t*)ray_data((ray_t*)b);
            memcpy((uint8_t*)ray_data(result) + na * 16, gd, 16); break;
        }
        default: ray_free(result); return ray_error("type", "concat: unsupported element type %s", ray_type_name(a->type));
        }
        result->len = na + 1;
        /* Null propagation: propagate nulls in the leading vector a
         * and null-ness of the trailing atom (sentinel written raw at na). */
        if (ray_vec_may_have_nulls(a) || RAY_ATOM_IS_NULL(b))
            result->attrs |= RAY_ATTR_HAS_NULLS;
        if (ray_index_has(a)) ray_index_carry_append(a, result);
        return result;
    }
    /* Atom + atom of same type -> 2-element vector */
    if (ray_is_atom(a) && ray_is_atom(b) && a->type == b->type && a->type != -RAY_STR) {
        int8_t vtype = -(a->type);
        ray_t* result = ray_vec_new(vtype, 2);
        if (RAY_IS_ERR(result)) return result;
        result->len = 2;
        switch (vtype) {
        case RAY_I64: case RAY_TIMESTAMP: case RAY_SYM:
            ((int64_t*)ray_data(result))[0] = a->i64;
            ((int64_t*)ray_data(result))[1] = b->i64;
            break;
        case RAY_F64:
            ((double*)ray_data(result))[0] = a->f64;
            ((double*)ray_data(result))[1] = b->f64;
            break;
        case RAY_I32: case RAY_DATE: case RAY_TIME:
            ((int32_t*)ray_data(result))[0] = a->i32;
            ((int32_t*)ray_data(result))[1] = b->i32;
            break;
        case RAY_I16:
            ((int16_t*)ray_data(result))[0] = a->i16;
            ((int16_t*)ray_data(result))[1] = b->i16;
            break;
        case RAY_BOOL:
            ((bool*)ray_data(result))[0] = a->b8;
            ((bool*)ray_data(result))[1] = b->b8;
            break;
        case RAY_U8:
            ((uint8_t*)ray_data(result))[0] = a->u8;
            ((uint8_t*)ray_data(result))[1] = b->u8;
            break;
        case RAY_GUID: {
            const uint8_t* ga = a->obj ? (const uint8_t*)ray_data(a->obj) : (const uint8_t*)ray_data((ray_t*)a);
            const uint8_t* gb = b->obj ? (const uint8_t*)ray_data(b->obj) : (const uint8_t*)ray_data((ray_t*)b);
            memcpy(ray_data(result), ga, 16);
            memcpy((uint8_t*)ray_data(result) + 16, gb, 16);
            break;
        }
        default: ray_free(result); return ray_error("type", "concat: unsupported element type %s", ray_type_name(vtype));
        }
        /* Null propagation: a typed-null atom (e.g. the I64 literal
         * INT64_MIN == NULL_I64) writes its sentinel bit-pattern straight
         * into the result column.  Flag HAS_NULLS so null-aware ops don't
         * treat the sentinel as a real value. */
        if (RAY_ATOM_IS_NULL(a) || RAY_ATOM_IS_NULL(b))
            result->attrs |= RAY_ATTR_HAS_NULLS;
        return result;
    }
    /* Dict concat: merge — keys/vals from b overwrite a's. */
    if (a->type == RAY_DICT && b->type == RAY_DICT) {
        ray_retain(a);
        ray_t* out = a;
        ray_t* bk = ray_dict_keys(b);
        ray_t* bv = ray_dict_vals(b);
        if (!bk || !bv) return out;
        int64_t bn = bk->len;
        for (int64_t i = 0; i < bn; i++) {
            /* Synthesize a key atom view from bk and the value pointer from bv. */
            ray_t k_storage; memset(&k_storage, 0, sizeof(k_storage));
            ray_t* k = NULL;
            if (bk->type == RAY_LIST) {
                k = ((ray_t**)ray_data(bk))[i];
            } else if (bk->type == RAY_SYM) {
                k_storage.type = -RAY_SYM;
                k_storage.i64  = ray_read_sym(ray_data(bk), i, RAY_SYM, bk->attrs);
                k = &k_storage;
            } else if (bk->type == RAY_I64 || bk->type == RAY_TIMESTAMP) {
                k_storage.type = -bk->type;
                k_storage.i64  = ((int64_t*)ray_data(bk))[i];
                k = &k_storage;
            } else {
                /* Heterogeneous element types fall back to boxing via collection_elem. */
                int alloc = 0;
                k = collection_elem(bk, i, &alloc);
                ray_t* v;
                if (bv->type == RAY_LIST) v = ((ray_t**)ray_data(bv))[i];
                else { int va = 0; v = collection_elem(bv, i, &va); (void)va; }
                out = ray_dict_upsert(out, k, v);
                if (alloc) ray_release(k);
                if (!out || RAY_IS_ERR(out)) return out;
                continue;
            }
            ray_t* v;
            if (bv->type == RAY_LIST) v = ((ray_t**)ray_data(bv))[i];
            else { int va = 0; v = collection_elem(bv, i, &va); (void)va; }
            out = ray_dict_upsert(out, k, v);
            if (!out || RAY_IS_ERR(out)) return out;
        }
        return out;
    }
    /* Table concat: append rows */
    if (a->type == RAY_TABLE && b->type == RAY_TABLE) {
        int64_t ncols_a = ray_table_ncols(a);
        int64_t ncols_b = ray_table_ncols(b);
        /* Match columns of a in b by name */
        ray_t* result = ray_table_new((int32_t)ncols_a);
        if (RAY_IS_ERR(result)) return result;
        for (int64_t c = 0; c < ncols_a; c++) {
            int64_t col_name_a = ray_table_col_name(a, c);
            ray_t* acol = ray_table_get_col_idx(a, c);
            /* Find matching column in b by name */
            ray_t* bcol = NULL;
            for (int64_t j = 0; j < ncols_b; j++) {
                if (ray_table_col_name(b, j) == col_name_a) {
                    bcol = ray_table_get_col_idx(b, j);
                    break;
                }
            }
            if (!bcol) {
                /* Column not present in b — schema mismatch is a "value"
                 * error (the table values have incompatible columns), not
                 * a "domain" error (which semantically means out-of-range). */
                ray_release(result);
                return ray_error("value", NULL);
            }
            /* Type check: columns must have the same type */
            if (acol->type != bcol->type) {
                ray_release(result);
                return ray_error("type", "concat: column type mismatch, got %s and %s", ray_type_name(acol->type), ray_type_name(bcol->type));
            }
            ray_t* col = ray_concat_fn(acol, bcol);
            if (RAY_IS_ERR(col)) { ray_release(result); return col; }
            result = ray_table_add_col(result, col_name_a, col);
            ray_release(col);
            if (RAY_IS_ERR(result)) return result;
        }
        return result;
    }
    /* Atom + boxed list -> prepend atom to list */
    if (ray_is_atom(a) && b->type == RAY_LIST && b->type != RAY_DICT) {
        int64_t nb = b->len;
        ray_t* result = ray_alloc((1 + nb) * sizeof(ray_t*));
        if (!result) return ray_error("oom", NULL);
        result->type = RAY_LIST;
        result->len = 1 + nb;
        ray_t** out = (ray_t**)ray_data(result);
        ray_retain(a);
        out[0] = a;
        ray_t** be = (ray_t**)ray_data(b);
        for (int64_t i = 0; i < nb; i++) { ray_retain(be[i]); out[1 + i] = be[i]; }
        return result;
    }
    /* Boxed list + atom -> append atom to list */
    if (a->type == RAY_LIST && a->type != RAY_DICT && ray_is_atom(b)) {
        int64_t na = a->len;
        ray_t* result = ray_alloc((na + 1) * sizeof(ray_t*));
        if (!result) return ray_error("oom", NULL);
        result->type = RAY_LIST;
        result->len = na + 1;
        ray_t** out = (ray_t**)ray_data(result);
        ray_t** ae = (ray_t**)ray_data(a);
        for (int64_t i = 0; i < na; i++) { ray_retain(ae[i]); out[i] = ae[i]; }
        ray_retain(b);
        out[na] = b;
        return result;
    }
    /* Atom + atom of different types -> 2-element boxed list */
    if (ray_is_atom(a) && ray_is_atom(b) && a->type != b->type) {
        ray_t* result = ray_alloc(2 * sizeof(ray_t*));
        if (!result) return ray_error("oom", NULL);
        result->type = RAY_LIST;
        result->len = 2;
        ray_t** out = (ray_t**)ray_data(result);
        ray_retain(a); out[0] = a;
        ray_retain(b); out[1] = b;
        return result;
    }
    return ray_error("type", "concat: cannot concatenate %s and %s", ray_type_name(a->type), ray_type_name(b->type));
}

/* (raze list-of-vecs) -> flattened vector */
ray_t* ray_raze_fn(ray_t* x) {
    /* Scalar passthrough */
    if (ray_is_atom(x)) { ray_retain(x); return x; }
    /* Typed vector passthrough */
    if (ray_is_vec(x)) { ray_retain(x); return x; }
    if (x->type != RAY_LIST)
        return ray_error("type", "raze: argument must be a list, got %s", ray_type_name(x->type));
    int64_t n = x->len;
    if (n == 0) return ray_list_new(0);
    ray_t** items = (ray_t**)ray_data(x);

    /* Fast path: all items are vectors of the same primitive type
     * (numeric/temporal, fixed-width, no SYM/STR/GUID/LIST/null).
     * Pre-size one output vector and memcpy each item's data — O(total)
     * instead of the pairwise concat loop's O(N²). */
    if (ray_is_vec(items[0])) {
        int8_t t = items[0]->type;
        bool fast = (t != RAY_LIST && t != RAY_STR && t != RAY_SYM && t != RAY_GUID);
        int64_t total = 0;
        if (fast) {
            for (int64_t i = 0; i < n; i++) {
                ray_t* it = items[i];
                if (!ray_is_vec(it) || it->type != t
                    || ray_vec_may_have_nulls(it)) {
                    fast = false; break;
                }
                total += it->len;
            }
        }
        if (fast) {
            ray_t* out = ray_vec_new(t, total);
            if (!out || RAY_IS_ERR(out)) return out ? out : ray_error("oom", NULL);
            out->len = total;
            uint8_t esz = ray_elem_size(t);
            char* dst = (char*)ray_data(out);
            int64_t pos = 0;
            for (int64_t i = 0; i < n; i++) {
                int64_t k = items[i]->len;
                if (k > 0) memcpy(dst + pos * esz, ray_data(items[i]), (size_t)k * esz);
                pos += k;
            }
            return out;
        }
    }

    /* Slow path: pairwise concat — used for mixed types, null-bearing
     * inputs, and non-fixed-width vectors (SYM/STR/GUID/LIST). */
    ray_t* result = items[0];
    ray_retain(result);
    for (int64_t i = 1; i < n; i++) {
        ray_t* next = ray_concat_fn(result, items[i]);
        ray_release(result);
        if (RAY_IS_ERR(next)) return next;
        result = next;
    }
    return result;
}

/* ungroup t — flatten nested-list columns, replicate flat columns.
 *
 * Grouping a non-aggregate (e.g. `top(2) by k`) yields a table with a
 * nested LIST column (one list-of-values per group) alongside ATOM key
 * columns.  `ungroup` expands that to SQL/row-form shape: per row i, the
 * nested columns' cells have a common length n_i; flat columns are
 * replicated n_i times and nested columns are razed (concatenated) in
 * row order.  Total output rows = Σ_i n_i (n_i may be 0 → row drops out).
 *
 *   - No nested column → return `t` unchanged (identity).
 *   - Nested columns disagreeing on a row's length → ray_error("length").
 *
 * A nested cell that is an atom counts as length 1 (broadcast); otherwise
 * its `ray_len` is used.  Flat-column replication uses gather_by_idx
 * (preserves type/attrs/null bits/SYM domain); nested-column flatten
 * reuses ray_raze_fn (handles the fast memcpy path, nulls, empties). */
ray_t* ray_ungroup_fn(ray_t* x) {
    if (!x || RAY_IS_ERR(x) || x->type != RAY_TABLE)
        return ray_error("type", "ungroup: argument must be a table, got %s", x ? ray_type_name(x->type) : "null");

    int64_t ncols = ray_table_ncols(x);
    int64_t nrows = ray_table_nrows(x);

    /* Locate the nested (RAY_LIST) columns. */
    bool any_nested = false;
    for (int64_t c = 0; c < ncols; c++) {
        ray_t* col = ray_table_get_col_idx(x, c);
        if (col && !RAY_IS_ERR(col) && col->type == RAY_LIST) { any_nested = true; break; }
    }
    /* No nested column → identity. */
    if (!any_nested) { ray_retain(x); return x; }

    /* Per-row expansion count n_i from the nested columns; all nested
     * columns must agree on n_i for each row.  Build a replication index
     * vector idx[] = each row i repeated n_i times (for flat gather). */
    int64_t total = 0;
    int64_t* counts = (int64_t*)ray_alloc_raw((size_t)(nrows > 0 ? nrows : 1) * sizeof(int64_t));
    if (!counts) return ray_error("oom", NULL);

    for (int64_t i = 0; i < nrows; i++) {
        int64_t n_i = -1;
        for (int64_t c = 0; c < ncols; c++) {
            ray_t* col = ray_table_get_col_idx(x, c);
            if (!col || RAY_IS_ERR(col) || col->type != RAY_LIST) continue;
            ray_t* cell = ray_list_get(col, i);
            /* A valid nested cell is either an atom (length 1) or a typed
             * NON-LIST vector (length cell->len).  Anything else — NULL
             * (unfilled list slot → ray_len would deref NULL) or a boxed
             * RAY_LIST cell (whose top-level len would disagree with the
             * scalar count raze emits, corrupting column lengths) — is a
             * type error. */
            int64_t len_i;
            if (cell && ray_is_atom(cell))
                len_i = 1;
            else if (cell && ray_is_vec(cell) && cell->type != RAY_LIST)
                len_i = ray_len(cell);
            else { ray_free_raw(counts); return ray_error("type", "ungroup: nested cell must be an atom or non-list vector"); }
            if (n_i < 0) n_i = len_i;
            else if (n_i != len_i) { ray_free_raw(counts); return ray_error("length", "ungroup: nested columns disagree on row length, got %lld and %lld", (long long)n_i, (long long)len_i); }
        }
        if (n_i < 0) n_i = 0;
        counts[i] = n_i;
        total += n_i;
    }

    ray_t* idx = ray_vec_new(RAY_I64, total > 0 ? total : 1);
    if (!idx || RAY_IS_ERR(idx)) { ray_free_raw(counts); return idx ? idx : ray_error("oom", NULL); }
    idx->len = total;
    int64_t* idxd = (int64_t*)ray_data(idx);
    int64_t pos = 0;
    for (int64_t i = 0; i < nrows; i++)
        for (int64_t r = 0; r < counts[i]; r++) idxd[pos++] = i;

    /* Rebuild every column, preserving names and order. */
    ray_t* out = ray_table_new(ncols);
    if (!out || RAY_IS_ERR(out)) { ray_free_raw(counts); ray_release(idx); return out ? out : ray_error("oom", NULL); }

    for (int64_t c = 0; c < ncols; c++) {
        ray_t* col = ray_table_get_col_idx(x, c);
        int64_t name = ray_table_col_name(x, c);
        ray_t* newcol;
        if (col && !RAY_IS_ERR(col) && col->type == RAY_LIST) {
            /* Nested: raze the per-row cells in row order. */
            newcol = ray_raze_fn(col);
            /* raze of an empty/all-empty list yields an empty LIST — leave
             * it; the column is empty (total == 0) and shapes still align. */
            /* Defensive: with the up-front nested-cell validation, the razed
             * column length must equal the row count.  Convert any future
             * invariant break into a clean error instead of a corrupt table. */
            if (newcol && !RAY_IS_ERR(newcol) && ray_len(newcol) != total) {
                ray_release(newcol); ray_release(out); ray_release(idx); ray_free_raw(counts);
                return ray_error("length", "ungroup: razed column length mismatch, got %lld, expected %lld", (long long)ray_len(newcol), (long long)total);
            }
        } else if (col && !RAY_IS_ERR(col)) {
            /* Flat: gather row i repeated n_i times. */
            newcol = gather_by_idx(col, idxd, total);
        } else {
            ray_release(out); ray_release(idx); ray_free_raw(counts);
            return ray_error("type", "ungroup: invalid column");
        }
        if (!newcol || RAY_IS_ERR(newcol)) {
            ray_release(out); ray_release(idx); ray_free_raw(counts);
            return newcol ? newcol : ray_error("oom", NULL);
        }
        ray_t* added = ray_table_add_col(out, name, newcol);
        ray_release(newcol);
        if (!added || RAY_IS_ERR(added)) {
            ray_release(idx); ray_free_raw(counts);
            return added ? added : ray_error("oom", NULL);
        }
        out = added;
    }

    ray_release(idx);
    ray_free_raw(counts);
    return out;
}

/* (within vals [lo hi]) -> bool vector, true where lo <= val <= hi */
ray_t* ray_within_fn(ray_t* vals, ray_t* range) {
    if (!ray_is_vec(vals) || !ray_is_vec(range))
        return ray_error("type", "within: vals and range must be vectors, got %s and %s", ray_type_name(vals->type), ray_type_name(range->type));
    if (range->len != 2)
        return ray_error("length", "within: range must have 2 elements, got %lld", (long long)range->len);
    int64_t n = vals->len;
    ray_t* result = ray_vec_new(RAY_BOOL, n);
    if (RAY_IS_ERR(result)) return result;
    bool* out = (bool*)ray_data(result);

    if (vals->type == RAY_I64) {
        int64_t* d = (int64_t*)ray_data(vals);
        int64_t* r = (int64_t*)ray_data(range);
        int64_t lo = r[0], hi = r[1];
        for (int64_t i = 0; i < n; i++) out[i] = (d[i] >= lo && d[i] <= hi);
    } else if (vals->type == RAY_F64) {
        double* d = (double*)ray_data(vals);
        double* r = (double*)ray_data(range);
        double lo = r[0], hi = r[1];
        for (int64_t i = 0; i < n; i++) out[i] = (d[i] >= lo && d[i] <= hi);
    } else if (vals->type == RAY_I32 || vals->type == RAY_DATE || vals->type == RAY_TIME) {
        int32_t* d = (int32_t*)ray_data(vals);
        int32_t* r = (int32_t*)ray_data(range);
        int32_t lo = r[0], hi = r[1];
        for (int64_t i = 0; i < n; i++) out[i] = (d[i] >= lo && d[i] <= hi);
    } else {
        ray_free(result);
        return ray_error("type", "within: unsupported value type %s", ray_type_name(vals->type));
    }
    result->len = n;
    return result;
}