server: fan ticket results out to each event subscriber

tools/server/server-context.cpp

@@ -6605,12 +6605,12 @@ void server_routes::init_routes() {
     this->get_learn_events = [this](const server_http_req & req) {
         struct ticket_stream : server_res_generator {
             server_response & results;
+            const int subscriber;
             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();
-            }
+                : server_res_generator(tasks, results, sleep_idle_seconds, true), results(results),
+                  subscriber(results.subscribe_tickets()) {}
             void on_complete() override {
-                results.unsubscribe_tickets();
+                results.unsubscribe_tickets(subscriber);
                 server_res_generator::on_complete();
             }
         };
@@ -6629,12 +6629,10 @@ void server_routes::init_routes() {
                 res_this->data.clear();
                 return true;
             }
-            auto result = res_this->results.next_ticket(keepalive);
-            if (!result) {
+            output = res_this->results.next_ticket(res_this->subscriber, keepalive);
+            if (output.empty()) {
                 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;

tools/server/server-queue.cpp

@@ -482,9 +482,13 @@ void server_response::send(server_task_result_ptr && result) {
 
     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));
+        // stored before the notify, under the same lock the streams wait with (no lost wakeup)
+        if (!ticket_queues.empty()) {
+            const std::string event = "data: " + safe_json_to_str({
+                {"ticket", result->id}, {"error", result->is_error()}, {"result", result->to_json()}}) + "\n\n";
+            for (auto & [subscriber, queue] : ticket_queues) {
+                queue.push_back(event);
+            }
             condition_results.notify_all();
         }
         return;
@@ -633,26 +637,29 @@ void server_response::add_ticket(int id_task) {
     ticketed_ids.insert(id_task);
 }
 
-void server_response::subscribe_tickets() {
+int server_response::subscribe_tickets() {
     std::unique_lock<std::mutex> lock(mutex_results);
-    ++ticket_subscribers;
+    const int subscriber = next_ticket_subscriber++;
+    ticket_queues[subscriber];
+    return subscriber;
 }
 
-void server_response::unsubscribe_tickets() {
+void server_response::unsubscribe_tickets(int subscriber) {
     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
-    }
+    ticket_queues.erase(subscriber); // what it never read is gone with it
 }
 
-server_task_result_ptr server_response::next_ticket(int timeout) {
+std::string server_response::next_ticket(int subscriber, 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;
+    condition_results.wait_for(lock, std::chrono::seconds(timeout), [&] {
+        auto it = ticket_queues.find(subscriber);
+        return !running || it == ticket_queues.end() || !it->second.empty();
+    });
+    auto it = ticket_queues.find(subscriber);
+    if (!running || it == ticket_queues.end() || it->second.empty()) {
+        return {};
     }
-    server_task_result_ptr res = std::move(ticket_results.front());
-    ticket_results.erase(ticket_results.begin());
-    return res;
+    std::string event = std::move(it->second.front());
+    it->second.pop_front();
+    return event;
 }

tools/server/server-queue.h

@@ -8,6 +8,8 @@
 #include <mutex>
 #include <thread>
 #include <vector>
+#include <string>
+#include <unordered_map>
 #include <unordered_set>
 
 // struct for managing server tasks
@@ -164,23 +166,27 @@ 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.
+    // Continuum fork (card 7fc7e2ce): ticketed loop mutations. A ticketed task's result goes to EVERY
+    // open learning event stream, each through its own queue, 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. One shared queue let a stale stream (its client gone, its socket not yet failed) take the
+    // result a live core was waiting for, and that core waited forever (the IntelMac, 2026-11-11 19:58Z).
+    // The core matches by ticket id, so a copy on a stale stream costs nothing.
     std::unordered_set<int> ticketed_ids;
-    std::vector<server_task_result_ptr> ticket_results;
-    int ticket_subscribers = 0;
+    std::unordered_map<int, std::deque<std::string>> ticket_queues; // per subscriber: formatted events
+    int next_ticket_subscriber = 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();
+    // a learning event stream opens (returns its subscriber id) / closes (drops what it never read)
+    int subscribe_tickets();
+    void unsubscribe_tickets(int subscriber);
 
-    // blocks until a ticketed result arrives (returned) or `timeout` seconds pass (nullptr: keepalive)
-    server_task_result_ptr next_ticket(int timeout);
+    // blocks until an event for `subscriber` arrives (returned, SSE-ready) or `timeout` seconds pass
+    // (empty: keepalive)
+    std::string next_ticket(int subscriber, int timeout);
 
     // add the id_task to the list of tasks waiting for response
     void add_waiting_task_id(int id_task);

tools/server/tests/learn-tickets.py

@@ -54,6 +54,9 @@ def ticketed(route, body):
 unseen = ticketed("/learn/grants/revoke", {"id": gid_unseen})
 
 stream, events = subscribe()
+# a second open stream, as a reconnecting core leaves its predecessor's until its socket fails: each
+# stream gets every result, so a stale one can never take the result a live core waits for
+other, other_events = subscribe()
 
 # a ticketed grant revoke: its terminal result arrives on the stream under its ticket
 gid = engine.register_grant("-revoked")
@@ -63,6 +66,9 @@ def ticketed(route, body):
 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
+copy = other_events.get(timeout=engine.timeout)
+assert copy["ticket"] == ticket, ("every open stream gets the result, never just one of them", copy)
+other.close()
 
 # a ticketed unload: the adapter's retirement arrives the same way
 shadow = engine.fresh()