fix(ipc): bind a VM for lifecycle hooks on VM-less threads

src/core/ipc.c

@@ -314,6 +314,56 @@ static ray_t* hook_lookup(int idx) {
     return fn;
 }
 
+/* ── Temporary VM for a hook on a VM-less thread (#569) ──────────────
+ *
+ * `__VM` is bound only by ray_runtime_create, so any thread an embedder
+ * drives IPC from — the FFI teardown path in particular — has none.  #565
+ * made the *no hook installed* case safe by letting ray_env_get fall
+ * through to globals, which left this question open: with a hook actually
+ * bound, call_fn1 would run user code with __VM == NULL.
+ *
+ * A hook is user code and should behave the same wherever the event came
+ * from, so bind a VM for the duration rather than skipping the hook
+ * (silent, and invisible at the call site) or failing the close (which
+ * would break the FFI teardown that runtime.c:50-54 explicitly supports).
+ *
+ * Binding also restores `.ipc.handle` inside the hook: ipc_ctx_set stores
+ * through __VM and returns early without one, so on a VM-less thread the
+ * hook would otherwise see a handle of -1.
+ *
+ * ray_vm_t is ~68 KB, too large for a teardown-path stack, so it comes
+ * from the buddy heap (process-wide and already live — the heap belongs to
+ * the runtime, not the thread).  Teardown mirrors ray_runtime_destroy:
+ * release raise_val and trace, then free.  Grown scope frames are not
+ * walked, matching that same teardown — a hook that returns normally has
+ * already popped its scopes. */
+typedef struct { ray_vm_t* owned; } hook_vm_t;
+
+static bool hook_vm_bind(hook_vm_t* hv) {
+    hv->owned = NULL;
+    if (__VM) return true;                      /* already on a VM thread */
+    ray_vm_t* vm = (ray_vm_t*)ray_alloc_raw(sizeof(ray_vm_t));
+    if (!vm) {
+        /* Skipping a hook the operator installed is worth a line, matching
+         * how this file reports every other hook failure. */
+        fprintf(stderr, "ipc: cannot bind a VM for hook dispatch (out of memory); hook skipped\n");
+        return false;
+    }
+    ray_vm_init(vm, -1);                        /* id -1: not a pool VM */
+    __VM = vm;
+    hv->owned = vm;
+    return true;
+}
+
+static void hook_vm_unbind(hook_vm_t* hv) {
+    if (!hv->owned) return;
+    if (hv->owned->raise_val) ray_release(hv->owned->raise_val);
+    if (hv->owned->trace) ray_release(hv->owned->trace);
+    __VM = NULL;
+    ray_free_raw(hv->owned);
+    hv->owned = NULL;
+}
+
 /* Call a single-arg hook for lifecycle events (on.open / on.close).
  * Errors are logged and swallowed — a buggy logging hook must never
  * wedge connection teardown.  `poll` is the poll the connection lives
@@ -322,8 +372,14 @@ static ray_t* hook_lookup(int idx) {
 static void hook_call_lifecycle(ray_poll_t* poll, int idx, int64_t handle) {
     ray_t* fn = hook_lookup(idx);
     if (!fn) return;
+    hook_vm_t hv;
+    if (!hook_vm_bind(&hv)) return;
     ray_t* arg = make_i64(handle);
-    if (!arg || RAY_IS_ERR(arg)) { if (arg) ray_release(arg); return; }
+    if (!arg || RAY_IS_ERR(arg)) {
+        if (arg) ray_release(arg);
+        hook_vm_unbind(&hv);
+        return;
+    }
     int64_t prev = ipc_ctx_handle();
     ray_poll_t* prev_poll = ipc_ctx_poll();
     ipc_ctx_set(handle, poll ? poll : prev_poll);
@@ -336,6 +392,7 @@ static void hook_call_lifecycle(ray_poll_t* poll, int idx, int64_t handle) {
     }
     ray_release(arg);
     if (r && r != RAY_NULL_OBJ) ray_release(r);
+    hook_vm_unbind(&hv);
 }
 
 /* Call the on.auth hook with (user, pass) string atoms.  Returns:
@@ -344,6 +401,10 @@ static void hook_call_lifecycle(ray_poll_t* poll, int idx, int64_t handle) {
  *  - -1 → no hook installed; caller uses the existing pass-through.
  * The constant-time secret compare in validate_creds always runs first,
  * so this hook can only narrow access — never widen it. */
+/* Not VM-bound, unlike hook_call_lifecycle: this runs only from inside
+ * ray_poll_run, on a thread that necessarily has a VM.  If a future path
+ * ever drives the handshake from an embedder's own thread, it needs the
+ * same hook_vm_bind treatment (#569). */
 static int hook_call_auth(ray_poll_t* poll, int64_t handle,
                           const uint8_t* cred_buf, uint8_t cred_len) {
     ray_t* fn = hook_lookup(IPC_HOOK_AUTH);
@@ -596,6 +657,8 @@ static ray_t* eval_payload_core(uint8_t* payload, size_t payload_len,
          * as `{[m] eval m}` reproduces the default behaviour. */
         int hook_idx = (hdr->msgtype == RAY_IPC_MSG_SYNC) ? IPC_HOOK_SYNC
                                                            : IPC_HOOK_ASYNC;
+        /* Also not VM-bound — eval_payload runs under ray_poll_run, which
+         * owns a VM.  See hook_call_lifecycle for the case that does not. */
         ray_t* hook = hook_lookup(hook_idx);
         if (hook) {
             result = call_fn1(hook, msg);

test/test_ipc.c

@@ -672,6 +672,78 @@ static test_result_t test_ipc_send_async_invalid_handle(void) {
  * Covers ipc_accept, ipc_read_handshake (success path),
  * ipc_read_header, ipc_read_payload, ipc_on_close, ipc_send_fn.
  */
+/* A lifecycle hook must run on a thread that never bound a VM (#569).
+ *
+ * #565 made the *no hook installed* case safe by letting ray_env_get fall
+ * through to globals on such a thread.  With a hook actually bound,
+ * call_fn1 would then run user code with __VM == NULL — this asserts the
+ * hook runs and observes a live `.ipc.handle`.
+ *
+ * The worker deliberately never calls ray_runtime_create: that is the
+ * shape of an embedder's own thread driving teardown, which is how this
+ * was found. */
+static ray_poll_t* g_vmless_poll = NULL;
+static int64_t     g_vmless_sel  = -1;
+
+static void vmless_close_worker(void* unused) {
+    (void)unused;
+    /* No ray_runtime_create here — __VM is NULL on this thread. */
+    ray_poll_deregister(g_vmless_poll, g_vmless_sel);
+}
+
+static test_result_t test_ipc_close_hook_on_vmless_thread(void) {
+    ray_test_server_t srv;
+    RAY_TEST_SERVER_START(srv);
+
+    int64_t h = ray_ipc_connect("127.0.0.1", srv.port, NULL, NULL, 2000);
+    TEST_ASSERT((h) >= (0), "connect");
+
+    /* Stop the server *before* installing the hook.
+     *
+     * .ipc.on.close is one process-global binding and ipc_on_close fires it
+     * for both ends of a connection, so a server thread — which has a VM —
+     * would otherwise satisfy any counter this test could assert on, and the
+     * test would stay green with the fix reverted.  Shutting the server down
+     * first, with no hook installed, means the single firing below is
+     * unambiguously the client-side teardown driven from the VM-less thread.
+     * It also removes the poll loop that would otherwise be writing these
+     * globals while the main thread reads them. */
+    ray_test_server_stop(&srv);
+
+    ray_t* r = ray_eval_str(
+        "(do (set _vmless_fired 0) (set _vmless_h -99)"
+        "    (set .ipc.on.close (fn [x] (do (set _vmless_fired (+ _vmless_fired 1))"
+        "                                   (set _vmless_h (.ipc.handle))))) null)");
+    TEST_ASSERT(r && !RAY_IS_ERR(r), "install hook");
+    ray_release(r);
+
+    g_vmless_poll = ray_ipc_active_poll();
+    TEST_ASSERT_NOT_NULL(g_vmless_poll);
+    g_vmless_sel  = h;
+
+    ray_thread_t tid;
+    ray_thread_create(&tid, vmless_close_worker, NULL);
+    ray_thread_join(tid);
+
+    /* Exactly one firing, and it came from the VM-less thread. */
+    ray_t* fired = ray_eval_str("_vmless_fired");
+    TEST_ASSERT(fired && !RAY_IS_ERR(fired), "read counter");
+    TEST_ASSERT_EQ_I(fired->i64, 1);
+    ray_release(fired);
+
+    /* It saw this connection's handle.  Without a bound VM ipc_ctx_set
+     * stores nothing and .ipc.handle reports -1, so this is the assertion
+     * that actually separates fixed from unfixed. */
+    ray_t* seen = ray_eval_str("_vmless_h");
+    TEST_ASSERT(seen && !RAY_IS_ERR(seen), "read handle");
+    TEST_ASSERT_EQ_I(seen->i64, h);
+    ray_release(seen);
+
+    ray_t* cleanup = ray_eval_str("(do (set .ipc.on.close null) null)");
+    if (cleanup) ray_release(cleanup);
+    PASS();
+}
+
 static test_result_t test_ipc_poll_based_listen(void) {
     ray_poll_t* poll = ray_poll_create();
     TEST_ASSERT_NOT_NULL(poll);
@@ -2643,6 +2715,7 @@ const test_entry_t ipc_entries[] = {
     { "ipc/close_invalid_handle",       test_ipc_close_invalid_handle,           ipc_setup, ipc_teardown },
     { "ipc/send_invalid_handle",        test_ipc_send_invalid_handle,            ipc_setup, ipc_teardown },
     { "ipc/send_async_invalid_handle",  test_ipc_send_async_invalid_handle,      ipc_setup, ipc_teardown },
+    { "ipc/close_hook_vmless_thread",   test_ipc_close_hook_on_vmless_thread,     ipc_setup, ipc_teardown },
     { "ipc/poll_based_listen",          test_ipc_poll_based_listen,              ipc_setup, ipc_teardown },
     { "ipc/poll_public_restricted",     test_ipc_poll_public_restricted,         ipc_setup, ipc_teardown },
     { "ipc/poll_auth_creds_path",        test_ipc_poll_auth_creds_path,           ipc_setup, ipc_teardown },