server: a learning ledger read answers from the loop's last publication, never queued behind the chunk it reports on

tools/server/server-context-learn.inc

@@ -226,9 +226,18 @@
             learn_ledger.revoke(body.value("id", std::string()), now);
             out["revoked"] = true; // idempotent: this id can admit no new work
         }
-        out["max_period_ms"] = server_learn_ledger::kMaxPeriodMs;
-        out["max_expiry_ms"] = server_learn_ledger::kMaxPeriodMs;
-        out["grants"] = json::array();
+        const json receipt = ledger_receipt(now);
+        for (const auto & entry : receipt.items()) { out[entry.key()] = entry.value(); }
+        out["residencies"] = residency_snapshot();
+        result->receipt = std::move(out);
+        queue_results.send(std::move(result));
+    }
+
+    // The loop-owned half of the ledger (grants, their use, the memory the optimizer holds), read
+    // ON THE LOOP: the reply to a mutation, and the snapshot every boundary publishes.
+    json ledger_receipt(int64_t now) const {
+        json out = {{"max_period_ms", server_learn_ledger::kMaxPeriodMs}, {"max_expiry_ms", server_learn_ledger::kMaxPeriodMs},
+            {"grants", json::array()}};
         for (const auto & a : learn_ledger.snapshot()) {
             out["grants"].push_back({{"id", a.grant.id}, {"period_ms", a.grant.period_ms},
                 {"max_chunk_us", a.grant.max_chunk_us}, {"share_ppm", a.grant.share_ppm}, {"expires_at_us", a.grant.expires_at_us},
@@ -240,9 +249,55 @@
             }
         }
         out["memory"] = learn_memory_receipt();
+        return out;
+    }
+
+    // A control read never queues behind the work it reports on (card 7fc7e2ce, Fable's design):
+    // the loop publishes its half of the ledger at every task and slot-update boundary, stamped,
+    // and a reader on any thread takes the last publication plus the residency mirror (its own
+    // mutex). At most one boundary stale, and the stamp says when.
+    struct ledger_publication {
+        std::mutex mutex;
+        std::condition_variable published; // signalled at every publication: the wait-on-ticket wake
+        json receipt;
+        uint64_t sequence = 0;
+    };
+    mutable ledger_publication ledger_published;
+    uint64_t ledger_published_revision = 0; // loop-owned: the ledger revision the last publication holds
+
+    // ON THE LOOP, at a slot-update boundary: publish only when the ledger changed since the last
+    // publication (a task boundary, below, always publishes).
+    void publish_ledger_if_changed() {
+        if (learn_ledger.revision() != ledger_published_revision) {
+            publish_ledger();
+        }
+    }
+
+    // ON THE LOOP, at a boundary.
+    void publish_ledger() {
+        ledger_published_revision = learn_ledger.revision();
+        const auto now = ggml_time_us();
+        json receipt = ledger_receipt(now);
+        receipt["clock_us"] = now;
+        {
+            std::lock_guard<std::mutex> lock(ledger_published.mutex);
+            receipt["sequence"] = ++ledger_published.sequence;
+            ledger_published.receipt = std::move(receipt);
+        }
+        ledger_published.published.notify_all();
+    }
+
+    // ANY THREAD: the last publication, the live residency mirror and the engine identity (const).
+    json published_ledger() const {
+        json out;
+        {
+            std::lock_guard<std::mutex> lock(ledger_published.mutex);
+            out = ledger_published.receipt;
+        }
+        out["engine"] = learn_engine;
+        out["refusal"] = "none";
         out["residencies"] = residency_snapshot();
-        result->receipt = std::move(out);
-        queue_results.send(std::move(result));
+        return out;
     }
 
 

tools/server/server-context.cpp

@@ -910,6 +910,8 @@ struct server_context_impl {
         catch (const std::exception &) { return "invalid"; }
     }
     const std::string & learning_engine_identity() const { return learn_engine; }
+    // the ledger as the loop last published it; readable from any thread (card 7fc7e2ce)
+    json ledger_as_published() const { return published_ledger(); }
 
     json lora_report_snapshot() const {
         std::lock_guard<std::mutex> lock(lora_report_mutex);
@@ -1571,16 +1573,27 @@ struct server_context_impl {
 
         // wiring up server queues
         queue_tasks.on_new_task([this](server_task && task, bool is_yielding) {
-            return process_single_task(std::move(task), is_yielding);
+            // the loop's own per-step continuation changes the ledger only through it (revision);
+            // every other task (a request, a ledger or adapter operation) publishes at its boundary
+            const bool continuation = task.type == SERVER_TASK_TYPE_NEXT_RESPONSE;
+            auto handled = process_single_task(std::move(task), is_yielding);
+            if (continuation) {
+                publish_ledger_if_changed();
+            } else {
+                publish_ledger();
+            }
+            return handled;
         });
         queue_tasks.on_update_slots([this]() {
             update_slots();
+            publish_ledger_if_changed(); // a chunk boundary: a learn chunk moved the ledger
         });
         queue_tasks.on_sleeping_state([this](bool sleeping) {
             handle_sleeping_state(sleeping);
         });
 
         metrics.init();
+        publish_ledger(); // readable before the loop's first boundary
 
         if (params_base.cache_idle_slots) {
             if (params_base.cache_ram_mib == 0) {
@@ -6577,7 +6590,12 @@ void server_routes::init_routes() {
     this->post_learn_residency = [grant_route](const server_http_req & req) { return grant_route(req, "residency_register"); };
     this->release_learn_residency = [grant_route](const server_http_req & req) { return grant_route(req, "residency_release"); };
     this->revoke_learn_residency = [grant_route](const server_http_req & req) { return grant_route(req, "residency_revoke"); };
-    this->get_learn_grants = [grant_route](const server_http_req & req) { return grant_route(req, "list"); };
+    // a read answers from the loop's last publication, never queued behind the chunk it reports on
+    this->get_learn_grants = [this](const server_http_req &) {
+        auto res = create_response();
+        res->ok(ctx_server.ledger_as_published());
+        return res;
+    };
     this->post_learn_grants = [grant_route](const server_http_req & req) { return grant_route(req, "register"); };
     this->revoke_learn_grant = [grant_route](const server_http_req & req) { return grant_route(req, "revoke"); };
 

tools/server/server-learn-ledger.h

@@ -76,8 +76,13 @@ class server_learn_ledger {
 
     explicit server_learn_ledger(std::string engine) : engine(std::move(engine)) {}
 
+    // Bumped by every mutation: the loop publishes the ledger at a slot-update boundary only when
+    // this moved since its last publication (card 7fc7e2ce; Fable on fork #109: no JSON per batch).
+    uint64_t revision() const { return revision_; }
+
     // The authority registers (or re-confirms) a grant.
     server_learn_refusal register_grant(const server_learn_grant & g, int64_t now_us) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         if (!in_clock(now_us) || g.id.empty() || (g.period_ms < 0 || (g.period_ms == 0 && g.share_ppm != 0)) || g.period_ms > kMaxPeriodMs || g.share_ppm < 0 ||
                 g.share_ppm > 1000000 || g.max_chunk_us < 0 || g.max_chunk_us > kMaxCostUs || g.expires_at_us <= now_us || !in_clock(g.expires_at_us)) {
             return server_learn_refusal::invalid;
@@ -102,6 +107,7 @@ class server_learn_ledger {
 
     // The authority ends a grant. Its account goes once its last reservation settles.
     void revoke(const std::string & id, int64_t now_us) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         if (!in_clock(now_us)) {
             return;
         }
@@ -116,6 +122,7 @@ class server_learn_ledger {
     // serving at admission, so the step's time is debited; an uncontended step is admitted
     // without a debit but is still a live reservation.
     server_learn_admission admit(const std::string & grant, int64_t est_us, int64_t now_us, bool contended) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         server_learn_admission out;
         if (est_us < 0 || est_us > kMaxCostUs || !in_clock(now_us)) {
             out.refusal = server_learn_refusal::invalid;
@@ -155,6 +162,7 @@ class server_learn_ledger {
     // Latency admission is separate from duty and restoration custody. A new
     // serving arrival may wait for this segment even when admission is idle.
     server_learn_admission admit_chunk(const std::string & grant, int64_t measured_us, int64_t now_us, bool contended) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         if (measured_us < 0 || measured_us > kMaxCostUs || !in_clock(now_us)) {
             return {server_learn_refusal::invalid};
         }
@@ -179,6 +187,7 @@ class server_learn_ledger {
     // Restoration is an obligation, not permission for a gradient. Keep its
     // measured debit even on an idle-only grant and after revocation.
     server_learn_admission hold_restoration(const std::string & grant, int64_t now_us) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         auto out = admit(grant, 0, now_us, false);
         if (out.refusal == server_learn_refusal::none) {
             reservations.at(out.reservation).contended = true;
@@ -190,6 +199,7 @@ class server_learn_ledger {
     // grant was revoked or expired after admission (unknown_grant): the reservation stays held, so
     // the caller cancels it (it never ran) and nothing is erased silently.
     server_learn_refusal start(uint64_t id, int64_t now_us) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         if (!in_clock(now_us)) {
             return server_learn_refusal::invalid;
         }
@@ -211,6 +221,7 @@ class server_learn_ledger {
     // The step quiesced after measured_us with committed_steps optimizer steps: debit the measured
     // time (when it was admitted contended) and release the rest of its reservation.
     server_learn_refusal finish(uint64_t id, int64_t measured_us, int64_t committed_steps, int64_t now_us) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         if (measured_us < 0 || measured_us > kMaxCostUs || committed_steps < 0 || committed_steps > kMaxStepsPerSettle ||
                 !in_clock(now_us)) {
             return server_learn_refusal::invalid;
@@ -220,6 +231,7 @@ class server_learn_ledger {
 
     // The step never ran: its whole reservation returns. Refused once it started.
     server_learn_refusal cancel(uint64_t id, int64_t now_us) {
+        ++revision_; // every mutation is a new ledger the loop must publish
         if (!in_clock(now_us)) {
             return server_learn_refusal::invalid;
         }
@@ -263,6 +275,7 @@ class server_learn_ledger {
     }
 
 private:
+    uint64_t revision_ = 0;
     struct account {
         server_learn_grant grant;
         int64_t            period_start_us;

tools/server/tests/learn-ledger-off-loop.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""Real server test; requires an isolated server. Usage: learn-ledger-off-loop.py URL
+
+What this catches (card 7fc7e2ce): a ledger read queued behind the work it controls. On a CPU
+node one prefill chunk ran ~40 s while the core waited 4 s for GET /learn/grants, so every
+learner stop read as unreachable. The read answers from the loop's last publication: it carries
+the publication's sequence and clock, the sequence advances at boundaries, a grant registered
+through the loop is in the next read, and a read answers while a completion is in flight.
+Run with --train-dir FIXTURE.
+"""
+import os
+import sys
+import threading
+
+sys.path.insert(0, os.path.dirname(__file__))
+from learn_fixture import Engine
+
+engine = Engine(sys.argv[1], "ledger", timeout=120)
+
+first = engine.get("/learn/grants")
+for key in ("sequence", "clock_us", "grants", "memory", "residencies", "engine", "max_expiry_ms"):
+    assert key in first, ("a read carries the publication's stamp and both halves of the ledger", key, first)
+assert first["engine"] == engine.engine, first
+
+# a mutation goes through the loop; the publication after it holds its result
+gid = engine.register_grant("-seen")
+seen = engine.get("/learn/grants")
+assert seen["sequence"] > first["sequence"], ("a boundary publishes again", first["sequence"], seen["sequence"])
+assert any(g["id"] == gid for g in seen["grants"]), ("the registered grant is in the next read", gid, seen["grants"])
+
+# a read answers while a completion holds the loop: the completion fills its slot's window
+# (ignore_eos), the read is taken only once /slots reports it processing, and the completion must
+# still be running when the read returns
+done = threading.Event()
+budget = engine.get("/slots")[0]["n_ctx"] // 2
+def complete():
+    engine.ok("/completion", {"prompt": "Count the windows in the room:", "n_predict": budget,
+                              "temperature": 0, "ignore_eos": True, "id_slot": 0})
+    done.set()
+worker = threading.Thread(target=complete)
+worker.start()
+while not any(slot.get("is_processing") for slot in engine.get("/slots")):
+    assert not done.is_set(), "the completion ended before it was ever seen processing"
+during = engine.get("/learn/grants")
+assert not done.is_set(), "the ledger read waited for the completion it ran beside"
+worker.join()
+assert during["sequence"] >= seen["sequence"], during
+
+engine.close()
+print("ok: ledger reads answer from the loop's publication")