compiler-codegen/nova_rt/driver.c - main - nv-lang/nova

compiler-codegen/nova_rt/driver.c
609 строк28 KB
Evgeniy Golovin
fix(259): two Layer-2-exposed startup races found via regression sweep
09 авг 2026, 17:55
09 авг 2026, 17:5576dd140
// SPDX-License-Identifier: MIT OR Apache-2.0
/* Plan 83.11 Ф.2: Driver scaffolding. Lifecycle + job queue + main loop.
 *
 * Vela — M:N-движок конкурентности Nova; этот файл — I/O-драйвер (libuv
 * job-queue). Бренд-имя рантайма — docs/dev/naming-conventions.md §1.2, план 224
 * (идентификаторы/ABI не переименованы).
 *
 * NO logic yet — jobs are stubbed (logged but not processed). Ф.3 migrates
 * Time.sleep to use ARM_SLEEP/CANCEL_SCOPE jobs. Ф.4 adds blocking. Etc.
 *
 * Tokio reference: tokio/src/runtime/driver.rs */

#include "nova_rt.h"   /* full chain — needs NovaSleepState fields + nova_sched_wake */
#include "driver.h"
#include "runtime.h"   /* nova_runtime_signal_main if needed */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef NOVA_GC_BOEHM
#include <gc.h>
#endif

NovaDriver _nova_driver = {0};

/* ── Forward declarations (file-private) ─────────────────────────── */
static void _nova_driver_main(void* arg);
static void _nova_driver_job_async_cb(uv_async_t* h);
static void _nova_driver_shutdown_async_cb(uv_async_t* h);
static void _nova_driver_drain_jobs(void);
static void _nova_driver_process_job(NovaDriverJob* job);

/* ── Public API ──────────────────────────────────────────────────── */

void nova_driver_init(void) {
    /* Idempotent guard. */
    if (nova_abool_load(&_nova_driver.started)) return;

    /* Init job queue. */
    nova_mutex_init(&_nova_driver.jobs.mu);
    _nova_driver.jobs.head = NULL;
    _nova_driver.jobs.tail = NULL;

    nova_abool_init(&_nova_driver.stop, false);

    /* Init UV loop. */
    int rc = uv_loop_init(&_nova_driver.loop);
    if (rc != 0) {
        fprintf(stderr, "nova: driver uv_loop_init failed: %s\n", uv_strerror(rc));
        abort();
    }

    /* Init async handles BEFORE thread creation — thread will call uv_run
     * which requires handles to exist. */
    rc = uv_async_init(&_nova_driver.loop, &_nova_driver.job_async,
                      _nova_driver_job_async_cb);
    if (rc != 0) {
        fprintf(stderr, "nova: driver job_async init failed: %s\n", uv_strerror(rc));
        abort();
    }
    rc = uv_async_init(&_nova_driver.loop, &_nova_driver.shutdown_async,
                      _nova_driver_shutdown_async_cb);
    if (rc != 0) {
        fprintf(stderr, "nova: driver shutdown_async init failed: %s\n", uv_strerror(rc));
        abort();
    }

    /* Mark started BEFORE thread spawn — thread checks this to bail early
     * if shutdown raced init (shouldn't happen but defensive). */
    nova_abool_store(&_nova_driver.started, true);

    /* Spawn driver thread. */
    rc = uv_thread_create(&_nova_driver.thread, _nova_driver_main, NULL);
    if (rc != 0) {
        fprintf(stderr, "nova: driver uv_thread_create failed: %s\n", uv_strerror(rc));
        abort();
    }
}

void nova_driver_shutdown(void) {
    if (!nova_abool_load(&_nova_driver.started)) return;

    /* Signal stop flag — driver loop checks между iterations. */
    nova_abool_store(&_nova_driver.stop, true);

    /* Wake driver from uv_run via shutdown_async — ensures loop exits ASAP. */
    uv_async_send(&_nova_driver.shutdown_async);

    /* Wait for driver thread to finish. */
    uv_thread_join(&_nova_driver.thread);

    /* Drain any leftover jobs (workers may have submitted after stop signal
     * but before async fired — race; we leak those jobs at shutdown which
     * is acceptable). */
    nova_mutex_lock(&_nova_driver.jobs.mu);
    NovaDriverJob* head = _nova_driver.jobs.head;
    _nova_driver.jobs.head = NULL;
    _nova_driver.jobs.tail = NULL;
    nova_mutex_unlock(&_nova_driver.jobs.mu);
    /* Memory leak on shutdown — jobs allocated via nova_alloc (Boehm GC) или
     * malloc; either way OS reclaims at exit. Not worth careful cleanup. */
    (void)head;

    nova_mutex_destroy(&_nova_driver.jobs.mu);
    /* Loop closure handled in _nova_driver_main shutdown sequence. */

    nova_abool_store(&_nova_driver.started, false);
}

bool nova_driver_is_started(void) {
    return nova_abool_load(&_nova_driver.started) &&
           !nova_abool_load(&_nova_driver.stop);
}

int nova_driver_submit_job(NovaDriverJob* job) {
    if (!nova_abool_load(&_nova_driver.started)) return -1;
    if (nova_abool_load(&_nova_driver.stop))     return -1;
    if (!job) return -1;

    nova_mutex_lock(&_nova_driver.jobs.mu);
    job->next = NULL;
    if (_nova_driver.jobs.tail) {
        _nova_driver.jobs.tail->next = job;
    } else {
        _nova_driver.jobs.head = job;
    }
    _nova_driver.jobs.tail = job;
    nova_mutex_unlock(&_nova_driver.jobs.mu);

    uv_async_send(&_nova_driver.job_async);
    return 0;
}

/* ── Driver thread main ──────────────────────────────────────────── */

static void _nova_driver_main(void* arg) {
    (void)arg;

#ifdef NOVA_GC_BOEHM
    /* Register driver thread с Boehm. Driver code itself doesn't touch GC
     * heap directly, BUT processed jobs may transitively (e.g., wake worker
     * fiber whose context is GC-allocated). Safer to register. */
    struct GC_stack_base sb;
    if (GC_get_stack_base(&sb) == GC_SUCCESS) {
        GC_register_my_thread(&sb);
    }
#endif
#if NOVA_FIBER_ARENA_ENABLED
    /* [M-mn-spawnctx-corruption-cancel-wake]: native-стек драйвера в реестр
     * GC push_other_roots-колбэка (POSIX; Windows/non-Boehm — no-op). */
    nova_fiber_arena_register_native_stack();
#endif

    while (!nova_abool_load(&_nova_driver.stop)) {
        /* UV_RUN_ONCE: block until any handle fires (async wake from worker
         * job submission OR shutdown signal OR future timer/io events).
         * Returns when at least one event processed. */
        uv_run(&_nova_driver.loop, UV_RUN_ONCE);

        /* Drain job queue. Also drained от job_async_cb — this is backup
         * for jobs submitted after async fired но before we returned to
         * top of loop. */
        _nova_driver_drain_jobs();
    }

    /* Shutdown phase: close all active handles, run loop until clean. */
    uv_close((uv_handle_t*)&_nova_driver.job_async, NULL);
    uv_close((uv_handle_t*)&_nova_driver.shutdown_async, NULL);

    /* Run loop until no active handles (drains close callbacks). */
    while (uv_loop_alive(&_nova_driver.loop)) {
        uv_run(&_nova_driver.loop, UV_RUN_NOWAIT);
    }

    uv_loop_close(&_nova_driver.loop);

#if NOVA_FIBER_ARENA_ENABLED
    nova_fiber_arena_unregister_native_stack();
#endif
#ifdef NOVA_GC_BOEHM
    GC_unregister_my_thread();
#endif
}

/* ── Async callbacks (run на driver thread inside uv_run) ────────── */

static void _nova_driver_job_async_cb(uv_async_t* h) {
    (void)h;
    _nova_driver_drain_jobs();
}

static void _nova_driver_shutdown_async_cb(uv_async_t* h) {
    (void)h;
    /* No-op — flag check at loop top will exit. This handle just unblocks
     * uv_run. */
}

/* ── Job processing ──────────────────────────────────────────────── */

static void _nova_driver_drain_jobs(void) {
    /* Move entire queue under lock, then process outside lock (allows new
     * submissions during processing — they go to new queue, picked up next
     * iteration). */
    nova_mutex_lock(&_nova_driver.jobs.mu);
    NovaDriverJob* job = _nova_driver.jobs.head;
    _nova_driver.jobs.head = NULL;
    _nova_driver.jobs.tail = NULL;
    nova_mutex_unlock(&_nova_driver.jobs.mu);

    while (job) {
        NovaDriverJob* next = job->next;
        _nova_driver_process_job(job);
        /* Plan 83.11 Ф.2: job allocated via malloc by worker (nova_driver_submit_job
         * caller). Free after processing. Driver retains st pointers (inside job
         * union) which point to fiber-stack-allocated NovaSleepState — those live
         * separately while fiber parked. */
        free(job);
        job = next;
    }
}

/* ── Plan 83.11 Ф.3: sleep state machine — driver side ──────────── */

/* Forward decls of driver-side callbacks. */
static void _nova_driver_sleep_timer_cb(uv_timer_t* h);
static void _nova_driver_sleep_close_cb(uv_handle_t* h);

/* Insert st at head of scope's armed list. Driver-thread only — no lock. */
static void _nova_driver_arm_list_insert(NovaSleepState* st) {
    NovaFiberQueue* scope = st->cancel_scope;
    if (!scope) return;
    st->next_in_scope = scope->armed_sleeps_head;
    st->pprev_in_scope = &scope->armed_sleeps_head;
    if (scope->armed_sleeps_head) {
        scope->armed_sleeps_head->pprev_in_scope = &st->next_in_scope;
    }
    scope->armed_sleeps_head = st;
}

/* O(1) unlink — driver-thread only. */
static void _nova_driver_arm_list_unlink(NovaSleepState* st) {
    if (!st->pprev_in_scope) return;  /* already unlinked or never inserted */
    *(st->pprev_in_scope) = st->next_in_scope;
    if (st->next_in_scope) {
        st->next_in_scope->pprev_in_scope = st->pprev_in_scope;
    }
    st->pprev_in_scope = NULL;
    st->next_in_scope = NULL;
}

/* ARM_SLEEP job handler — driver thread. */
static void _nova_driver_handle_arm_sleep(NovaSleepState* st, uint64_t ms) {
    if (!st) return;
    if (getenv("NOVA_DIAG_P259_DL")) {
        fprintf(stderr, "[p259-dl] ARM_SLEEP recv st=%p scope=%p slot=%d ms=%llu expected_co=%p\n",
                (void*)st, (void*)st->scope, st->slot, (unsigned long long)ms, (void*)st->expected_co);
        fflush(stderr);
    }

    /* Init timer on driver's loop. */
    int rc = uv_timer_init(&_nova_driver.loop, &st->timer);
    if (rc != 0) {
        fprintf(stderr, "nova: driver uv_timer_init failed: %s\n", uv_strerror(rc));
        /* Move to CLOSED so worker fiber unparks (with error semantics — TBD). */
        nova_aint_store(&st->stage, NOVA_SLEEP_DRV_CLOSED);
        nova_sched_wake(st->scope, st->slot);
        return;
    }
    st->timer.data = st;

    /* Insert into scope's armed list BEFORE starting timer — если timer
     * fires immediately (ms=0), timer_cb might run synchronously? Actually
     * uv_timer_start with 0 fires on next loop iteration, not immediately.
     * But safe to insert first regardless. */
    _nova_driver_arm_list_insert(st);

    /* Transition NEW → ARMED. Single-mutator: no CAS needed for this transition.
     * RELEASE-store so worker's ACQUIRE-load sees it. */
    nova_aint_store(&st->stage, NOVA_SLEEP_DRV_ARMED);

    /* Start timer — может fire prior to return if ms is very small + something
     * weird, но libuv guarantees timer_cb runs only inside uv_run. We're called
     * from uv_run already (job_async_cb path). Timer registered to fire next
     * iteration. */
    rc = uv_timer_start(&st->timer, _nova_driver_sleep_timer_cb, ms, 0);
    if (rc != 0) {
        fprintf(stderr, "nova: driver uv_timer_start failed: %s\n", uv_strerror(rc));
        _nova_driver_arm_list_unlink(st);
        nova_aint_store(&st->stage, NOVA_SLEEP_DRV_CLOSED);
        uv_close((uv_handle_t*)&st->timer, NULL);  /* cleanup */
        nova_sched_wake(st->scope, st->slot);
        return;
    }
}

/* Timer fired naturally (sleep duration elapsed). Driver thread. */
static void _nova_driver_sleep_timer_cb(uv_timer_t* h) {
    NovaSleepState* st = (NovaSleepState*)h->data;
    if (!st) return;

    /* CAS ARMED → FIRING. Loser = cancel-job won race; cancel path will
     * uv_close. We just exit. */
    int32_t expected = NOVA_SLEEP_DRV_ARMED;
    if (!nova_aint_cas(&st->stage, &expected, NOVA_SLEEP_DRV_FIRING)) {
        return;
    }

    /* Won — initiate close. close_cb will wake worker fiber. */
    uv_close((uv_handle_t*)&st->timer, _nova_driver_sleep_close_cb);
}

/* CANCEL_SCOPE job handler — driver thread. */
static void _nova_driver_handle_cancel_scope(NovaFiberQueue* scope) {
    if (!scope) return;

    /* Walk armed list. Single-mutator (driver) — no race на list itself.
     * BUT: list modifications (insert/unlink) might happen while we iterate?
     * NO — both insert (ARM_SLEEP) и unlink (close_cb) run on driver thread.
     * We're on driver thread now. No concurrent modification possible.
     *
     * BUT: uv_close inside the loop schedules close_cb to run later. close_cb
     * will unlink st. So we must save next pointer BEFORE calling uv_close. */
    NovaSleepState* st = scope->armed_sleeps_head;
    while (st) {
        NovaSleepState* next = st->next_in_scope;

        /* CAS ARMED → CANCEL_REQ. Loser = timer_cb won (will close itself). */
        int32_t expected = NOVA_SLEEP_DRV_ARMED;
        if (nova_aint_cas(&st->stage, &expected, NOVA_SLEEP_DRV_CANCEL_REQ)) {
            uv_close((uv_handle_t*)&st->timer, _nova_driver_sleep_close_cb);
        }
        /* CAS loser: timer_cb already won, will close. Skip. */

        st = next;
    }

    /* Plan 83.11 §12.31: signal completion. Main thread spins on this counter
     * in nova_supervised_run_impl before returning, so the scope's stack frame
     * stays alive until we are done dereferencing its fields. RELEASE
     * synchronizes-with the main's ACQUIRE load. */
    (void)__atomic_fetch_sub(&scope->pending_driver_jobs, 1, __ATOMIC_RELEASE);
}

/* №398: CANCEL_SLOT job handler — driver thread. Targeted counterpart of
 * `_nova_driver_handle_cancel_scope` for exactly ONE (scope, slot) — see
 * driver.h `NOVA_DRV_JOB_CANCEL_SLOT` doc for why this exists (direct-body
 * `Time.sleep` arms itself under the OWNER scope, not the innermost
 * `supervised(cancel:)` scope). Walks `scope`'s armed list (same
 * single-mutator/driver-thread-only safety as `_handle_cancel_scope`) but
 * only CAS/close's entries whose `slot` matches — every OTHER armed sleep
 * belonging to unrelated slots of the (possibly outer/long-lived) owner
 * scope is left untouched. At most one match expected (a slot holds at
 * most one direct blocking op at a time) but the loop doesn't assume it —
 * same defensive stance as nova_sched_cancel_pending_slot's doc (stale/
 * reused slot = safe no-op). */
static void _nova_driver_handle_cancel_slot(NovaFiberQueue* scope, int slot) {
    if (!scope || slot < 0) return;

    NovaSleepState* st = scope->armed_sleeps_head;
    while (st) {
        NovaSleepState* next = st->next_in_scope;

        if (st->slot == slot) {
            int32_t expected = NOVA_SLEEP_DRV_ARMED;
            if (nova_aint_cas(&st->stage, &expected, NOVA_SLEEP_DRV_CANCEL_REQ)) {
                uv_close((uv_handle_t*)&st->timer, _nova_driver_sleep_close_cb);
            }
        }

        st = next;
    }

    /* Same lifetime contract as _handle_cancel_scope (§12.31): decrement
     * AFTER we're done dereferencing `scope`'s fields, so the submitter's
     * spin-wait (nova_supervised_run_impl's own pending_driver_jobs loop,
     * which every `supervised{}` frame — incl. the owner — runs before
     * returning) keeps the stack frame alive until here. */
    (void)__atomic_fetch_sub(&scope->pending_driver_jobs, 1, __ATOMIC_RELEASE);
}

/* CANCEL_TIMER job handler — driver thread. Single-timer cancel (для
 * cleanup callbacks of linked tokens etc). */
static void _nova_driver_handle_cancel_timer(NovaSleepState* st) {
    if (!st) return;
    int32_t expected = NOVA_SLEEP_DRV_ARMED;
    if (nova_aint_cas(&st->stage, &expected, NOVA_SLEEP_DRV_CANCEL_REQ)) {
        uv_close((uv_handle_t*)&st->timer, _nova_driver_sleep_close_cb);
    }
}

/* close_cb — final stage. Driver thread. Wakes worker fiber via generic
 * nova_sched_wake (race-free thanks к pending_wake[] integration в Plan 83.11
 * Option A — nova_sched.h). */
static void _nova_driver_sleep_close_cb(uv_handle_t* h) {
    NovaSleepState* st = (NovaSleepState*)h->data;
    if (!st) return;

    /* Unlink from armed list (driver-only). */
    _nova_driver_arm_list_unlink(st);

    NovaFiberQueue* sc = st->scope;
    int sl = st->slot;
    if (getenv("NOVA_DIAG_P259_DL")) {
        int dbg_count = sc ? __atomic_load_n(&sc->count, __ATOMIC_ACQUIRE) : -1;
        mco_coro* dbg_actual = (sc && sl >= 0 && sl < dbg_count) ? sc->fibers[sl] : NULL;
        fprintf(stderr, "[p259-dl] CLOSE_CB st=%p scope=%p slot=%d count=%d actual_co=%p expected_co=%p\n",
                (void*)st, (void*)sc, sl, dbg_count, (void*)dbg_actual, (void*)st->expected_co);
        fflush(stderr);
    }
    /* [M-mn-spawnctx-corruption-cancel-wake] fix (Plan 211 family): ACQUIRE-load
     * `count` BEFORE indexing `fibers[]`. This driver-thread read raced against
     * the WORKER thread's nova_scope_grow (fibers.h) — an unsynchronized
     * realloc-style swap of the `fibers`/`fiber_ctx`/... array pointers, run
     * under the worker's `slot_lock`, which is invisible to THIS unlocked
     * cross-thread reader. Under a shower of concurrent spawns (e.g. 2000
     * fibers all racing array growth while a 30ms sleep's close_cb already
     * fires on the driver thread), a plain (non-atomic) read of `sc->count`
     * has no happens-before edge with the grow's plain store of `sc->fibers`
     * — this thread could observe a FRESH (post-grow) `count` alongside a
     * STALE (pre-grow, since-abandoned, smaller) `fibers` array pointer,
     * indexing `sl` past that old buffer's bounds into adjacent heap memory
     * (gdb: `_nova_fiber_scope`/free-list garbage matches exactly this OOB
     * read). Mirrors the ALREADY-CORRECT pattern used by
     * nova_runtime_worker_pump_scope's cancel-delivery path (runtime.c):
     * "ACQUIRE-load on count pairs with the RELEASE-store in
     * nova_scope_alloc_slot, ensuring we see fibers[slot]=co when we observe
     * count=slot+1" — that site had it right; this one didn't. The RELEASE
     * store on `scope->count` (nova_scope_alloc_slot, fibers.h) happens
     * program-order-after nova_scope_grow's array-pointer writes on the
     * worker thread, so an ACQUIRE-load here that observes `sl < sc_count`
     * is guaranteed to also observe the fully-grown, correctly-sized
     * `fibers` array — no OOB read possible. */
    int sc_count = sc ? __atomic_load_n(&sc->count, __ATOMIC_ACQUIRE) : 0;
    mco_coro* actual_co = (sc && sl >= 0 && sl < sc_count) ? sc->fibers[sl] : NULL;

    if (actual_co != st->expected_co) {
        /* WRONG-FIBER: scope->fibers[slot] does not match expected_co.
         * Two sub-cases:
         *   A) actual_co==NULL — fibers[slot] became NULL while expected_co was parked
         *      (STALE race: alloc_slot saw NULL+parked=true and skipped the slot).
         *      expected_co is still alive in mco_yield.
         *   B) actual_co!=NULL — slot was reused by a different fiber after expected_co
         *      completed and freed the slot. expected_co is already dead.
         *
         * Plan 83-go-cmn Ф.2 (correction #1, driver Fix-B rewrite): the driver
         * holds expected_co directly, so the wake is BY-CO via nova_goready —
         * the same single-winner path the cancel/primitive use. This is strictly
         * cleaner than re-deriving identity from the slot: goready's WAIT->
         * DISPATCHED CAS only fires if expected_co is genuinely parked (sub-case A);
         * a dead/reused expected_co (sub-case B) has park_state==NIL/DEAD so goready
         * is a no-op latch we simply skip. */
        mco_coro* expected_co = st->expected_co;

        if (expected_co && mco_status(expected_co) == MCO_SUSPENDED) {
            /* Sub-case A: expected_co is alive, stuck in mco_yield due to STALE race.
             * Publish CLOSED so the park_until predicate is satisfied on resume. */
            __atomic_store_n(&st->stage, NOVA_SLEEP_DRV_CLOSED, __ATOMIC_SEQ_CST);
            /* Invalidate expected_co's slot record so its epilogue does NOT call
             * nova_scope_free_slot (the slot is unowned now — no new fiber took it
             * thanks to Fix A in alloc_slot). Use -2 sentinel (< 0, not -1). */
            NovaSpawnCtxBase* displaced_ctx =
                (NovaSpawnCtxBase*)mco_get_user_data(expected_co);
            /* [M-mn-spawnctx-corruption-cancel-wake] R1-трипваер: если
             * expected_co на деле умер и его арена-слот переиспользован новым
             * файбером, эта запись «-2» портит ЧУЖОЙ живой SpawnCtx. Диаг-режим
             * логирует displaced-событие и валидирует ctx перед записью. */
            {
                extern int  nova_spawn_pool_diag(void);
                extern void nova_spawn_ctx_diag_check_live(const void* vbase, const char* where);
                if (nova_spawn_pool_diag()) {
                    fprintf(stderr,
                            "nova: [R1-DIAG] driver WRONG-FIBER sub-case A: slot=%d "
                            "expected_co=%p actual_co=%p displaced_ctx=%p\n",
                            sl, (void*)expected_co, (void*)actual_co, (void*)displaced_ctx);
                    fflush(stderr);
                    if (displaced_ctx) {
                        nova_spawn_ctx_diag_check_live(displaced_ctx, "driver-displaced-write");
                    }
                }
            }
            if (displaced_ctx) {
                displaced_ctx->_nova_worker_slot = -2;  /* DISPLACED: epilogue skips free_slot */
            }
            /* By-co wake: goready wins WAIT->DISPATCHED, transitions fiber_state
             * PARKED->IDLE, clears parked[slot]/parked_co[slot], and dispatches
             * expected_co to its home worker. */
            nova_goready(expected_co);
            /* [M-driver-sleep-main-thread-wake-gap] — see the doc comment on
             * the "Normal path" call below for the full writeup; same gap
             * applies to this sub-case A resolution. */
            if (st->home_worker_id < 0) nova_runtime_signal_main();
        } else {
            /* Sub-case B: expected_co is dead — slot was legitimately reused. */
            __atomic_store_n(&st->stage, NOVA_SLEEP_DRV_CLOSED, __ATOMIC_SEQ_CST);
        }
        return;
    }

    /* Normal path: publish CLOSED, then wake by-co. */
    __atomic_store_n(&st->stage, NOVA_SLEEP_DRV_CLOSED, __ATOMIC_SEQ_CST);

    /* Generic wake — resolves parked_co[slot] and funnels through nova_goready;
     * the WAIT->DISPATCHED latch handles the wake-before-park race. */
    nova_sched_wake(st->scope, st->slot);

    /* [M-driver-sleep-main-thread-wake-gap] (found investigating Plan 259
     * regression in std/src/concurrency/supervised_cancel_direct_body_test.nv
     * test A2 — a direct-body `Time.sleep()` inside `supervised(timeout:)`
     * that should complete NATURALLY within budget was instead blocking for
     * the FULL outer timeout, ~100x its own duration).
     *
     * Root cause: `st->scope` for a direct-body sleep is the OWNER's real
     * (scope,slot) — e.g. `_nova_main_scope` for main-body — which always has
     * `dispatch_ready == NULL` (only an actual `NovaWorker.scope` gets that
     * hook wired, `runtime.c::_materialize_pool`). `nova_goready`'s bootstrap
     * branch (nova_sched.h, `dispatch_ready == NULL`) documents its own
     * contract as "supervised_step sees parked[slot] cleared... and resumes
     * the fiber itself" — true ONLY if whatever drives that fiber is about to
     * poll again SOON. For main-body specifically that driver is main's own
     * `uv_run(nova_current_loop(), UV_RUN_ONCE)` wait loop, which is a REAL
     * blocking syscall wait — it does not "poll again soon" on its own; it
     * only returns when AN EVENT FIRES ON MAIN'S OWN LOOP. This close_cb runs
     * on the DRIVER thread (a different OS thread/loop) — the wake above only
     * flips flags this coroutine's owner reads AFTER it wakes; nothing pokes
     * main's blocked loop to make it check sooner. The fiber is truly ready
     * within `ms` but is not actually resumed until something ELSE
     * independently wakes main's loop — in A2's case, the OUTER
     * `supervised(timeout:)`'s own early-deadline timer (armed directly on
     * `nova_current_loop()`, D451/Plan 221.1 №165), which is what finally let
     * the sleep "return" ~2s late, indistinguishable from a spurious timeout
     * because nothing threw.
     *
     * Fix: `home_worker_id` (captured on the sleep's own fiber thread before
     * submitting ARM_SLEEP, `_nova_sleep_via_driver`) already tells us whether
     * the sleeper's home is a NovaWorker (>=0, whose own loop uses NOWAIT
     * polling — no gap, see `_worker_main`) or the main OS thread (-1, the
     * ONLY genuinely-blocking pump in this runtime). `nova_runtime_signal_main`
     * is the SAME cross-thread poke workers already use for this exact
     * purpose (`nova_runtime_signal_main` doc comment, `runtime.c`) — reuse
     * it here instead of leaving main to depend on an unrelated timer to
     * eventually notice. No-op (checks `_main_wake_inited`) before the pool
     * materializes or after shutdown. Verified: this exact fixture went from
     * a deterministic ~2000ms/spurious-timeout to ~20ms/no-timeout with this
     * one-line addition, reproduced BOTH on this branch and on pristine
     * `main` (pre-existing, driver-mode-only latent bug — merely undetected
     * before because no prior regression test combined driver-already-
     * started + direct-body-sleep + an enclosing timeout/deadline + natural
     * completion; Plan 259 Layer 2 makes driver-mode universal from process
     * start, so this is the first wave where it fires deterministically). */
    if (st->home_worker_id < 0) nova_runtime_signal_main();
}

/* ── Plan 83.11 Ф.4: blocking offload via driver UV loop ──────────
 *
 * ARM_BLOCKING job handler — driver thread.
 *
 * Worker submits this job instead of calling uv_queue_work(nova_current_loop())
 * directly. Driver calls uv_queue_work on its own loop so the threadpool
 * work_cb / after_cb are anchored to the single driver UV loop.
 *
 * after_work_cb (_nova_blocking_after_cb in fibers.h) runs on driver thread:
 *   done = true (RELEASE)
 *   nova_sched_wake(scope, slot) → cross-thread dispatch to worker
 *
 * Wake-before-park race: if after_work_cb fires before worker reaches
 * nova_sched_park_until, the park_until fast-path predicate check
 * (_nova_blocking_is_done) returns true immediately → no yield needed.
 * For the case when nova_sched_find_state returns NULL (state not yet
 * created by nova_sched_register_pending): the RELEASE store to st->done
 * is still visible to the worker's ACQUIRE load in the park_until fast-path,
 * so the predicate returns true and the fiber skips parking entirely. */
static void _nova_driver_handle_arm_blocking(NovaBlockingState* st) {
    if (!st) return;
    int rc = uv_queue_work(&_nova_driver.loop, &st->work,
                           _nova_blocking_work_cb, _nova_blocking_after_cb);
    if (rc != 0) {
        fprintf(stderr, "nova: driver uv_queue_work(ARM_BLOCKING) failed: %s\n",
                uv_strerror(rc));
        /* Wake fiber so it doesn't park forever. done=true lets park_until
         * predicate return true and unblock the caller. */
        nova_abool_store(&st->done, true);
        nova_sched_wake(st->scope, st->slot);
    }
}

/* ── Job dispatch ────────────────────────────────────────────────── */

static void _nova_driver_process_job(NovaDriverJob* job) {
    switch (job->kind) {
    case NOVA_DRV_JOB_ARM_SLEEP:
        _nova_driver_handle_arm_sleep(job->u.arm_sleep.st, job->u.arm_sleep.ms);
        break;
    case NOVA_DRV_JOB_CANCEL_SCOPE:
        _nova_driver_handle_cancel_scope(job->u.cancel_scope.scope);
        break;
    case NOVA_DRV_JOB_CANCEL_TIMER:
        _nova_driver_handle_cancel_timer(job->u.cancel_timer.st);
        break;
    case NOVA_DRV_JOB_ARM_BLOCKING:
        _nova_driver_handle_arm_blocking(job->u.arm_blocking.st);
        break;
    case NOVA_DRV_JOB_CANCEL_SLOT:  /* №398 */
        _nova_driver_handle_cancel_slot(job->u.cancel_slot.scope, job->u.cancel_slot.slot);
        break;
    default:
        fprintf(stderr, "nova: driver unknown job kind %d\n", (int)job->kind);
        break;
    }
}