/*
  StarForth — Steady-State Virtual Machine Runtime

  Copyright (c) 2023–2025 Robert A. James
  All rights reserved.

  This file is part of the StarForth project.

  Licensed under the StarForth License, Version 1.0 (the "License");
  you may not use this file except in compliance with the License.

  You may obtain a copy of the License at:
      https://github.com/star.4th@proton.me/StarForth/LICENSE.txt

  This 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.

 See the License for the specific language governing permissions and
 limitations under the License.

 */

/**
 * kernel_hermes.c - Kernel-resident Hermes: the heat-coupled allocator,
 * its release path, and the four-counter ledger (FABRIC-3.6.md tasks
 * 2.2/2.3/2.4, item 28).
 *
 * CORRECTION (task 2.3, found while starting it): task 2.2's first cut of
 * sk_hermes_alloc() pulled reservoir heat but never admitted a real
 * Stadium-floor patron -- it just set a local `in_use` flag. That is
 * wrong: FABRIC-3.5.md SXXXIII.4 item 1 states plainly that
 * "MSG-FREE-NODE returns it via STADIUM-EVICT", which only has something
 * to evict if allocation admitted something. More importantly, SXL.4's
 * invariant is Sigma(resident patron heat) + reservoir + consumed ==
 * Q48_ONE -- if held message heat is not a resident patron, it is
 * invisible to every term of that equation while held, which cannot be
 * right. Corrected here: sk_hermes_alloc() now calls stadium_admit()
 * with the pulled heat, storing the returned cell index in the message's
 * own stadium_cell field for release to evict later. Behaviour
 * DELIVER (1), matching messaging.4th's own `SB-DELIVER STADIUM-ADMIT`
 * at its MSG-ALLOC site.
 */

#ifdef __STARKERNEL__

#include <string.h>
#include "freestanding/stdio.h" /* task 3.9 evidence print -- snprintf() */
#include "starkernel/vm/kernel_hermes.h"
#include "starkernel/vm/stadium.h"
#include "starkernel/q48_16.h"
#include "starkernel/kmalloc.h"
#include "console.h"
#include "vm.h" /* task 3.4 -- VM struct fields, vm_interpret() */

/* Freestanding: no libc printf. Prints an unsigned decimal, no leading
 * zeros -- same small helper stadium.c/capsule_vm_switch_signal.c each
 * carry their own copy of. */
static void console_put_u64(uint64_t v) {
    char buf[21];
    int  i = 20;
    buf[20] = '\0';
    if (v == 0) {
        console_puts("0");
        return;
    }
    while (v > 0 && i > 0) {
        buf[--i] = (char)('0' + (v % 10));
        v /= 10;
    }
    console_puts(&buf[i]);
}

static SkHermesMessage sk_hermes_msgs[SK_HERMES_MSG_MAX];

/* Task 2.4 (item 28), FABRIC-3.5.md SXXXVII.3/SXL.4's four counters.
 * `consumed` is declared now but touched nowhere yet -- task 2.5 adds
 * its one increment site when decay exists to record a delta from. */
static uint64_t sk_hermes_held     = 0;
static uint64_t sk_hermes_pulled   = 0;
static uint64_t sk_hermes_returned = 0;
static uint64_t sk_hermes_consumed = 0;

void sk_hermes_ledger(uint64_t *held, uint64_t *pulled, uint64_t *returned, uint64_t *consumed) {
    if (held)     *held     = sk_hermes_held;
    if (pulled)   *pulled   = sk_hermes_pulled;
    if (returned) *returned = sk_hermes_returned;
    if (consumed) *consumed = sk_hermes_consumed;
}

int sk_hermes_audit_values(uint64_t held, uint64_t pulled, uint64_t returned, uint64_t consumed) {
    /* Epsilon zero (SXXXVII.3): exact integer equality, no tolerance. */
    return held == pulled - returned - consumed;
}

static uint64_t sk_hermes_audit_failures = 0;

int sk_hermes_audit(void) {
    int ok = sk_hermes_audit_values(sk_hermes_held, sk_hermes_pulled,
                                    sk_hermes_returned, sk_hermes_consumed);
    if (!ok) {
        sk_hermes_audit_failures++;
        console_println("Kernel-Hermes AUDIT FAILURE: held != pulled - returned - consumed");
    }
    return ok;
}

uint64_t sk_hermes_audit_failure_count(void) {
    return sk_hermes_audit_failures;
}

uint64_t sk_hermes_scan_held(size_t *live_count) {
    uint64_t sum = 0;
    size_t n = 0;
    int i;

    for (i = 0; i < SK_HERMES_MSG_MAX; i++) {
        if (!sk_hermes_msgs[i].in_use || sk_hermes_msgs[i].stadium_cell < 0) continue;
        sum += stadium_cells()[sk_hermes_msgs[i].stadium_cell].header.heat;
        n++;
    }
    if (live_count) *live_count = n;
    return sum;
}

int sk_hermes_scan_check(void) {
    return sk_hermes_scan_held((size_t *)0) == sk_hermes_held;
}

static int sk_hermes_find_free_slot(void) {
    int i;
    for (i = 0; i < SK_HERMES_MSG_MAX; i++) {
        if (!sk_hermes_msgs[i].in_use) return i;
    }
    return -1;
}

int sk_hermes_alloc(VMUuid vm_id, SkHermesMessage **out_msg) {
    int slot;
    uint64_t pulled;
    StadiumPatronHeader candidate;
    size_t cell;

    if (!out_msg) return -1;

    /* Check affordability before touching the reservoir at all -- this is
     * what "roll back on refusal" reduces to when the check happens
     * first: there is nothing to roll back. */
    if (stadium_reservoir_peek(vm_id) < SK_HERMES_Q_SLOT) return -1;

    slot = sk_hermes_find_free_slot();
    if (slot < 0) return -1; /* arena full, reservoir untouched */

    pulled = stadium_reservoir_pull(vm_id, SK_HERMES_Q_SLOT);
    if (pulled < SK_HERMES_Q_SLOT) {
        /* Should not happen given the peek check above (nothing else runs
         * between the two calls on this single-core, cooperative kernel),
         * but roll back explicitly rather than trust that invariant
         * silently. */
        if (pulled > 0) stadium_reservoir_push(vm_id, pulled);
        return -1;
    }

    memset(&candidate, 0, sizeof(candidate));
    /* Matches messaging.4th's own MSG-ALLOC: the slot's own index becomes
     * the admitted patron's identity, which stadium_dispatch()'s DELIVER
     * case later prints back as "msg_idx=" on release -- an arbitrary
     * distinct value would satisfy the conservation math just as well,
     * but this keeps the diagnostic meaningful instead of every message
     * printing identity 0. */
    candidate.identity  = (uint64_t)slot;
    candidate.heat      = pulled; /* the reservoir pull itself, unmodified */
    candidate.ttl       = 0;      /* decay/TTL is task 2.5's scope, not this one's */
    candidate.link      = 0;
    candidate.contains  = STADIUM_CONTAINS_NONE;
    candidate.mass      = 1;      /* message body lives in sk_hermes_msgs[], not here */
    candidate.flags     = 0;
    candidate.behaviour = (uint8_t)STADIUM_BEHAVIOUR_DELIVER;

    cell = stadium_admit(vm_id, &candidate);
    if (cell == STADIUM_CELL_NONE) {
        /* Stadium floor itself refused (unrelated to heat affordability,
         * already confirmed above) -- roll back the pull, same "leave
         * everything exactly as found" discipline as every other refusal
         * path here. */
        stadium_reservoir_push(vm_id, pulled);
        return -1;
    }

    memset(&sk_hermes_msgs[slot], 0, sizeof(SkHermesMessage));
    sk_hermes_msgs[slot].in_use      = 1;
    sk_hermes_msgs[slot].stadium_cell = (int32_t)cell;
    sk_hermes_msgs[slot].owner        = vm_id;
    *out_msg = &sk_hermes_msgs[slot];

    /* Task 2.4: the one site where held/pulled increment -- only reached
     * once every refusal path above has already returned. */
    sk_hermes_held   += pulled;
    sk_hermes_pulled += pulled;
    sk_hermes_audit();

    return 0;
}

int sk_hermes_release(SkHermesMessage *msg) {
    int rc;
    uint64_t remaining;

    if (!msg || !msg->in_use) return -1;

    if (msg->stadium_cell < 0) {
        memset(msg, 0, sizeof(*msg));
        return -1;
    }

    /* Read the patron's current heat BEFORE eviction -- stadium_evict()
     * zeroes the header as part of returning the cell to the free list,
     * and its own return value is a success code, not the amount
     * credited. Today (before task 2.5's decay exists) this always
     * equals exactly what was pulled at allocation; once decay exists,
     * this is the message's true remaining heat, which is what actually
     * flows back to the reservoir -- reading it here rather than
     * assuming the original pulled amount is what keeps this correct
     * without changes once task 2.5 lands. */
    remaining = stadium_cells()[msg->stadium_cell].header.heat;

    /* stadium_evict() returns the departing patron's remaining heat to
     * its owner's reservoir itself (stadium.c: "the departing patron's
     * remaining heat must flow back to its owner's reservoir before the
     * cell returns to the free list") -- release does not touch the
     * reservoir directly, the eviction path does it, matching
     * MSG-FREE-NODE's own shape ("DUP 5 CELLS + @ STADIUM-EVICT DROP"). */
    rc = stadium_evict((size_t)msg->stadium_cell);

    /* Task 2.4: the one site where held/returned change on release. Only
     * on successful eviction -- a refused eviction leaves the patron
     * (and its heat) exactly where it was, so the ledger must not move
     * either. */
    if (rc == 0) {
        sk_hermes_held     -= remaining;
        sk_hermes_returned += remaining;
        sk_hermes_audit();
    }

    memset(msg, 0, sizeof(*msg));
    return rc;
}

int sk_hermes_decay(SkHermesMessage *msg) {
    uint64_t before, after;
    StadiumCell *cell;

    if (!msg || !msg->in_use || msg->stadium_cell < 0) return -1;

    cell   = &stadium_cells()[msg->stadium_cell];
    before = cell->header.heat;
    after  = (uint64_t)q48_mul((q48_16_t)before, (q48_16_t)SK_HERMES_Q_DECAY);

    cell->header.heat = after;

    /* Task 2.5: the one site where consumed increments (and where held
     * pays for it). SXXXVII.3: decay's delta is exact and in hand here. */
    sk_hermes_consumed += before - after;
    sk_hermes_held     -= before - after;
    stadium_consumed_record(msg->owner, before - after);
    sk_hermes_audit();

    return 0;
}

/* Task 3.2 (item 27, B1 / FABRIC-3.5.md SXLV.1): the channel table. See
 * kernel_hermes.h's own doc comment on this section for scope and sizing
 * reasoning. */
static SkHermesChannel *sk_hermes_channels          = (SkHermesChannel *)0;
static int               sk_hermes_channel_capacity_val = 0;

int sk_hermes_channels_boot_init(void) {
    size_t max_vm_count = stadium_max_vm_count();
    SkHermesChannel *table;
    size_t i;

    if (max_vm_count == 0) return -1; /* Stadium not yet initialized */

    table = (SkHermesChannel *)kmalloc(max_vm_count * sizeof(SkHermesChannel));
    if (!table) return -1;

    for (i = 0; i < max_vm_count; i++) {
        memset(&table[i], 0, sizeof(SkHermesChannel));
    }

    /* The common channel exists from boot (SXLV.1), empty -- VMs join it
     * at birth (kernel_main.c/capsule_birth.c task 3.2 wiring), not here. */
    table[SK_HERMES_CHANNEL_COMMON].in_use = 1;

    sk_hermes_channels          = table;
    sk_hermes_channel_capacity_val = (int)max_vm_count;

    console_puts("Kernel-Hermes: ");
    console_put_u64((uint64_t)max_vm_count);
    console_println(" channel slots");

    return 0;
}

int sk_hermes_channel_capacity(void) {
    return sk_hermes_channel_capacity_val;
}

static int channel_valid(int channel_id) {
    return sk_hermes_channels &&
           channel_id >= 0 &&
           channel_id < sk_hermes_channel_capacity_val &&
           sk_hermes_channels[channel_id].in_use;
}

int sk_hermes_channel_create(void) {
    int i;

    if (!sk_hermes_channels) return -1;

    /* Index 0 is always in_use (the common channel), so this scan never
     * returns it -- start at 1 purely as a scan-cost micro-optimisation,
     * not for correctness (the in_use check alone would already skip it). */
    for (i = 1; i < sk_hermes_channel_capacity_val; i++) {
        if (!sk_hermes_channels[i].in_use) {
            memset(&sk_hermes_channels[i], 0, sizeof(SkHermesChannel));
            sk_hermes_channels[i].in_use = 1;
            return i;
        }
    }
    return -1;
}

int sk_hermes_channel_destroy(int channel_id) {
    if (channel_id == SK_HERMES_CHANNEL_COMMON) return -1; /* permanent, SXLV.1 */
    if (!channel_valid(channel_id)) return -1;

    memset(&sk_hermes_channels[channel_id], 0, sizeof(SkHermesChannel));
    return 0;
}

int sk_hermes_channel_subscribe(int channel_id, VMUuid vm_id) {
    SkHermesMembership *m;
    size_t i;

    if (!channel_valid(channel_id)) return -1;

    m = &sk_hermes_channels[channel_id].membership;
    for (i = 0; i < m->count; i++) {
        if (vm_uuid_equal(m->members[i], vm_id)) return -1; /* already a member */
    }
    if (m->count >= SK_HERMES_MEMBER_MAX) return -1;

    m->members[m->count] = vm_id;
    m->count++;
    return 0;
}

int sk_hermes_channel_unsubscribe(int channel_id, VMUuid vm_id) {
    SkHermesMembership *m;
    size_t i;

    if (!channel_valid(channel_id)) return -1;

    m = &sk_hermes_channels[channel_id].membership;
    for (i = 0; i < m->count; i++) {
        if (vm_uuid_equal(m->members[i], vm_id)) {
            size_t j;
            for (j = i; j < m->count - 1; j++) {
                m->members[j] = m->members[j + 1];
            }
            m->count--;
            return 0;
        }
    }
    return -1; /* not a member */
}

int sk_hermes_channel_is_member(int channel_id, VMUuid vm_id) {
    SkHermesMembership *m;
    size_t i;

    if (!channel_valid(channel_id)) return 0;

    m = &sk_hermes_channels[channel_id].membership;
    for (i = 0; i < m->count; i++) {
        if (vm_uuid_equal(m->members[i], vm_id)) return 1;
    }
    return 0;
}

int sk_hermes_channel_member_count(int channel_id) {
    if (!channel_valid(channel_id)) return -1;
    return (int)sk_hermes_channels[channel_id].membership.count;
}

/* Task 3.3: per-subscriber pending queue table. See kernel_hermes.h's own
 * doc comments on sk_hermes_queues_boot_init()/SK_HERMES_PENDING_MAX for
 * scope and sizing reasoning. A queue is found-or-created lazily by
 * vm_id, same shape as stadium.c's quota_slot_for_vm() and session.c's
 * session_find() -- not pre-populated at birth like channel membership
 * is, since not every VM ever receives a message. */
typedef struct {
    VMUuid vm_id;
    int    in_use;
    int    slots[SK_HERMES_PENDING_MAX]; /* sk_hermes_msgs[] indices, circular FIFO */
    size_t head;
    size_t count;
} SkHermesPendingQueue;

static SkHermesPendingQueue *sk_hermes_queues             = (SkHermesPendingQueue *)0;
static int                   sk_hermes_queue_capacity_val = 0;

int sk_hermes_queues_boot_init(void) {
    size_t max_vm_count = stadium_max_vm_count();
    SkHermesPendingQueue *table;
    size_t i;

    if (max_vm_count == 0) return -1; /* Stadium not yet initialized */

    table = (SkHermesPendingQueue *)kmalloc(max_vm_count * sizeof(SkHermesPendingQueue));
    if (!table) return -1;

    for (i = 0; i < max_vm_count; i++) {
        memset(&table[i], 0, sizeof(SkHermesPendingQueue));
    }

    sk_hermes_queues             = table;
    sk_hermes_queue_capacity_val = (int)max_vm_count;

    console_puts("Kernel-Hermes: ");
    console_put_u64((uint64_t)max_vm_count);
    console_println(" pending-queue slots");

    return 0;
}

static SkHermesPendingQueue *find_queue(VMUuid vm_id) {
    int i;
    if (!sk_hermes_queues) return (SkHermesPendingQueue *)0;
    for (i = 0; i < sk_hermes_queue_capacity_val; i++) {
        if (sk_hermes_queues[i].in_use && vm_uuid_equal(sk_hermes_queues[i].vm_id, vm_id)) {
            return &sk_hermes_queues[i];
        }
    }
    return (SkHermesPendingQueue *)0;
}

static SkHermesPendingQueue *find_or_create_queue(VMUuid vm_id) {
    int i, free_slot = -1;

    if (!sk_hermes_queues) return (SkHermesPendingQueue *)0;

    for (i = 0; i < sk_hermes_queue_capacity_val; i++) {
        if (sk_hermes_queues[i].in_use && vm_uuid_equal(sk_hermes_queues[i].vm_id, vm_id)) {
            return &sk_hermes_queues[i];
        }
        if (!sk_hermes_queues[i].in_use && free_slot < 0) free_slot = i;
    }
    if (free_slot < 0) return (SkHermesPendingQueue *)0; /* table full */

    memset(&sk_hermes_queues[free_slot], 0, sizeof(SkHermesPendingQueue));
    sk_hermes_queues[free_slot].vm_id  = vm_id;
    sk_hermes_queues[free_slot].in_use = 1;
    return &sk_hermes_queues[free_slot];
}

/* Task 3.4: system-wide sum of every queue's count, maintained alongside
 * queue_push()/sk_hermes_pending_pop() below -- lets
 * sk_hermes_drain_checkpoint() skip its per-VM queue lookup entirely
 * (stadium_max_vm_count()-sized linear scan) on every word dispatch when
 * nothing is pending anywhere, the overwhelmingly common case. */
static int sk_hermes_pending_total = 0;

static int queue_push(SkHermesPendingQueue *q, int msg_index) {
    size_t tail;
    if (q->count >= SK_HERMES_PENDING_MAX) return -1;
    tail = (q->head + q->count) % SK_HERMES_PENDING_MAX;
    q->slots[tail] = msg_index;
    q->count++;
    sk_hermes_pending_total++;
    return 0;
}

int sk_hermes_send_one(VMUuid from, VMUuid to, uint32_t type, uint32_t channel_id,
                        void *payload_addr, uint32_t payload_len) {
    SkHermesMessage *msg;
    SkHermesPendingQueue *q;
    int idx;

    /* Task 3.5: the one-block bound, enforced uniformly here whether or
     * not payload_addr happens to be a chunk carrier -- see
     * kernel_hermes.h's own sizing note on why SK_HERMES_CHUNK_MAX_SLICE
     * (not SK_HERMES_CHUNK_MAX_PAYLOAD) is a chunk's own content size. */
    if (payload_len > SK_HERMES_CHUNK_MAX_PAYLOAD) return -1;

    if (sk_hermes_alloc(from, &msg) != 0) return -1;

    /* Real defect, found 2026-09-22 (FABRIC-3.6.md task 3.8's own
     * findings log): this used to store the caller's own payload_addr
     * as-is, a pointer this function does not own. Two sends before
     * either drains meant both messages pointed at the same
     * caller-owned buffer -- whichever send wrote last silently won.
     * Copy into this message's own payload_buf instead (kernel_hermes.h,
     * sized to SK_HERMES_CHUNK_MAX_PAYLOAD, the bound already enforced
     * above) and point payload_addr at that -- every message now owns
     * its payload bytes, independent of whatever the caller does with
     * its own buffer afterward. payload_len == 0 is a real, valid case
     * (an empty payload) -- memcpy of 0 bytes is well-defined even with
     * a NULL payload_addr -- guarded anyway (C's memcpy() is technically
     * undefined for a NULL argument even at length 0), matching this
     * project's own STRICT_PTR discipline elsewhere. */
    if (payload_len > 0) memcpy(msg->payload_buf, payload_addr, payload_len);
    msg->payload_addr = msg->payload_buf;

    msg->type         = type;
    msg->from         = from;
    msg->to           = to;
    msg->payload_len  = payload_len;
    msg->channel      = channel_id;

    q = find_or_create_queue(to);
    if (!q) {
        sk_hermes_release(msg); /* roll back -- no destination queue */
        return -1;
    }

    idx = (int)(msg - sk_hermes_msgs);
    if (queue_push(q, idx) != 0) {
        sk_hermes_release(msg); /* roll back -- destination queue full */
        return -1;
    }

    return 0;
}

int sk_hermes_publish(VMUuid from, int channel_id, uint32_t type,
                       void *payload_addr, uint32_t payload_len) {
    SkHermesMembership *m;
    size_t i;
    int delivered = 0;

    if (!channel_valid(channel_id)) return -1;
    if (payload_len > SK_HERMES_CHUNK_MAX_PAYLOAD) return -1; /* see sk_hermes_send_one()'s
                                                                * own check -- duplicated here
                                                                * so publish() itself returns
                                                                * -1 (not a silent 0) on an
                                                                * oversized payload */

    m = &sk_hermes_channels[channel_id].membership;
    for (i = 0; i < m->count; i++) {
        if (sk_hermes_send_one(from, m->members[i], type, (uint32_t)channel_id,
                                payload_addr, payload_len) == 0) {
            delivered++;
        }
    }
    return delivered;
}

int sk_hermes_pending_count(VMUuid vm_id) {
    SkHermesPendingQueue *q = find_queue(vm_id);
    return q ? (int)q->count : 0;
}

SkHermesMessage *sk_hermes_pending_peek(VMUuid vm_id) {
    SkHermesPendingQueue *q = find_queue(vm_id);
    if (!q || q->count == 0) return (SkHermesMessage *)0;
    return &sk_hermes_msgs[q->slots[q->head]];
}

int sk_hermes_pending_pop(VMUuid vm_id) {
    SkHermesPendingQueue *q = find_queue(vm_id);
    if (!q || q->count == 0) return -1;
    q->head = (q->head + 1) % SK_HERMES_PENDING_MAX;
    q->count--;
    sk_hermes_pending_total--;
    return 0;
}

int sk_hermes_drain_checkpoint(VM *vm) {
    SkHermesMessage *msg;
    VMUuid self;
    char   saved_input[INPUT_BUFFER_SIZE];
    size_t saved_length, saved_pos;
    vm_mode_t saved_mode;
    int    saved_error, saved_abort;
    int    drain_error;

    if (!vm) return -1;
    if (sk_hermes_pending_total == 0) return 0; /* fast path -- see this
                                                  * counter's own doc comment */

    self = vm->stadium_vm_id;
    msg  = sk_hermes_pending_peek(self);
    if (!msg) return 0;

    /* See kernel_hermes.h's own doc comment on this function for why all
     * six of these must be preserved, not just the obvious ones. */
    memcpy(saved_input, vm->input_buffer, sizeof(saved_input));
    saved_length = vm->input_length;
    saved_pos    = vm->input_pos;
    saved_mode   = vm->mode;
    saved_error  = vm->error;
    saved_abort  = vm->abort_requested;

    /* FABRIC-3.6.md task 3.9 evidence: CONSOLE-CMD-EVENT has no fixed
     * FORTH handler word to hang a print on (task 3.8's BLK-ATTACH-ACK
     * did) -- its payload is an arbitrary console line, interpreted
     * as-is below. So the evidence line lives here instead, gated on
     * msg->type, printed before vm_interpret() runs while this
     * message's heat is still held (release()/pop() are further down
     * in this same function) -- the same mid-hold-instant evidence task
     * 3.8 used, same honest limit: evidence for mid-hold, not
     * post-release. console_println, not log_message() (confirmed
     * invisible in this build's serial capture, task 3.8's own
     * finding); fires once per relayed console line, not per word. */
    if (msg->type == SK_HERMES_MSG_TYPE_CONSOLE_CMD) {
        uint64_t held, pulled, returned, consumed;
        char line[160];
        int conserved = stadium_conserved(self);
        sk_hermes_ledger(&held, &pulled, &returned, &consumed);
        snprintf(line, sizeof(line),
            "Kernel-Hermes CONSOLE-CMD-EVENT (real, Stage D): ledger held=%llu "
            "pulled=%llu returned=%llu consumed=%llu stadium_conserved(self)=%s",
            (unsigned long long)held, (unsigned long long)pulled,
            (unsigned long long)returned, (unsigned long long)consumed,
            conserved ? "true" : "FALSE");
        console_println(line);
    }

    /* FABRIC-3.6.md task 3.10 evidence: ELEVATE-REQUEST also has no fixed
     * C-side handler word -- its payload is a FORTH command string
     * calling ELEVATE-GRANT (zuse-eligibility.4th), interpreted as-is
     * below, same shape as CONSOLE-CMD-EVENT above. Same mid-hold-instant
     * evidence, same reasoning; `self` here is always Hera (`vm`'s own
     * caller already resolved it that way in KH-ELEVATE-SEND, since
     * ELEVATE-GRANT only ever runs on Hera). */
    if (msg->type == SK_HERMES_MSG_TYPE_ELEVATE_REQUEST) {
        uint64_t held, pulled, returned, consumed;
        char line[160];
        int conserved = stadium_conserved(self);
        sk_hermes_ledger(&held, &pulled, &returned, &consumed);
        snprintf(line, sizeof(line),
            "Kernel-Hermes ELEVATE-REQUEST (real, Stage D): ledger held=%llu "
            "pulled=%llu returned=%llu consumed=%llu stadium_conserved(self)=%s",
            (unsigned long long)held, (unsigned long long)pulled,
            (unsigned long long)returned, (unsigned long long)consumed,
            conserved ? "true" : "FALSE");
        console_println(line);
    }

    vm->mode = MODE_INTERPRET; /* a payload is never compiled into whatever
                                * the enclosing context was mid-defining */
    vm_interpret(vm, (const char *)msg->payload_addr);
    drain_error = vm->error;

    memcpy(vm->input_buffer, saved_input, sizeof(saved_input));
    vm->input_length    = saved_length;
    vm->input_pos       = saved_pos;
    vm->mode            = saved_mode;
    vm->error            = saved_error; /* a bad message must not abort the
                                          * enclosing execution */
    vm->abort_requested  = saved_abort;

    sk_hermes_release(msg);
    sk_hermes_pending_pop(self);

    return drain_error ? -1 : 1;
}

uint32_t sk_hermes_chunk_count(uint32_t len) {
    if (len == 0) return 0;
    return (len + SK_HERMES_CHUNK_MAX_SLICE - 1) / SK_HERMES_CHUNK_MAX_SLICE;
}

int sk_hermes_reassemble(SkHermesMessage **chunks, int n_chunks,
                          uint8_t *out_buf, uint32_t out_buf_cap,
                          uint32_t *out_len) {
    int present[SK_HERMES_MSG_MAX];
    int i;
    uint32_t msg_id;
    uint32_t total;
    uint32_t last_slice_len = 0;

    if (!chunks || n_chunks <= 0 || !out_buf || !out_len) return -1;
    if (n_chunks > SK_HERMES_MSG_MAX) return -1; /* guard before indexing present[] */

    for (i = 0; i < n_chunks; i++) present[i] = 0;

    if (!chunks[0] || !chunks[0]->payload_addr ||
        chunks[0]->payload_len < (uint32_t)sizeof(SkHermesChunkHeader)) return -1;
    msg_id = ((SkHermesChunkHeader *)chunks[0]->payload_addr)->msg_id;

    for (i = 0; i < n_chunks; i++) {
        SkHermesMessage *m = chunks[i];
        SkHermesChunkHeader *h;
        uint32_t slice_len;

        if (!m || !m->payload_addr || m->payload_len < (uint32_t)sizeof(SkHermesChunkHeader)) return -1;
        h = (SkHermesChunkHeader *)m->payload_addr;
        slice_len = m->payload_len - (uint32_t)sizeof(SkHermesChunkHeader);

        if (h->msg_id != msg_id) return -1;
        if (h->seq >= (uint32_t)n_chunks) return -1;
        if (present[h->seq]) return -1; /* duplicate seq */
        present[h->seq] = 1;

        if (h->seq == (uint32_t)(n_chunks - 1)) {
            if (!h->is_last) return -1;
            if (slice_len == 0 || slice_len > SK_HERMES_CHUNK_MAX_SLICE) return -1;
            last_slice_len = slice_len;
        } else {
            if (h->is_last) return -1;
            if (slice_len != SK_HERMES_CHUNK_MAX_SLICE) return -1;
        }
    }
    for (i = 0; i < n_chunks; i++) if (!present[i]) return -1; /* gap */

    /* Total computed once from validated seq completeness, checked once
     * against out_buf_cap, before any memcpy -- never order-dependent on
     * which chunk happens to overflow first. */
    total = (uint32_t)(n_chunks - 1) * SK_HERMES_CHUNK_MAX_SLICE + last_slice_len;
    if (total > out_buf_cap) return -1;

    for (i = 0; i < n_chunks; i++) {
        SkHermesMessage *m = chunks[i];
        SkHermesChunkHeader *h = (SkHermesChunkHeader *)m->payload_addr;
        uint32_t slice_len = m->payload_len - (uint32_t)sizeof(SkHermesChunkHeader);
        uint32_t offset = h->seq * SK_HERMES_CHUNK_MAX_SLICE;

        memcpy(out_buf + offset, (uint8_t *)m->payload_addr + sizeof(SkHermesChunkHeader), slice_len);
    }

    *out_len = total;
    return 0;
}

int sk_hermes_channel_request(VMUuid requester, VMUuid target) {
    return sk_hermes_send_one(requester, target, SK_HERMES_MSG_TYPE_CH_REQUEST,
                               SK_HERMES_CHANNEL_COMMON, (void *)0, 0);
}

int sk_hermes_channel_respond(VMUuid target, VMUuid requester, int approved) {
    int ch = -1;

    if (approved) {
        ch = sk_hermes_channel_create();
        if (ch >= 0) {
            if (sk_hermes_channel_subscribe(ch, requester) != 0 ||
                sk_hermes_channel_subscribe(ch, target) != 0) {
                sk_hermes_channel_destroy(ch); /* leave no half-open channel */
                ch = -1;
            }
        }
    }

    if (ch >= 0) {
        sk_hermes_send_one(target, requester, SK_HERMES_MSG_TYPE_CH_GRANT,
                            (uint32_t)ch, (void *)0, 0);
        /* Ruled ACK cadence (2026-09-21): channel-open + delivery, not
         * every message -- this IS that moment. */
        sk_hermes_send_one(target, requester, SK_HERMES_MSG_TYPE_ACK,
                            (uint32_t)ch, (void *)0, 0);
    } else {
        /* SXLV.1: "a deny is a NACK" -- no separate CH_DENY type. */
        sk_hermes_send_one(target, requester, SK_HERMES_MSG_TYPE_NACK,
                            SK_HERMES_CHANNEL_COMMON, (void *)0, 0);
    }

    return ch;
}

int sk_hermes_channel_close(VMUuid closer, int channel_id) {
    if (channel_id == SK_HERMES_CHANNEL_COMMON) return -1;
    if (!sk_hermes_channel_is_member(channel_id, closer)) return -1;
    return sk_hermes_channel_destroy(channel_id);
}

int sk_hermes_channel_open_policy(VM *target_vm, VMUuid requester) {
    static const char word_name[] = "HERMES-CHANNEL-OPEN?";
    DictEntry *entry;
    DictEntry *saved_entry;
    cell_t result;

    if (!target_vm) return 0;

    entry = vm_find_word(target_vm, word_name, sizeof(word_name) - 1);
    if (!entry || !entry->func) return 0; /* fail closed: no policy, no open */

    /* Colon words dispatch through execute_colon_word(), which reads its
     * own body address from vm->current_executing_entry and no-ops
     * silently if it is NULL (vm_core.c:730) -- vm_interpret_word() always
     * sets this immediately before calling entry->func(); matched here
     * for the same reason (found live: without it, every call silently
     * did nothing, leaving the pushed requester untouched on the stack
     * rather than erroring, which is why this needed a debug session to
     * catch rather than showing up as an obvious crash). Saved and
     * restored, not just set, in case target_vm is ever called into
     * mid-dispatch (not exercised by this task's own self-test, but no
     * reason to assume it away). */
    saved_entry = target_vm->current_executing_entry;
    target_vm->current_executing_entry = entry;
    vm_push(target_vm, (cell_t)requester.hi);
    vm_push(target_vm, (cell_t)requester.lo);
    entry->func(target_vm);
    target_vm->current_executing_entry = saved_entry;

    if (target_vm->error || target_vm->dsp < 0) {
        target_vm->error = 0; /* a broken policy word must not leak into
                                * whatever else target_vm is doing */
        return 0;
    }

    result = vm_pop(target_vm);
    return result != 0;
}

#endif /* __STARKERNEL__ */