LithosAnanke/src/starkernel/vm/kernel_hermes.c

/*
  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 "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_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;

    m = &sk_hermes_channels[channel_id].membership;
    for (i = 0; i < m->count; i++) {
        VMUuid to = m->members[i];
        SkHermesMessage *msg;
        SkHermesPendingQueue *q;
        int idx;

        if (sk_hermes_alloc(from, &msg) != 0) continue; /* this subscriber skipped, not fatal */

        msg->type         = type;
        msg->from         = from;
        msg->to           = to;
        msg->payload_addr = payload_addr;
        msg->payload_len  = payload_len;
        msg->channel      = (uint32_t)channel_id;

        q = find_or_create_queue(to);
        if (!q) {
            sk_hermes_release(msg); /* roll back this subscriber's allocation */
            continue;
        }

        idx = (int)(msg - sk_hermes_msgs);
        if (queue_push(q, idx) != 0) {
            sk_hermes_release(msg); /* this subscriber's own queue is full */
            continue;
        }

        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;

    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;
}

#endif /* __STARKERNEL__ */