server: a ticketed learning mutation answers at once, and its result is pushed on GET /learn/events

tools/server/server-context.cpp

@@ -6554,6 +6554,15 @@ void server_routes::init_routes() {
             }
         }
         task.learn_grants["op"] = op;
+        if (task.learn_grants.value("ticket", false)) {
+            // a ticketed mutation answers at once; its result arrives on GET /learn/events (card 7fc7e2ce)
+            task.id = queue_tasks.get_new_id();
+            const int ticket = task.id;
+            queue_results.add_ticket(ticket);
+            queue_tasks.post(std::move(task));
+            res->ok({{"ticket", ticket}});
+            return res;
+        }
         auto & rd = res->rd;
         task.id = rd.get_new_id();
         rd.post_task(std::move(task));
@@ -6590,6 +6599,47 @@ 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"); };
+    // The learning event stream (card 7fc7e2ce): every ticketed mutation's terminal result, pushed to
+    // the one live subscriber, then gone. Opened once per engine incarnation by the core; its end is
+    // the incarnation gate (the stream drops with the engine). Idle, it sends SSE keepalives.
+    this->get_learn_events = [this](const server_http_req & req) {
+        struct ticket_stream : server_res_generator {
+            server_response & results;
+            ticket_stream(server_queue & tasks, server_response & results, int sleep_idle_seconds)
+                : server_res_generator(tasks, results, sleep_idle_seconds, true), results(results) {
+                results.subscribe_tickets();
+            }
+            void on_complete() override {
+                results.unsubscribe_tickets();
+                server_res_generator::on_complete();
+            }
+        };
+        auto res = std::make_unique<ticket_stream>(queue_tasks, queue_results, params.sleep_idle_seconds);
+        res->set_req(&req); // its connection ends the stream (should_stop)
+        res->status = 200;
+        res->content_type = "text/event-stream";
+        res->data = "data: " + safe_json_to_str({{"engine", ctx_server.learning_engine_identity()}, {"subscribed", true}}) + "\n\n";
+        const int keepalive = params.sse_ping_interval > 0 ? params.sse_ping_interval : 1;
+        res->set_next([res_this = res.get(), keepalive](std::string & output) -> bool {
+            if (res_this->should_stop()) {
+                return false;
+            }
+            if (!res_this->data.empty()) {
+                output = std::move(res_this->data);
+                res_this->data.clear();
+                return true;
+            }
+            auto result = res_this->results.next_ticket(keepalive);
+            if (!result) {
+                output = ":\n\n"; // keepalive
+                return true;
+            }
+            output = "data: " + safe_json_to_str({{"ticket", result->id}, {"error", result->is_error()}, {"result", result->to_json()}}) + "\n\n";
+            return true;
+        });
+        return res;
+    };
+
     // 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();
@@ -6708,6 +6758,18 @@ void server_routes::init_routes() {
             res->error(format_error_response("\"id\" (an integer adapter id) is required", ERROR_TYPE_INVALID_REQUEST));
             return res;
         }
+        if (body.value("ticket", false)) {
+            // a ticketed unload answers at once; its result arrives on GET /learn/events (card 7fc7e2ce)
+            server_task task(SERVER_TASK_TYPE_UNLOAD_LORA);
+            task.id          = queue_tasks.get_new_id();
+            task.unload_lora = body.at("id").get<int32_t>();
+            task.unload_engine = body.at("engine").get<std::string>();
+            const int ticket = task.id;
+            queue_results.add_ticket(ticket);
+            queue_tasks.post(std::move(task));
+            res->ok({{"ticket", ticket}});
+            return res;
+        }
         auto & rd = res->rd;
         {
             server_task task(SERVER_TASK_TYPE_UNLOAD_LORA);

tools/server/server-context.h

@@ -177,6 +177,7 @@ struct server_routes {
     server_http_context::handler_t post_learn;
     server_http_context::handler_t post_learn_residency, release_learn_residency, revoke_learn_residency;
     server_http_context::handler_t get_learn_grants, post_learn_grants, revoke_learn_grant, post_learn_preflight;
+    server_http_context::handler_t get_learn_events;
     server_http_context::handler_t get_train;
     server_http_context::handler_t post_train;
     server_http_context::handler_t post_train_cancel;

tools/server/server-queue.cpp

@@ -481,6 +481,14 @@ void server_response::send(server_task_result_ptr && result) {
     RES_DBG("sending result for task id = %d\n", result->id);
 
     std::unique_lock<std::mutex> lock(mutex_results);
+    if (ticketed_ids.erase(result->id) > 0) {
+        // stored before the notify, under the same lock the stream waits with (no lost wakeup)
+        if (ticket_subscribers > 0) {
+            ticket_results.emplace_back(std::move(result));
+            condition_results.notify_all();
+        }
+        return;
+    }
     for (const auto & id_task : waiting_task_ids) {
         if (result->id == id_task) {
             RES_DBG("task id = %d pushed to result queue\n", result->id);
@@ -619,3 +627,32 @@ void server_response_reader::stop() {
         SRV_DBG("%s", "all tasks already finished, no need to cancel\n");
     }
 }
+
+void server_response::add_ticket(int id_task) {
+    std::unique_lock<std::mutex> lock(mutex_results);
+    ticketed_ids.insert(id_task);
+}
+
+void server_response::subscribe_tickets() {
+    std::unique_lock<std::mutex> lock(mutex_results);
+    ++ticket_subscribers;
+}
+
+void server_response::unsubscribe_tickets() {
+    std::unique_lock<std::mutex> lock(mutex_results);
+    if (--ticket_subscribers <= 0) {
+        ticket_subscribers = 0;
+        ticket_results.clear(); // nobody is listening: undelivered results are gone
+    }
+}
+
+server_task_result_ptr server_response::next_ticket(int timeout) {
+    std::unique_lock<std::mutex> lock(mutex_results);
+    condition_results.wait_for(lock, std::chrono::seconds(timeout), [&] { return !running || !ticket_results.empty(); });
+    if (!running || ticket_results.empty()) {
+        return nullptr;
+    }
+    server_task_result_ptr res = std::move(ticket_results.front());
+    ticket_results.erase(ticket_results.begin());
+    return res;
+}

tools/server/server-queue.h

@@ -164,7 +164,24 @@ struct server_response {
     std::mutex mutex_results;
     std::condition_variable condition_results;
 
+    // Continuum fork (card 7fc7e2ce): ticketed loop mutations. A ticketed task's result goes to the
+    // learning event stream's subscriber, then it is gone (delivery is the drop); with no subscriber
+    // it is dropped, and a core that (re)connects reads the physical state, never a replay.
+    std::unordered_set<int> ticketed_ids;
+    std::vector<server_task_result_ptr> ticket_results;
+    int ticket_subscribers = 0;
+
 public:
+    // register `id_task` as ticketed: its result is delivered on the event stream, not to a waiter
+    void add_ticket(int id_task);
+
+    // the learning event stream opens / closes; the last close drops anything undelivered
+    void subscribe_tickets();
+    void unsubscribe_tickets();
+
+    // blocks until a ticketed result arrives (returned) or `timeout` seconds pass (nullptr: keepalive)
+    server_task_result_ptr next_ticket(int timeout);
+
     // add the id_task to the list of tasks waiting for response
     void add_waiting_task_id(int id_task);
 

tools/server/server.cpp

@@ -224,6 +224,7 @@ int llama_server(common_params & params, int argc, char ** argv) {
         routes.post_lora_adapters_unload   = models_routes->proxy_post;
         routes.post_learn                  = models_routes->proxy_post;
         routes.get_learn_grants            = models_routes->proxy_get;
+        routes.get_learn_events            = models_routes->proxy_get;
         routes.post_learn_grants           = models_routes->proxy_post;
         routes.post_learn_preflight        = models_routes->proxy_post;
         routes.post_learn_residency        = models_routes->proxy_post;
@@ -295,6 +296,7 @@ int llama_server(common_params & params, int argc, char ** argv) {
     ctx_http.post("/learn/residency/revoke",    ex_wrapper(routes.revoke_learn_residency));
     ctx_http.post("/learn/preflight",           ex_wrapper(routes.post_learn_preflight));
     ctx_http.post("/learn/grants",              ex_wrapper(routes.post_learn_grants));
+    ctx_http.get ("/learn/events",              ex_wrapper(routes.get_learn_events));
     ctx_http.post("/learn/grants/revoke",       ex_wrapper(routes.revoke_learn_grant));
     ctx_http.get ("/train",                    ex_wrapper(routes.get_train));
     ctx_http.post("/train",                    ex_wrapper(routes.post_train));

tools/server/tests/learn-tickets.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Real server test; requires an isolated server. Usage: learn-tickets.py URL
+
+What this catches (card 7fc7e2ce): a learning mutation the core must wait out behind the work it
+controls. On a CPU node one prefill chunk ran ~40 s while the core waited 4 s, so every learner
+stop read as unreachable. A ticketed mutation answers at once with its ticket; its terminal result
+is pushed on GET /learn/events (one stream per subscriber, no polling) and then gone. A result
+made while nobody subscribes is dropped, never replayed to a later subscriber: a reconnecting core
+reads the physical state instead.
+Run with --train-dir FIXTURE --no-repack.
+"""
+import json
+import os
+import queue
+import sys
+import threading
+import urllib.request
+
+sys.path.insert(0, os.path.dirname(__file__))
+from learn_fixture import Engine
+
+engine = Engine(sys.argv[1], "tickets", timeout=120)
+
+
+def subscribe():
+    """Open the event stream; its data events land on the returned queue."""
+    events = queue.Queue()
+    stream = urllib.request.urlopen(engine.url + "/learn/events", timeout=engine.timeout)
+
+    def read():
+        try:
+            for raw in stream:
+                line = raw.decode().strip()
+                if line.startswith("data: "):
+                    events.put(json.loads(line[len("data: "):]))
+        except Exception as error:  # the stream closed under us
+            events.put({"closed": str(error)})
+
+    threading.Thread(target=read, daemon=True).start()
+    first = events.get(timeout=engine.timeout)
+    assert first.get("subscribed") is True and first.get("engine") == engine.engine, first
+    return stream, events
+
+
+def ticketed(route, body):
+    reply = engine.ok(route, dict(body, ticket=True))
+    assert set(reply) == {"ticket"}, ("a ticketed mutation answers only its ticket, at once", reply)
+    return reply["ticket"]
+
+
+# with nobody subscribed, a ticketed result is dropped: never replayed later
+gid_unseen = engine.register_grant("-unseen")
+engine.grants.remove(gid_unseen)
+unseen = ticketed("/learn/grants/revoke", {"id": gid_unseen})
+
+stream, events = subscribe()
+
+# a ticketed grant revoke: its terminal result arrives on the stream under its ticket
+gid = engine.register_grant("-revoked")
+engine.grants.remove(gid)
+ticket = ticketed("/learn/grants/revoke", {"id": gid})
+event = events.get(timeout=engine.timeout)
+assert event["ticket"] != unseen, ("a result made while nobody listened is never replayed", event)
+assert event["ticket"] == ticket and not event["error"], event
+assert event["result"]["revoked"] is True, event
+
+# a ticketed unload: the adapter's retirement arrives the same way
+shadow = engine.fresh()
+ticket = ticketed("/lora-adapters/unload", {"id": shadow["id"], "engine": engine.engine})
+event = events.get(timeout=engine.timeout)
+assert event["ticket"] == ticket and not event["error"], event
+engine.shadows.remove(shadow["id"])
+
+stream.close()
+engine.close()
+print("ok: ticketed mutations answer at once and their results arrive on the event stream")