rayforce-sys 1.1.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.
 */

#include "sym.h"
#include "core/platform.h"
#include "store/fileio.h"
#include "mem/heap.h"
#include "mem/sys.h"
#include "mem/arena.h"
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdatomic.h>
#include <errno.h>
#include "ops/hash.h"

/* --------------------------------------------------------------------------
 * Symbol table structure (static global, sequential mode only).
 * NOT thread-safe: all interning must happen before ray_parallel_begin().
 * -------------------------------------------------------------------------- */

#define SYM_INIT_CAP     256
#define SYM_LOAD_FACTOR  0.7
#define SYM_STRL_MAGIC   0x4C525453U  /* "STRL" */
#define SYM_LAZY_LOAD_MIN_BYTES (64u * 1024u * 1024u)

/* Cached segment list for a dotted sym: nsegs sym_ids that together make up
 * the dotted path.  segs is arena-allocated (same lifetime as sym table). */
typedef struct {
    uint8_t  nsegs;
    int64_t* segs;   /* length nsegs; NULL for non-dotted entries */
} sym_segs_t;

typedef struct {
    /* Hash table: each bucket stores (hash32 << 32) | (id + 1), 0 = empty */
    uint64_t*  buckets;
    uint32_t   bucket_cap;   /* always power of 2 */

    /* String array: strings[id] = ray_t* string atom */
    ray_t**     strings;
    uint32_t   str_count;
    uint32_t   str_cap;

    /* Per-sym dotted-path metadata, parallel to strings[].
     * `dotted` is a bitmap (1 bit per sym_id); bit set = name is dotted
     *   and segment sym_ids are cached in `segments`.
     * `scanned` is a bitmap; bit set = sym_cache_segments has settled this
     *   sym (either cached successfully, or decided it is a plain name).
     *   Unset = needs to be (re-)scanned on the next intern call, which is
     *   how we recover from a transient cache OOM on first intern: the
     *   bit stays clear, so future interns of the same name retry.
     * `segments` holds cached segment sym_ids; segs == NULL when dotted
     *   bit is clear. */
    uint64_t*   dotted;       /* (str_cap + 63) / 64 words */
    uint64_t*   scanned;      /* (str_cap + 63) / 64 words */
    sym_segs_t* segments;     /* length str_cap */

    /* Large on-disk dictionaries stay mapped and are materialized by id.
     * lazy_count = number of records in the mapped file; ids in
     * [lazy_next_id, lazy_count) are still backed only by the mapping. */
    uint8_t*      lazy_map;
    size_t        lazy_size;
    uint32_t      lazy_count;
    uint32_t      lazy_next_id;
    const uint8_t* lazy_ptr;
    size_t        lazy_remaining;

    /* Arena for string atoms — avoids per-string buddy allocator calls */
    ray_arena_t*  arena;
} sym_table_t;

static sym_table_t g_sym;
static _Atomic(bool) g_sym_inited = false;
static bool sym_lazy_materialize_to_locked(uint32_t target_id);

/* Spinlock protecting g_sym mutations in ray_sym_intern */
static _Atomic(int) g_sym_lock = 0;
static inline void sym_lock(void) {
    while (atomic_exchange_explicit(&g_sym_lock, 1, memory_order_acquire)) {
        RAY_CPU_RELAX();
    }
}
static inline void sym_unlock(void) {
    atomic_store_explicit(&g_sym_lock, 0, memory_order_release);
}

/* String atoms for the global sym table live in its arena (off the per-thread
 * buddy heap) — see ray_arena_str in mem/arena.c. */

/* Forward decl — used from ray_sym_init below to reserve sym ID 0 as
 * the canonical empty string.  Definition is further down with the
 * other intern helpers. */
static int64_t sym_intern_nolock(uint32_t hash, const char* str, size_t len,
                                 bool search_lazy);

/* --------------------------------------------------------------------------
 * ray_sym_init
 * -------------------------------------------------------------------------- */

static _Atomic uint64_t g_sym_epoch_ctr = 0;

uint64_t ray_sym_epoch(void) {
    return atomic_load_explicit(&g_sym_epoch_ctr, memory_order_acquire);
}

ray_err_t ray_sym_init(void) {
    atomic_fetch_add_explicit(&g_sym_epoch_ctr, 1, memory_order_release);
    bool expected = false;
    if (!atomic_compare_exchange_strong_explicit(&g_sym_inited, &expected, true,
            memory_order_acq_rel, memory_order_acquire))
        return RAY_OK; /* already initialized by another thread */

    /* RAY_SYM_AUDIT: cache the env check once (the hot ray_sym_vec_cell
     * inline branches on the cached byte, never on getenv). */
    ray_g_sym_audit = getenv("RAY_SYM_AUDIT") != NULL ? 1 : 0;

    g_sym.bucket_cap = SYM_INIT_CAP;
    /* ray_sys_alloc uses mmap(MAP_ANONYMOUS) which zero-initializes. */
    g_sym.buckets = (uint64_t*)ray_sys_alloc(g_sym.bucket_cap * sizeof(uint64_t));
    if (!g_sym.buckets) {
        atomic_store_explicit(&g_sym_inited, false, memory_order_release);
        return RAY_ERR_OOM;
    }

    g_sym.str_cap = SYM_INIT_CAP;
    g_sym.str_count = 0;
    g_sym.strings = (ray_t**)ray_sys_alloc(g_sym.str_cap * sizeof(ray_t*));
    if (!g_sym.strings) {
        ray_sys_free(g_sym.buckets);
        g_sym.buckets = NULL;
        atomic_store_explicit(&g_sym_inited, false, memory_order_release);
        return RAY_ERR_OOM;
    }

    g_sym.arena = ray_arena_new(1024 * 1024);  /* 1MB chunks */
    if (!g_sym.arena) {
        ray_sys_free(g_sym.strings);
        ray_sys_free(g_sym.buckets);
        g_sym.strings = NULL;
        g_sym.buckets = NULL;
        atomic_store_explicit(&g_sym_inited, false, memory_order_release);
        return RAY_ERR_OOM;
    }

    /* Dotted-path sidecars sized to str_cap.  ray_sys_alloc is MAP_ANONYMOUS
     * so memory is zero-initialised — bitmaps start all-zero, segments[i]
     * structs start {nsegs:0, segs:NULL}.  Failures free prior allocations
     * and roll the sym table back to uninitialised. */
    uint32_t bm_words = (g_sym.str_cap + 63) / 64;
    g_sym.dotted = (uint64_t*)ray_sys_alloc((size_t)bm_words * sizeof(uint64_t));
    g_sym.scanned = (uint64_t*)ray_sys_alloc((size_t)bm_words * sizeof(uint64_t));
    g_sym.segments = (sym_segs_t*)ray_sys_alloc((size_t)g_sym.str_cap * sizeof(sym_segs_t));
    if (!g_sym.dotted || !g_sym.scanned || !g_sym.segments) {
        if (g_sym.dotted) ray_sys_free(g_sym.dotted);
        if (g_sym.scanned) ray_sys_free(g_sym.scanned);
        if (g_sym.segments) ray_sys_free(g_sym.segments);
        g_sym.dotted = NULL;
        g_sym.scanned = NULL;
        g_sym.segments = NULL;
        ray_arena_destroy(g_sym.arena);
        g_sym.arena = NULL;
        ray_sys_free(g_sym.strings);
        ray_sys_free(g_sym.buckets);
        g_sym.strings = NULL;
        g_sym.buckets = NULL;
        atomic_store_explicit(&g_sym_inited, false, memory_order_release);
        return RAY_ERR_OOM;
    }

    /* Reserve sym ID 0 as the empty string.  This makes the empty
     * symbol the canonical "no value" representation for SYM columns:
     * a missing CSV cell, a null-marked SYM atom, and an explicit ""
     * literal all collapse to ID 0.  SYM columns therefore never need
     * a parallel null bitmap or HAS_NULLS flag to identify null cells.  Done before
     * returning so every subsequent intern observes ID 0 as taken. */
    int64_t empty_id = sym_intern_nolock(
        (uint32_t)ray_hash_bytes("", 0), "", 0, true);
    if (empty_id != 0) {
        /* Should be unreachable — table just initialised, no other
         * thread has touched it yet.  If it ever fires, fail loudly. */
        ray_arena_destroy(g_sym.arena);
        ray_sys_free(g_sym.segments);
        ray_sys_free(g_sym.scanned);
        ray_sys_free(g_sym.dotted);
        ray_sys_free(g_sym.strings);
        ray_sys_free(g_sym.buckets);
        g_sym.arena = NULL;
        g_sym.segments = NULL; g_sym.scanned = NULL; g_sym.dotted = NULL;
        g_sym.strings = NULL; g_sym.buckets = NULL;
        atomic_store_explicit(&g_sym_inited, false, memory_order_release);
        return RAY_ERR_OOM;
    }

    /* g_sym_inited already set to true by CAS above */
    return RAY_OK;
}

/* --------------------------------------------------------------------------
 * ray_sym_destroy
 * -------------------------------------------------------------------------- */

void ray_sym_destroy(void) {
    atomic_fetch_add_explicit(&g_sym_epoch_ctr, 1, memory_order_release);
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return;

    if (g_sym.lazy_map) {
        ray_vm_unmap_file(g_sym.lazy_map, g_sym.lazy_size);
        g_sym.lazy_map = NULL;
        g_sym.lazy_size = 0;
        g_sym.lazy_count = 0;
        g_sym.lazy_next_id = 0;
        g_sym.lazy_ptr = NULL;
        g_sym.lazy_remaining = 0;
    }

    /* Arena-backed strings: ray_release is a no-op (RAY_ATTR_ARENA).
     * Destroy the arena to free all string atoms at once.
     * segments[i].segs pointers are arena-allocated too, freed with it. */
    if (g_sym.arena) {
        ray_arena_destroy(g_sym.arena);
        g_sym.arena = NULL;
    }

    if (g_sym.segments) ray_sys_free(g_sym.segments);
    if (g_sym.scanned)  ray_sys_free(g_sym.scanned);
    if (g_sym.dotted)   ray_sys_free(g_sym.dotted);
    ray_sys_free(g_sym.strings);
    ray_sys_free(g_sym.buckets);

    memset(&g_sym, 0, sizeof(g_sym));
    atomic_store_explicit(&g_sym_inited, false, memory_order_release);
}

/* --------------------------------------------------------------------------
 * Hash table helpers
 * -------------------------------------------------------------------------- */

static void ht_insert(uint64_t* buckets, uint32_t cap, uint32_t hash, uint32_t id) {
    uint32_t mask = cap - 1;
    uint32_t slot = hash & mask;
    uint64_t entry = ((uint64_t)hash << 32) | ((uint64_t)(id + 1));

    for (;;) {
        if (buckets[slot] == 0) {
            buckets[slot] = entry;
            return;
        }
        slot = (slot + 1) & mask;
    }
}

/* Grow hash table to new_cap (must be power of 2 and > current cap). */
static bool ht_grow_to(uint32_t new_cap) {
    uint64_t* new_buckets = (uint64_t*)ray_sys_alloc((size_t)new_cap * sizeof(uint64_t));
    if (!new_buckets) return false;

    /* Re-insert all existing entries */
    for (uint32_t i = 0; i < g_sym.bucket_cap; i++) {
        uint64_t e = g_sym.buckets[i];
        if (e == 0) continue;
        uint32_t h = (uint32_t)(e >> 32);
        uint32_t id = (uint32_t)(e & 0xFFFFFFFF) - 1;
        ht_insert(new_buckets, new_cap, h, id);
    }

    ray_sys_free(g_sym.buckets);
    g_sym.buckets = new_buckets;
    g_sym.bucket_cap = new_cap;
    return true;
}

static bool ht_grow(void) {
    /* Overflow guard: bucket_cap is always power of 2.
     * At 2^31, doubling overflows uint32_t. */
    if (g_sym.bucket_cap >= (UINT32_MAX / 2 + 1)) return false;
    return ht_grow_to(g_sym.bucket_cap * 2);
}

/* --------------------------------------------------------------------------
 * sym_grow_str_cap — grow strings[], dotted[] bitmap, and segments[] array
 * to hold at least new_cap entries.  Must be called with sym_lock held
 * (or from within single-threaded prehashed intern).  Zero-fills the new
 * portion of segments[] explicitly (realloc of a mapped region may return
 * pages that weren't touched but we don't want to rely on virgin mmap).
 * -------------------------------------------------------------------------- */
static bool sym_grow_str_cap(uint32_t new_cap) {
    uint32_t old_cap = g_sym.str_cap;
    if (new_cap <= old_cap) return true;

    ray_t** new_strings = (ray_t**)ray_sys_realloc(g_sym.strings,
                                                   (size_t)new_cap * sizeof(ray_t*));
    if (!new_strings) return false;
    memset(new_strings + old_cap, 0, (size_t)(new_cap - old_cap) * sizeof(ray_t*));
    g_sym.strings = new_strings;

    uint32_t old_bm_words = (old_cap + 63) / 64;
    uint32_t new_bm_words = (new_cap + 63) / 64;
    if (new_bm_words > old_bm_words) {
        uint64_t* new_dotted = (uint64_t*)ray_sys_realloc(g_sym.dotted,
                                                          (size_t)new_bm_words * sizeof(uint64_t));
        if (!new_dotted) return false;
        memset(new_dotted + old_bm_words, 0,
               (size_t)(new_bm_words - old_bm_words) * sizeof(uint64_t));
        g_sym.dotted = new_dotted;

        uint64_t* new_scanned = (uint64_t*)ray_sys_realloc(g_sym.scanned,
                                                           (size_t)new_bm_words * sizeof(uint64_t));
        if (!new_scanned) return false;
        memset(new_scanned + old_bm_words, 0,
               (size_t)(new_bm_words - old_bm_words) * sizeof(uint64_t));
        g_sym.scanned = new_scanned;
    }

    sym_segs_t* new_segments = (sym_segs_t*)ray_sys_realloc(g_sym.segments,
                                                            (size_t)new_cap * sizeof(sym_segs_t));
    if (!new_segments) return false;
    memset(new_segments + old_cap, 0,
           (size_t)(new_cap - old_cap) * sizeof(sym_segs_t));
    g_sym.segments = new_segments;

    g_sym.str_cap = new_cap;
    return true;
}

/* Forward declarations — sym_cache_segments (below) needs these helpers
 * that are defined further down in the file.  ray_sym_bytes_upper is
 * declared in sym.h as a public inline so both the intern path and the
 * test suite can refer to the same formula. */
static int64_t sym_intern_nolock(uint32_t hash, const char* str, size_t len,
                                 bool search_lazy);
static int64_t sym_probe(uint32_t hash, const char* str, size_t len);
static int64_t sym_commit_new(uint32_t hash, const char* str, size_t len);
static bool    sym_reserve_capacity(uint32_t new_sym_count, size_t arena_bytes);
static bool    sym_lazy_materialize_to_locked(uint32_t target_id);

/* --------------------------------------------------------------------------
 * sym_cache_segments — idempotent cache-and-apply for an EXISTING sym.
 * Used by ray_sym_rebuild_segments (after bulk persistence loads) and by
 * the probe-found branch of sym_intern_nolock (a prior intern via
 * ray_sym_intern_no_split may have committed the sym without ever
 * running the cache prep).
 *
 * Atomic: same inspect + reserve + commit pattern as sym_intern_nolock,
 * so a failure here leaves no orphan segment syms and no half-applied
 * cache state.  Returns false only on real OOM — scanned stays clear
 * in that case so future retries pick up where we left off.
 * -------------------------------------------------------------------------- */
static bool sym_cache_segments(uint32_t new_id, const char* str, size_t len) {
    uint64_t bit = (uint64_t)1 << (new_id & 63);
    uint32_t word = new_id >> 6;
    if (g_sym.scanned[word] & bit) return true;

    const char* first_dot = (const char*)memchr(str, '.', len);
    if (!first_dot) {
        /* Plain — mark settled. */
        g_sym.scanned[word] |= bit;
        return true;
    }

    /* Validate structure.  Trailing dot → not dotted.  Leading `.` is
     * allowed ONLY when followed by another dot (e.g. `.sys.gc`) —
     * in that case segment 0 includes the leading dot (`.sys`), so
     * reserved-namespace names resolve against their root dict via
     * the regular segment walk. */
    if (str[len - 1] == '.') {
        g_sym.scanned[word] |= bit;
        return true;
    }
    bool leading_dot = (str[0] == '.');
    if (leading_dot) {
        /* `.sys` alone (no second dot) is a plain name. */
        const char* second = (const char*)memchr(str + 1, '.', len - 1);
        if (!second) { g_sym.scanned[word] |= bit; return true; }
    }
    size_t sep_dots = 0;
    for (size_t i = (leading_dot ? 1 : 0); i < len; i++)
        if (str[i] == '.') sep_dots++;
    if (sep_dots + 1 > 255) {
        g_sym.scanned[word] |= bit;
        return true;
    }
    uint8_t nsegs = (uint8_t)(sep_dots + 1);

    struct { const char* p; size_t len; uint32_t hash; int64_t id; } descs[256];
    uint32_t new_seg_count = 0;
    size_t   new_seg_bytes = 0;
    {
        const char* p = str;
        size_t remaining = len;
        uint8_t i = 0;
        while (remaining && i < nsegs) {
            /* Segment 0 starts at str[0] but skips the leading `.` when
             * searching for the segment-terminating dot — so seg 0 of
             * `.sys.gc` is `.sys`, not `` (empty). */
            size_t skip = (i == 0 && leading_dot) ? 1 : 0;
            const char* dot = remaining > skip
                ? (const char*)memchr(p + skip, '.', remaining - skip)
                : NULL;
            size_t seg_len = dot ? (size_t)(dot - p) : remaining;
            if (seg_len == 0) { g_sym.scanned[word] |= bit; return true; }
            uint32_t h = (uint32_t)ray_hash_bytes(p, seg_len);
            descs[i].p    = p;
            descs[i].len  = seg_len;
            descs[i].hash = h;
            descs[i].id   = sym_probe(h, p, seg_len);
            if (descs[i].id < 0) {
                new_seg_count++;
                new_seg_bytes += ray_sym_bytes_upper(seg_len);
            }
            i++;
            if (!dot) break;
            remaining -= (seg_len + 1);
            p = dot + 1;
        }
    }

    /* Reserve capacity for new segments + segs array. */
    size_t segs_payload = (size_t)nsegs * sizeof(int64_t);
    size_t arena_bytes  = new_seg_bytes +
                          (((size_t)32 + segs_payload + 31) & ~(size_t)31);
    if (!sym_reserve_capacity(new_seg_count, arena_bytes)) return false;

    /* Commit.  Allocations covered by reservation above. */
    for (uint8_t i = 0; i < nsegs; i++) {
        if (descs[i].id < 0) {
            int64_t sid = sym_commit_new(descs[i].hash, descs[i].p, descs[i].len);
            if (sid < 0) return false;   /* reservation should have prevented */
            descs[i].id = sid;
            g_sym.scanned[sid >> 6] |= ((uint64_t)1 << (sid & 63));
        }
    }

    int64_t* segs = (int64_t*)ray_arena_alloc(g_sym.arena, segs_payload);
    if (!segs) return false;             /* reservation should have prevented */
    for (uint8_t i = 0; i < nsegs; i++) segs[i] = descs[i].id;

    g_sym.segments[new_id].nsegs = nsegs;
    g_sym.segments[new_id].segs  = segs;
    g_sym.dotted[word]  |= bit;
    g_sym.scanned[word] |= bit;
    return true;
}

/* --------------------------------------------------------------------------
 * sym_probe — hash-table lookup only.  Returns sym_id for an existing
 * entry or -1 if not present.  No side effects.
 * -------------------------------------------------------------------------- */
static int64_t sym_probe(uint32_t hash, const char* str, size_t len) {
    uint32_t mask = g_sym.bucket_cap - 1;
    uint32_t slot = hash & mask;
    for (;;) {
        uint64_t e = g_sym.buckets[slot];
        if (e == 0) return -1;
        uint32_t e_hash = (uint32_t)(e >> 32);
        if (e_hash == hash) {
            uint32_t e_id = (uint32_t)(e & 0xFFFFFFFF) - 1;
            ray_t* existing = g_sym.strings[e_id];
            if (ray_str_len(existing) == len &&
                memcmp(ray_str_ptr(existing), str, len) == 0) {
                return (int64_t)e_id;
            }
        }
        slot = (slot + 1) & mask;
    }
}

/* --------------------------------------------------------------------------
 * sym_commit_new — insert a NEW sym (caller must have confirmed it does
 * not already exist).  Grows the hash/strings tables as needed, allocates
 * the string atom from the arena, inserts into the hash table.  Returns
 * new sym_id or -1 on OOM.  No cache side effect.
 * -------------------------------------------------------------------------- */
static int64_t sym_commit_new(uint32_t hash, const char* str, size_t len) {
    /* Grow hash table if load factor exceeds threshold, or if critically
     * full.  Attempt grow before refusing insert.
     * Cast to uint64_t to prevent overflow when bucket_cap >= 2^26. */
    if ((uint64_t)g_sym.str_count * 100 >= (uint64_t)g_sym.bucket_cap * 70) {
        if (!ht_grow()) {
            /* If critically full even after failed grow, refuse insert
             * to prevent infinite probe loops. */
            if ((uint64_t)g_sym.str_count * 100 >= (uint64_t)g_sym.bucket_cap * 95) {
                return -1;
            }
        }
    }

    uint32_t new_id = g_sym.str_count;

    if (new_id >= g_sym.str_cap) {
        if (g_sym.str_cap >= UINT32_MAX / 2) return -1;
        if (!sym_grow_str_cap(g_sym.str_cap * 2)) return -1;
    }

    /* Create string atom from arena — avoids buddy allocator overhead.
     * Arena blocks have rc=1 and RAY_ATTR_ARENA set. */
    ray_t* s = ray_arena_str(g_sym.arena, str, len);
    if (!s) return -1;
    g_sym.strings[new_id] = s;
    g_sym.str_count++;

    /* Insert into hash table.
     * Note: ht_insert probes from hash & mask to find an empty slot,
     * so it works correctly even if ht_grow changed the bucket array. */
    ht_insert(g_sym.buckets, g_sym.bucket_cap, hash, new_id);

    return (int64_t)new_id;
}

/* --------------------------------------------------------------------------
 * sym_intern_nolock_noseg — intern WITHOUT the segment-caching side
 * effect.  The snapshot-restore path (ray_sym_load) uses this variant
 * because segment sub-interning during load would append new ids
 * mid-sequence and break the disk-position==sym_id invariant.  After
 * the bulk op, call ray_sym_rebuild_segments to populate the dotted
 * bitmap + segments cache.  Assumes caller holds sym_lock (or is in
 * the single-threaded prehashed caller contract).
 * -------------------------------------------------------------------------- */
static int64_t sym_intern_nolock_noseg(uint32_t hash, const char* str, size_t len) {
    int64_t existing = sym_probe(hash, str, len);
    if (existing >= 0) return existing;
    if (g_sym.lazy_map && g_sym.lazy_next_id < g_sym.lazy_count) {
        if (!sym_lazy_materialize_to_locked(g_sym.lazy_count - 1))
            return -1;
        existing = sym_probe(hash, str, len);
        if (existing >= 0) return existing;
    }
    return sym_commit_new(hash, str, len);
}

/* Reserve hash-table, strings-array, and arena capacity for `new_sym_count`
 * new syms plus `arena_bytes` of additional arena usage (for the segs array
 * if we're interning a dotted name).  Returns true on success; on failure
 * returns false with no commits made. */
static bool sym_reserve_capacity(uint32_t new_sym_count, size_t arena_bytes) {
    /* Hash table — grow if adding new_sym_count entries would exceed 70%. */
    uint64_t new_count = (uint64_t)g_sym.str_count + new_sym_count;
    uint32_t target = g_sym.bucket_cap;
    while (new_count * 100 >= (uint64_t)target * 70) {
        if (target >= (UINT32_MAX / 2 + 1)) return false;
        target *= 2;
    }
    if (target > g_sym.bucket_cap) {
        if (!ht_grow_to(target)) return false;
    }

    /* Strings and sidecars. */
    if (new_count > g_sym.str_cap) {
        uint32_t str_target = g_sym.str_cap;
        while (str_target < new_count) {
            if (str_target >= UINT32_MAX / 2) return false;
            str_target *= 2;
        }
        if (!sym_grow_str_cap(str_target)) return false;
    }

    /* Arena — reserve one chunk large enough for every forthcoming alloc. */
    if (arena_bytes && !ray_arena_reserve(g_sym.arena, arena_bytes)) return false;

    return true;
}

static void sym_lazy_unmap_locked(void) {
    if (!g_sym.lazy_map) return;
    ray_vm_unmap_file(g_sym.lazy_map, g_sym.lazy_size);
    g_sym.lazy_map = NULL;
    g_sym.lazy_size = 0;
    g_sym.lazy_count = 0;
    g_sym.lazy_next_id = 0;
    g_sym.lazy_ptr = NULL;
    g_sym.lazy_remaining = 0;
}

static bool sym_lazy_materialize_to_locked(uint32_t target_id) {
    if (!g_sym.lazy_map) return false;
    if (target_id >= g_sym.lazy_count) return false;
    if (target_id < g_sym.lazy_next_id) return g_sym.strings[target_id] != NULL;

    while (g_sym.lazy_next_id <= target_id) {
        if (g_sym.lazy_remaining < 4) return false;
        uint32_t slen;
        memcpy(&slen, g_sym.lazy_ptr, 4);
        g_sym.lazy_ptr += 4;
        g_sym.lazy_remaining -= 4;
        if ((size_t)slen > g_sym.lazy_remaining) return false;

        uint32_t id = g_sym.lazy_next_id;
        const char* sp = (const char*)g_sym.lazy_ptr;
        ray_t* existing = g_sym.strings[id];
        if (existing) {
            if (ray_str_len(existing) != (size_t)slen ||
                memcmp(ray_str_ptr(existing), sp, slen) != 0)
                return false;
        } else {
            ray_t* s = ray_arena_str(g_sym.arena, sp, (size_t)slen);
            if (!s) return false;
            g_sym.strings[id] = s;
            ht_insert(g_sym.buckets, g_sym.bucket_cap,
                      (uint32_t)ray_hash_bytes(sp, (size_t)slen), id);
        }

        g_sym.lazy_ptr += slen;
        g_sym.lazy_remaining -= slen;
        g_sym.lazy_next_id++;
    }
    return true;
}

/* --------------------------------------------------------------------------
 * sym_intern_nolock — fully atomic intern.
 *
 * Three phases:
 *   A. Inspect: probe the main name, validate its dotted shape, probe
 *      every segment.  No side effects.
 *   B. Reserve: pre-grow hash/strings/arena to accommodate everything
 *      we might need to commit.  Can fail → return -1 with no state
 *      change (no orphan segment syms, no cache fragments).
 *   C. Commit: all allocations in this phase are guaranteed by the
 *      reservations above, so they cannot fail.  Creates any new
 *      segment syms, creates the main sym, fills the segs cache, sets
 *      scanned + dotted bits.
 *
 * This closes two prior traps:
 *  - A committed main sym whose dotted bit disagrees with its name's
 *    structure (env silently routing dotted-path writes/reads through
 *    the flat path).
 *  - Orphan segment syms persisting when the main-sym commit fails.
 *
 * For an existing sym found in phase A, we still opportunistically try
 * the cache — that path is the lazy fallback for ray_sym_intern_no_split,
 * which commits the main sym without a cache on purpose.  A cache-OOM
 * there is tolerated (scanned bit stays clear → future interns retry).
 * -------------------------------------------------------------------------- */
static int64_t sym_intern_nolock(uint32_t hash, const char* str, size_t len,
                                 bool search_lazy) {
    /* Phase A.1: probe main. */
    int64_t existing = sym_probe(hash, str, len);
    if (search_lazy && existing < 0 && g_sym.lazy_map &&
        g_sym.lazy_next_id < g_sym.lazy_count) {
        if (!sym_lazy_materialize_to_locked(g_sym.lazy_count - 1))
            return -1;
        existing = sym_probe(hash, str, len);
    }
    if (existing >= 0) {
        (void)sym_cache_segments((uint32_t)existing, str, len);
        return existing;
    }

    /* Phase A.2: structural validation + per-segment probe. */
    struct { const char* p; size_t len; uint32_t hash; int64_t id; } descs[256];
    uint8_t  nsegs = 0;
    uint32_t new_seg_count = 0;
    size_t   new_seg_bytes = 0;
    bool     is_dotted = false;

    const char* first_dot = (const char*)memchr(str, '.', len);
    if (first_dot) {
        /* Dotted-name rules (parallel to sym_cache_segments):
         *   - Trailing dot            → plain (not dotted).
         *   - Leading dot alone       → plain (`.sys` with no inner dot).
         *   - Leading dot + inner dot → segment 0 is `.<head>` including
         *                                the leading dot.  This is how
         *                                reserved-namespace names like
         *                                `.sys.gc` resolve against the
         *                                `.sys` root dict. */
        bool valid = str[len - 1] != '.';
        bool leading_dot = (str[0] == '.');
        if (valid && leading_dot) {
            const char* second = (const char*)memchr(str + 1, '.', len - 1);
            if (!second) valid = false;
        }
        size_t sep_dots = 0;
        if (valid) {
            for (size_t i = (leading_dot ? 1 : 0); i < len; i++)
                if (str[i] == '.') sep_dots++;
            if (sep_dots + 1 > 255) valid = false;
        }
        if (valid) {
            nsegs = (uint8_t)(sep_dots + 1);
            const char* p = str;
            size_t remaining = len;
            uint8_t i = 0;
            while (remaining && i < nsegs) {
                size_t skip = (i == 0 && leading_dot) ? 1 : 0;
                const char* dot = remaining > skip
                    ? (const char*)memchr(p + skip, '.', remaining - skip)
                    : NULL;
                size_t seg_len = dot ? (size_t)(dot - p) : remaining;
                if (seg_len == 0) { valid = false; break; }
                uint32_t seg_hash = (uint32_t)ray_hash_bytes(p, seg_len);
                descs[i].p    = p;
                descs[i].len  = seg_len;
                descs[i].hash = seg_hash;
                descs[i].id   = sym_probe(seg_hash, p, seg_len);
                if (descs[i].id < 0) {
                    new_seg_count++;
                    new_seg_bytes += ray_sym_bytes_upper(seg_len);
                }
                i++;
                if (!dot) break;
                remaining -= (seg_len + 1);
                p = dot + 1;
            }
            if (valid) is_dotted = true;
        }
    }

    /* Phase B: reserve capacity for main + new segments + segs array. */
    size_t arena_bytes = ray_sym_bytes_upper(len);
    if (is_dotted) {
        arena_bytes += new_seg_bytes;
        /* segs array is arena-allocated via ray_arena_alloc(_, nsegs*8). */
        size_t segs_payload = (size_t)nsegs * sizeof(int64_t);
        arena_bytes += ((size_t)32 + segs_payload + 31) & ~(size_t)31;
    }
    if (!sym_reserve_capacity(1 + new_seg_count, arena_bytes)) return -1;

    /* Phase C: commit.  Every allocation below is covered by the
     * reservation above, so nothing here can fail. */
    if (is_dotted) {
        for (uint8_t i = 0; i < nsegs; i++) {
            if (descs[i].id < 0) {
                int64_t sid = sym_commit_new(descs[i].hash, descs[i].p, descs[i].len);
                /* Reservation guarantees success; defensive check kept. */
                if (sid < 0) return -1;
                descs[i].id = sid;
                /* Segment is itself a plain name (no dot inside). */
                g_sym.scanned[sid >> 6] |= ((uint64_t)1 << (sid & 63));
            }
        }
    }

    int64_t main_id = sym_commit_new(hash, str, len);
    if (main_id < 0) return -1;

    if (is_dotted) {
        int64_t* segs = (int64_t*)ray_arena_alloc(g_sym.arena,
                                                  (size_t)nsegs * sizeof(int64_t));
        if (!segs) return main_id;   /* reservation should have prevented this */
        for (uint8_t i = 0; i < nsegs; i++) segs[i] = descs[i].id;
        g_sym.segments[main_id].nsegs = nsegs;
        g_sym.segments[main_id].segs  = segs;
        g_sym.dotted[main_id >> 6] |= ((uint64_t)1 << (main_id & 63));
    }
    g_sym.scanned[main_id >> 6] |= ((uint64_t)1 << (main_id & 63));

    return main_id;
}

/* --------------------------------------------------------------------------
 * ray_sym_intern — locked public API
 * -------------------------------------------------------------------------- */

int64_t ray_sym_intern(const char* str, size_t len) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;
    uint32_t hash = (uint32_t)ray_hash_bytes(str, len);
    sym_lock();
    int64_t id = sym_intern_nolock(hash, str, len, true);
    sym_unlock();
    return id;
}

int64_t ray_sym_intern_runtime(const char* str, size_t len) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;
    uint32_t hash = (uint32_t)ray_hash_bytes(str, len);
    sym_lock();
    int64_t id = sym_intern_nolock(hash, str, len, false);
    sym_unlock();
    return id;
}

/* --------------------------------------------------------------------------
 * ray_sym_intern_prehashed -- intern with pre-computed hash, no lock.
 *
 * CALLER CONTRACT: must only be called when no other thread is interning
 * (e.g., after ray_pool_dispatch returns during CSV merge).
 * -------------------------------------------------------------------------- */

int64_t ray_sym_intern_prehashed(uint32_t hash, const char* str, size_t len) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;
    return sym_intern_nolock(hash, str, len, true);
}

/* --------------------------------------------------------------------------
 * ray_sym_intern_batch -- intern n pre-hashed strings under one lock.
 *
 * For decoders that have already deduplicated their input: each distinct
 * string costs one probe, and the whole batch one lock round-trip.  The
 * per-string work is exactly ray_sym_intern's (search_lazy = true), so ids
 * and dotted-segment caching are identical.
 * Returns 0, or -1 if any intern failed (out_ids then partially filled).
 * -------------------------------------------------------------------------- */

int64_t ray_sym_intern_batch(const uint32_t* hashes, const char* const* strs,
                             const size_t* lens, int64_t n, int64_t* out_ids) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;
    if (n <= 0) return 0;
    sym_lock();
    for (int64_t i = 0; i < n; i++) {
        int64_t id = sym_intern_nolock(hashes[i], strs[i], lens[i], true);
        if (id < 0) { sym_unlock(); return -1; }
        out_ids[i] = id;
    }
    sym_unlock();
    return 0;
}

/* --------------------------------------------------------------------------
 * ray_sym_intern_no_split — persistence-only bulk intern
 * -------------------------------------------------------------------------- */

int64_t ray_sym_intern_no_split(const char* str, size_t len) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;
    uint32_t hash = (uint32_t)ray_hash_bytes(str, len);
    sym_lock();
    int64_t id = sym_intern_nolock_noseg(hash, str, len);
    sym_unlock();
    return id;
}

int64_t ray_sym_intern_no_split_unlocked(const char* str, size_t len) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;
    uint32_t hash = (uint32_t)ray_hash_bytes(str, len);
    return sym_intern_nolock_noseg(hash, str, len);
}

/* --------------------------------------------------------------------------
 * ray_sym_rebuild_segments — populate dotted cache for any not-yet-cached
 * entries.  Must follow a batch of ray_sym_intern_no_split calls.
 *
 * Propagates the first allocation/sub-intern failure as RAY_ERR_OOM so
 * persistence callers (ray_sym_load) can abort cleanly rather than
 * silently leaving dotted names un-cached — that
 * would degrade them to flat-sym semantics and break env lookup for any
 * name the user wrote with a '.' in it.
 * -------------------------------------------------------------------------- */

ray_err_t ray_sym_rebuild_segments(void) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return RAY_ERR_IO;
    sym_lock();
    /* Snapshot upper bound — sym_cache_segments may append segment entries
     * beyond the original range, but those new entries themselves are
     * non-dotted segment names and so produce no further work.  Use the
     * scanned bitmap to skip: anything already settled (plain or dotted)
     * avoids even the memchr inside sym_cache_segments. */
    uint32_t count = g_sym.str_count;
    for (uint32_t i = 0; i < count; i++) {
        if (g_sym.scanned[i >> 6] & ((uint64_t)1 << (i & 63))) continue;
        ray_t* s = g_sym.strings[i];
        if (!s) continue;
        if (!sym_cache_segments(i, ray_str_ptr(s), ray_str_len(s))) {
            sym_unlock();
            return RAY_ERR_OOM;
        }
    }
    sym_unlock();
    return RAY_OK;
}

/* --------------------------------------------------------------------------
 * Dotted-name accessors
 * -------------------------------------------------------------------------- */

bool ray_sym_is_dotted(int64_t sym_id) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return false;
    if (sym_id < 0 || (uint32_t)sym_id >= g_sym.str_count) return false;
    uint64_t word = g_sym.dotted[(uint32_t)sym_id >> 6];
    return (word >> ((uint32_t)sym_id & 63)) & 1;
}

int ray_sym_segs(int64_t sym_id, const int64_t** out_segs) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return 0;
    if (sym_id < 0 || (uint32_t)sym_id >= g_sym.str_count) return 0;
    sym_segs_t s = g_sym.segments[sym_id];
    if (s.nsegs == 0 || !s.segs) return 0;
    if (out_segs) *out_segs = s.segs;
    return (int)s.nsegs;
}

/* --------------------------------------------------------------------------
 * ray_sym_find
 * -------------------------------------------------------------------------- */

int64_t ray_sym_find(const char* str, size_t len) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return -1;

    /* Lock required: concurrent ray_sym_intern may trigger ht_grow which
     * frees and replaces g_sym.buckets -- reading without lock is UAF. */
    sym_lock();

    uint32_t hash = (uint32_t)ray_hash_bytes(str, len);
    uint32_t mask = g_sym.bucket_cap - 1;
    uint32_t slot = hash & mask;

    for (;;) {
        uint64_t e = g_sym.buckets[slot];
        if (e == 0) {
            if (g_sym.lazy_map && g_sym.lazy_next_id < g_sym.lazy_count) {
                if (sym_lazy_materialize_to_locked(g_sym.lazy_count - 1)) {
                    mask = g_sym.bucket_cap - 1;
                    slot = hash & mask;
                    continue;
                }
            }
            sym_unlock();
            return -1;
        }  /* empty -- not found */

        uint32_t e_hash = (uint32_t)(e >> 32);
        if (e_hash == hash) {
            uint32_t e_id = (uint32_t)(e & 0xFFFFFFFF) - 1;
            ray_t* existing = g_sym.strings[e_id];
            if (ray_str_len(existing) == len &&
                memcmp(ray_str_ptr(existing), str, len) == 0) {
                sym_unlock();
                return (int64_t)e_id;
            }
        }
        slot = (slot + 1) & mask;
    }
}

/* --------------------------------------------------------------------------
 * ray_sym_str
 * -------------------------------------------------------------------------- */

/* Returned pointer is valid only while no concurrent ray_sym_intern occurs.
 * Safe during read-only execution phase (after all interning is complete).
 * Caller must not store the pointer across sym table mutations (ht_grow
 * or strings realloc). */
ray_t* ray_sym_str(int64_t id) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return NULL;

    /* Lock required: concurrent ray_sym_intern may realloc g_sym.strings. */
    sym_lock();
    if (id < 0 || (uint32_t)id >= g_sym.str_count) { sym_unlock(); return NULL; }
    if (!g_sym.strings[id] && (uint32_t)id < g_sym.lazy_count) {
        if (!sym_lazy_materialize_to_locked((uint32_t)id)) {
            sym_unlock();
            return NULL;
        }
    }
    ray_t* s = g_sym.strings[id];
    sym_unlock();
    return s;
}

/* --------------------------------------------------------------------------
 * ray_sym_count
 * -------------------------------------------------------------------------- */

uint32_t ray_sym_count(void) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return 0;

    /* Lock required: concurrent ray_sym_intern may modify str_count. */
    sym_lock();
    uint32_t count = g_sym.str_count;
    sym_unlock();
    return count;
}

/* --------------------------------------------------------------------------
 * ray_sym_strings_borrow
 *
 * Single-shot snapshot of the sym→string table for hot read-only
 * scanners (LIKE, dictionary projection, …).  ray_sym_str takes a spin
 * lock per call; iterating all 1.7M URL dict entries via ray_sym_str
 * means 1.7M lock acquisitions.  This routine takes the lock once,
 * captures the array pointer + length, drops the lock, and lets the
 * caller iterate lock-free.
 *
 * Validity: only safe during read-only phases (no concurrent
 * ray_sym_intern).  ray_sym_intern can realloc g_sym.strings, after
 * which the returned pointer is dangling.  Today's pipeline is one
 * pass: bulk-intern at CSV load, then run queries against the frozen
 * table — exactly the contract this borrow form needs.
 * -------------------------------------------------------------------------- */
void ray_sym_strings_borrow(ray_t*** out_strings, uint32_t* out_count) {
    if (out_strings) *out_strings = NULL;
    if (out_count)   *out_count   = 0;
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return;
    sym_lock();
    if (g_sym.lazy_map && g_sym.lazy_count > 0) {
        (void)sym_lazy_materialize_to_locked(g_sym.lazy_count - 1);
    }
    if (out_strings) *out_strings = g_sym.strings;
    if (out_count)   *out_count   = g_sym.str_count;
    sym_unlock();
}

/* --------------------------------------------------------------------------
 * ray_sym_ensure_cap -- pre-grow hash table and strings array
 *
 * Ensures the symbol table can hold at least `needed` total symbols without
 * rehashing.  Call before bulk interning (e.g., CSV merge) to prevent
 * mid-insert OOM that silently drops symbols.
 * -------------------------------------------------------------------------- */

bool ray_sym_ensure_cap(uint32_t needed) {
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return false;

    sym_lock();

    /* Grow strings array (and sidecars) if needed */
    while (g_sym.str_cap < needed) {
        if (g_sym.str_cap >= UINT32_MAX / 2) { sym_unlock(); return false; }
        uint32_t new_str_cap = g_sym.str_cap * 2;
        if (new_str_cap < needed) { /* jump directly to needed */
            new_str_cap = needed;
            /* Round up to power of 2 */
            new_str_cap--;
            new_str_cap |= new_str_cap >> 1;
            new_str_cap |= new_str_cap >> 2;
            new_str_cap |= new_str_cap >> 4;
            new_str_cap |= new_str_cap >> 8;
            new_str_cap |= new_str_cap >> 16;
            new_str_cap++;
            if (new_str_cap == 0) { sym_unlock(); return false; }
        }
        if (!sym_grow_str_cap(new_str_cap)) { sym_unlock(); return false; }
    }

    /* Grow hash table so load factor stays below threshold after filling */
    double raw_buckets = (double)needed / SYM_LOAD_FACTOR + 1.0;
    if (raw_buckets > (double)UINT32_MAX) { sym_unlock(); return false; }
    uint32_t needed_buckets = (uint32_t)raw_buckets;
    /* Round up to power of 2 */
    needed_buckets--;
    needed_buckets |= needed_buckets >> 1;
    needed_buckets |= needed_buckets >> 2;
    needed_buckets |= needed_buckets >> 4;
    needed_buckets |= needed_buckets >> 8;
    needed_buckets |= needed_buckets >> 16;
    needed_buckets++;

    if (needed_buckets > g_sym.bucket_cap) {
        if (!ht_grow_to(needed_buckets)) { sym_unlock(); return false; }
    }

    sym_unlock();
    return true;
}

/* --------------------------------------------------------------------------
 * ray_sym_save -- snapshot the RUNTIME dictionary to a STRL file
 *
 * This persists the process-global intern table (the runtime domain) so an
 * embedder session can be resumed via ray_runtime_create_with_sym /
 * ray_sym_load.  It is NOT a table symfile: stored tables carry their own
 * per-vocabulary symfiles, owned by the FILE domain layer (table/domain.c).
 *
 * Snapshot semantics: the current table is written whole via tmp file +
 * fsync + atomic rename; whatever was at `path` is replaced (single-writer
 * contract — there is no merge with other writers' contents).
 * -------------------------------------------------------------------------- */

ray_err_t ray_sym_save(const char* path) {
    if (!path) return RAY_ERR_IO;
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return RAY_ERR_IO;

    /* Build lock and temp paths */
    char lock_path[1024];
    char tmp_path[1024];
    if (snprintf(lock_path, sizeof(lock_path), "%s.lk", path) >= (int)sizeof(lock_path))
        return RAY_ERR_IO;
    if (snprintf(tmp_path, sizeof(tmp_path), "%s.tmp", path) >= (int)sizeof(tmp_path))
        return RAY_ERR_IO;

    /* Acquire cross-process exclusive lock */
    ray_fd_t lock_fd = ray_file_open(lock_path, RAY_OPEN_READ | RAY_OPEN_WRITE | RAY_OPEN_CREATE);
    if (lock_fd == RAY_FD_INVALID) return RAY_ERR_IO;
    ray_err_t err = ray_file_lock_ex(lock_fd);
    if (err != RAY_OK) { ray_file_close(lock_fd); return err; }

    /* Snapshot string pointers under sym_lock, then build list without it.
     * Strings are append-only and never freed, so pointers remain valid.
     * A lazily-mapped dictionary must be fully materialized first — the
     * snapshot loop below needs a real string atom in every slot. */
    sym_lock();
    if (g_sym.lazy_map && g_sym.lazy_count > 0 &&
        !sym_lazy_materialize_to_locked(g_sym.lazy_count - 1)) {
        sym_unlock();
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_CORRUPT;
    }
    uint32_t count = g_sym.str_count;
    size_t snap_sz = count * sizeof(ray_t*);
    ray_t* snap_block = ray_alloc(snap_sz);
    if (!snap_block || RAY_IS_ERR(snap_block)) {
        sym_unlock();
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_OOM;
    }
    ray_t** snap = (ray_t**)ray_data(snap_block);
    memcpy(snap, g_sym.strings, snap_sz);
    sym_unlock();

    /* Save STRL directly instead of first materializing a giant RAY_LIST. */
    {
        FILE* f = fopen(tmp_path, "wb");
        if (!f) {
            ray_free(snap_block);
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_IO;
        }
        uint32_t magic = SYM_STRL_MAGIC;
        int64_t n64 = (int64_t)count;
        err = RAY_OK;
        if (fwrite(&magic, 4, 1, f) != 1 ||
            fwrite(&n64, 8, 1, f) != 1) {
            err = RAY_ERR_IO;
        } else {
            for (uint32_t i = 0; i < count; i++) {
                ray_t* s = snap[i];
                if (!s || s->type != -RAY_STR) { err = RAY_ERR_CORRUPT; break; }
                const char* sp = ray_str_ptr(s);
                size_t slen = ray_str_len(s);
                if (slen > UINT32_MAX) { err = RAY_ERR_RANGE; break; }
                uint32_t len32 = (uint32_t)slen;
                if (fwrite(&len32, 4, 1, f) != 1 ||
                    (slen > 0 && fwrite(sp, 1, slen, f) != slen)) {
                    err = RAY_ERR_IO;
                    break;
                }
            }
        }
        if (fclose(f) != 0 && err == RAY_OK) err = RAY_ERR_IO;
    }
    ray_free(snap_block);
    if (err != RAY_OK) {
        remove(tmp_path);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return err;
    }

    /* Fsync temp file for durability */
    ray_fd_t tmp_fd = ray_file_open(tmp_path, RAY_OPEN_READ | RAY_OPEN_WRITE);
    if (tmp_fd == RAY_FD_INVALID) {
        remove(tmp_path);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_IO;
    }
    err = ray_file_sync(tmp_fd);
    ray_file_close(tmp_fd);
    if (err != RAY_OK) {
        remove(tmp_path);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return err;
    }

    /* Atomic rename: tmp -> final path */
    err = ray_file_rename(tmp_path, path);
    if (err != RAY_OK) {
        remove(tmp_path);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return err;
    }

    /* Fsync parent directory so the new directory entry is durable.
     * Without this, a crash after rename can lose the new file. */
    err = ray_file_sync_dir(path);
    if (err != RAY_OK) {
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return err;
    }

    ray_file_unlock(lock_fd);
    ray_file_close(lock_fd);
    return RAY_OK;
}

/* --------------------------------------------------------------------------
 * ray_sym_load -- restore a RUNTIME dictionary snapshot (STRL format)
 *
 * Counterpart of ray_sym_save: every disk entry must land at the in-memory
 * id equal to its file position, so symbol ids persisted by an earlier
 * session (env bindings, bare-saved SYM vectors) stay valid.  An entry that
 * cannot take its slot (a different symbol already interned there) is a
 * divergence -> RAY_ERR_CORRUPT.  File locking prevents reading a partial
 * write.  NOT a table symfile loader — that is ray_sym_domain_open.
 * -------------------------------------------------------------------------- */

ray_err_t ray_sym_load(const char* path) {
    if (!path) return RAY_ERR_IO;
    if (!atomic_load_explicit(&g_sym_inited, memory_order_acquire)) return RAY_ERR_IO;

    /* Acquire cross-process shared lock.
     * Try read-only open first so that read-only users (snapshots, read-only
     * mounts) can load without write permission on the directory.  Fall back
     * to read-write+create if the lock file doesn't exist yet.  If both fail,
     * only proceed without locking on read-only filesystem (EROFS) — other
     * errors (EMFILE, ENFILE, EACCES on writable fs, etc.) are real failures
     * that would silently drop the shared-lock guarantee. */
    char lock_path[1024];
    if (snprintf(lock_path, sizeof(lock_path), "%s.lk", path) >= (int)sizeof(lock_path))
        return RAY_ERR_IO;
    ray_fd_t lock_fd = ray_file_open(lock_path, RAY_OPEN_READ);
    if (lock_fd == RAY_FD_INVALID) {
        int saved_errno = errno;
        lock_fd = ray_file_open(lock_path, RAY_OPEN_READ | RAY_OPEN_WRITE | RAY_OPEN_CREATE);
        if (lock_fd == RAY_FD_INVALID) {
            /* Only proceed unlocked on read-only filesystem (EROFS) where
             * concurrent writes are impossible.  All other failures are
             * real errors that should not be silently ignored. */
            if (saved_errno != EROFS && errno != EROFS)
                return RAY_ERR_IO;
        }
    }
    if (lock_fd != RAY_FD_INVALID) {
        ray_err_t err = ray_file_lock_sh(lock_fd);
        if (err != RAY_OK) { ray_file_close(lock_fd); return err; }
    }

    size_t mapped_size = 0;
    uint8_t* mapped = (uint8_t*)ray_vm_map_file(path, &mapped_size);
    if (!mapped) {
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_IO;
    }
    if (mapped_size < 12) {
        ray_vm_unmap_file(mapped, mapped_size);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_CORRUPT;
    }
    uint32_t magic;
    memcpy(&magic, mapped, 4);
    if (magic != SYM_STRL_MAGIC) {
        ray_vm_unmap_file(mapped, mapped_size);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_CORRUPT;
    }

    int64_t disk_count;
    memcpy(&disk_count, mapped + 4, 8);
    if (disk_count < 0 || disk_count > UINT32_MAX) {
        ray_vm_unmap_file(mapped, mapped_size);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_CORRUPT;
    }

    if (mapped_size >= SYM_LAZY_LOAD_MIN_BYTES) {
        sym_lock();
        uint32_t current = g_sym.str_count;
        uint32_t disk_u = (uint32_t)disk_count;
        uint32_t target_count = current > disk_u ? current : disk_u;
        if (target_count > current && !sym_reserve_capacity(target_count - current, 0)) {
            sym_unlock();
            ray_vm_unmap_file(mapped, mapped_size);
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_OOM;
        }
        if (disk_u > current) {
            memset(g_sym.strings + current, 0,
                   ((size_t)disk_u - current) * sizeof(ray_t*));
        }

        sym_lazy_unmap_locked();
        g_sym.lazy_map = mapped;
        g_sym.lazy_size = mapped_size;
        g_sym.lazy_count = disk_u;
        g_sym.lazy_next_id = 0;
        g_sym.lazy_ptr = mapped + 12;
        g_sym.lazy_remaining = mapped_size - 12;
        g_sym.str_count = target_count;

        uint32_t validate_count = current < disk_u ? current : disk_u;
        bool ok = validate_count == 0 ||
                  sym_lazy_materialize_to_locked(validate_count - 1);
        sym_unlock();
        if (!ok) {
            sym_lock();
            sym_lazy_unmap_locked();
            sym_unlock();
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_CORRUPT;
        }

        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_OK;
    }

    /* Intern every disk entry at its file position.  Entries already in
     * memory at the right id are a no-op (probe hit); a position conflict
     * (a different symbol holds the slot) is a divergence.  A file with
     * FEWER entries than memory is fine: it is an older snapshot of the
     * same dictionary, and the prefix check below still validates it. */
    const uint8_t* ptr = mapped + 12;
    size_t remaining = mapped_size - 12;
    for (int64_t i = 0; i < disk_count; i++) {
        if (remaining < 4) {
            ray_vm_unmap_file(mapped, mapped_size);
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_CORRUPT;
        }
        uint32_t slen;
        memcpy(&slen, ptr, 4);
        ptr += 4;
        remaining -= 4;
        if ((size_t)slen > remaining) {
            ray_vm_unmap_file(mapped, mapped_size);
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_CORRUPT;
        }

        const char* sp = (const char*)ptr;
        /* Bulk load uses no-split interning so dotted names cannot append
         * segment symbols mid-stream and shift disk-position IDs. */
        int64_t id = ray_sym_intern_no_split(sp, (size_t)slen);
        if (id < 0) {
            ray_vm_unmap_file(mapped, mapped_size);
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_OOM;
        }
        if (id != i) {
            ray_vm_unmap_file(mapped, mapped_size);
            ray_file_unlock(lock_fd);
            ray_file_close(lock_fd);
            return RAY_ERR_CORRUPT;
        }
        ptr += slen;
        remaining -= slen;
    }
    if (remaining != 0) {
        ray_vm_unmap_file(mapped, mapped_size);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return RAY_ERR_CORRUPT;
    }

    /* Populate dotted cache for every loaded (and previously-loaded) sym.
     * Idempotent — already-cached entries are skipped.  Runs once per load.
     * An OOM here must surface: leaving dotted names un-cached would make
     * env lookup silently resolve them as flat syms, quietly losing
     * namespace semantics on anything the user stored with a '.' in it. */
    ray_err_t rebuild_err = ray_sym_rebuild_segments();
    if (rebuild_err != RAY_OK) {
        ray_vm_unmap_file(mapped, mapped_size);
        ray_file_unlock(lock_fd);
        ray_file_close(lock_fd);
        return rebuild_err;
    }

    ray_vm_unmap_file(mapped, mapped_size);
    ray_file_unlock(lock_fd);
    ray_file_close(lock_fd);
    return RAY_OK;
}