From 273cc3abc9d1f60394662bea4f502f1c49efdeba Mon Sep 17 00:00:00 2001 From: joelteply Date: Sun, 1 Nov 2026 17:31:46 -0600 Subject: [PATCH 1/2] server: the lost-reply binding: a learn names its execution, base step and engine; a repeat is answered from its receipt, a stale one is fenced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An engine step can commit while its reply or the core's task terminal is lost; a missing completion is not proof the step never ran (Codex on continuum #5695; ONE-RESIDENT-MODEL ยง9.6). Card 4669d845. - Every learn (POST /learn and a completion's learn) names `execution` (the task that owns the step), `base_step` (the shadow's committed step it builds on) and `engine` (the incarnation it was dispatched to). A learn without an execution is refused. - The engine mints its incarnation at start and reports it in /props and on every learn outcome, with the shadow file the step changed. - The fence: a step whose engine is not this one (stale_engine) or whose base_step is not the shadow's committed step (stale_base_step, with the shadow's step in the reply) is refused. So a delayed duplicate, even after its receipt was acknowledged and dropped, or a redispatch racing the original under a new execution id, commits at most once per (shadow generation, step), with no tombstones kept. - Receipts: a committed step's outcome is kept by execution; a learn naming it is answered with it (replayed, nothing steps). POST /learn/receipt reads one execution's state (committed with its receipt, pending behind a busy slot, or unknown to this incarnation); POST /learn/receipts/ack releases receipts the core adopted. Unacknowledged receipts are bounded (refused as receipts_full at the bound, never dropped). - /lora-adapters/load takes `step`: a restored generation (a snapshot) resumes its fence at its snapshot's step. - A learn outcome reports `step` whenever the shadow has one: after this step, or where it stands when refused stale. Tests (IntelMac CPU, Qwen2.5-0.5B, fresh servers): learn-every-format.py --receipts covers replay, a stale base and engine, a missing execution, committed/unknown receipt reads, acknowledgement, and the delayed duplicate after acknowledgement fenced (disabling the base_step check fails it); every existing mode now sends fenced steps (--two-minds, --snapshot, --shadow-retirement and all ten format x stream cases pass). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FRQWzgSfo79JtnwZywHKrE --- tools/server/server-context.cpp | 148 +++++++++++++++++++++++ tools/server/server-context.h | 3 + tools/server/server-schema.cpp | 6 +- tools/server/server-task.cpp | 14 ++- tools/server/server-task.h | 21 ++++ tools/server/server.cpp | 4 + tools/server/tests/learn-every-format.py | 97 ++++++++++----- 7 files changed, 263 insertions(+), 30 deletions(-) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 2521c1a57227..74867d62ebab 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -29,6 +29,8 @@ #include #include #include +#include +#include #include #include @@ -905,6 +907,26 @@ struct server_context_impl { float learn_lr = 1e-4f; // each shadow's committed optimizer steps on this server (a snapshot records where she was) std::map learn_steps; + // This engine's identity for this run: minted at start, reported in /props and on every learn + // receipt, so a receipt from before a restart is never mistaken for this run's. + const std::string engine_incarnation = mint_engine_incarnation(); + // Committed learn steps by the task that owns them (the lost-reply binding): a learn naming an + // execution already here is answered with its receipt and steps nothing. An entry stays until + // the core acknowledges it (POST /learn/receipts/ack). + std::map learn_receipts; + // /learn executions deferred behind a busy slot: in flight, not yet committed or refused + std::set learn_pending; + // A backstop on unacknowledged receipts, not a tuned rate: the core acknowledges each receipt as + // it adopts it, so the table holds only steps whose adoption is in flight; at the bound, new + // learns are refused (receipts_full) rather than receipts dropped. ~200 bytes each. + static constexpr size_t LEARN_RECEIPTS_MAX = 4096; + + static std::string mint_engine_incarnation() { + std::random_device rd; + char buf[33]; + snprintf(buf, sizeof(buf), "%08x%08x%08x%08x", rd(), rd(), rd(), rd()); + return buf; + } // LEARNING BARELY TOUCHES THE FLOW: while any other slot is working, learning may take at most // LEARN_DUTY of the loop's time (the step after a learn of t ms waits t*(1/LEARN_DUTY - 1) ms); // with every other slot idle it runs at once. Back to back, learns took the 4B's other slot from @@ -2069,6 +2091,43 @@ struct server_context_impl { // completion's own learn before its slot is released): the slot's stable set plus the shadow on // the reply's labelled rows, then the slot back to its whole served prompt. learn_outcome learn_on_slot(server_slot & slot, const learn_request & L, int64_t t0) { + // the lost-reply binding: every step names the task that owns it, and a step already + // committed under that name is answered from its receipt, never stepped twice + if (L.execution.empty()) { + learn_outcome out; + out.failure = "learn needs \"execution\": the id of the task that owns this step (its receipt's key)"; + return out; + } + if (const auto it = learn_receipts.find(L.execution); it != learn_receipts.end()) { + learn_outcome out = it->second; + out.replayed = true; + return out; + } + // The fence: a step builds on exactly the shadow generation and committed step the caller + // saw, on this engine. A delayed duplicate (its receipt already acknowledged and gone), or a + // redispatch racing the original under a new execution id, finds the shadow past its base and + // is refused: at most one step commits per (shadow generation, step), with no tombstones kept. + if (L.engine != engine_incarnation) { + learn_outcome out; + out.failure = "stale_engine: dispatched to engine " + L.engine + ", this is " + engine_incarnation; + return out; + } + if (L.lora_id >= 0 && (size_t) L.lora_id < params_base.lora_adapters.size() && params_base.lora_adapters[L.lora_id].ptr) { + const auto at = learn_steps.find(params_base.lora_adapters[L.lora_id].ptr); + const int64_t current = at == learn_steps.end() ? 0 : at->second; + if (L.base_step != current) { + learn_outcome out; + out.failure = "stale_base_step: the shadow is at committed step " + std::to_string(current) + + ", the step was dispatched on " + std::to_string(L.base_step); + out.step = current; + return out; + } + } + if (learn_receipts.size() >= LEARN_RECEIPTS_MAX) { + learn_outcome out; + out.failure = "receipts_full: acknowledge adopted receipts (POST /learn/receipts/ack) before learning more"; + return out; + } for (const auto & other : slots) { if (other.id != slot.id && other.is_processing() && t0 < learn_next_us) { learn_outcome out; @@ -2079,6 +2138,12 @@ struct server_context_impl { } } auto out = learn_on_slot_admitted(slot, L, t0); + out.execution = L.execution; + out.engine_incarnation = engine_incarnation; + if (out.ok) { + out.shadow = params_base.lora_adapters[L.lora_id].path; + learn_receipts.emplace(L.execution, out); + } const int64_t finished = ggml_time_us(); learn_next_us = finished + (int64_t) ((finished - t0) * (1.0/LEARN_DUTY - 1.0)); return out; @@ -3137,6 +3202,9 @@ struct server_context_impl { llama_adapter_meta_val_str(la.ptr, "adapter.lora.prompt_prefix", buf, sizeof(buf)); la.prompt_prefix = buf; params_base.lora_adapters.push_back(la); + if (task.load_lora_step > 0) { + learn_steps[la.ptr] = task.load_lora_step; // a restored generation resumes its fence at its snapshot's step + } SRV_INF("loaded lora adapter id=%zu path=%s (dormant)\n", params_base.lora_adapters.size() - 1, la.path.c_str()); auto res = std::make_unique(); res->id = task.id; @@ -3163,12 +3231,44 @@ struct server_context_impl { } if (slot->is_processing()) { // her slot is serving: learn between its turns, never during one + if (!L.execution.empty()) { + learn_pending.insert(L.execution); + } queue_tasks.defer(std::move(task)); break; } + learn_pending.erase(L.execution); res->outcome = learn_on_slot(*slot, L, t0); queue_results.send(std::move(res)); } break; + case SERVER_TASK_TYPE_LEARN_RECEIPT: + { + // What became of one execution, from the loop that owns the receipts: committed + // (its receipt), pending (deferred, not yet run), or neither in THIS incarnation. + const std::string & execution = task.learn_executions.at(0); + auto res = std::make_unique(); + res->id = task.id; + res->body = { { "execution", execution }, { "engine_incarnation", engine_incarnation } }; + if (const auto it = learn_receipts.find(execution); it != learn_receipts.end()) { + res->body["state"] = "committed"; + res->body["receipt"] = it->second.to_json(); + } else { + res->body["state"] = learn_pending.count(execution) ? "pending" : "unknown"; + } + queue_results.send(std::move(res)); + } break; + case SERVER_TASK_TYPE_LEARN_RECEIPTS_ACK: + { + // The core adopted these steps: their receipts may go. + size_t released = 0; + for (const auto & execution : task.learn_executions) { + released += learn_receipts.erase(execution); + } + auto res = std::make_unique(); + res->id = task.id; + res->body = { { "released", released }, { "outstanding", learn_receipts.size() } }; + queue_results.send(std::move(res)); + } break; case SERVER_TASK_TYPE_SNAPSHOT_LORA: { // Between steps on the loop, so the file is one whole step's weights, never a @@ -4826,6 +4926,7 @@ server_context_meta server_context::get_meta() const { return server_context_meta { /* build_info */ std::string(llama_build_info()), + /* engine_incarnation */ impl->engine_incarnation, /* model_name */ impl->model_name, /* model_aliases */ impl->model_aliases, /* model_tags */ impl->model_tags, @@ -5326,6 +5427,7 @@ static json get_res_props(const server_context_meta & meta, const common_params { "bos_token", meta.bos_token_str }, { "eos_token", meta.eos_token_str }, { "build_info", meta.build_info }, + { "engine_incarnation", meta.engine_incarnation }, { "is_sleeping", is_sleeping }, { "cors_proxy_enabled", params.ui_mcp_proxy }, }; @@ -6033,6 +6135,7 @@ void server_routes::init_routes() { task.id = rd.get_new_id(); task.load_lora = adapter; task.load_lora_path = path; + task.load_lora_step = std::max(0, json_value(body, "step", (int64_t) 0)); rd.post_task(std::move(task)); } auto result = rd.next(req.should_stop); @@ -6068,6 +6171,9 @@ void server_routes::init_routes() { task.learn.n_prefix = json_value(body, "n_prefix", 0); task.learn.lora_id = json_value(body, "lora_id", -1); task.learn.lr = json_value(body, "lr", 1e-4f); + task.learn.execution = json_value(body, "execution", std::string()); + task.learn.base_step = json_value(body, "base_step", (int64_t) -1); + task.learn.engine = json_value(body, "engine", std::string()); if (body.contains("messages")) { // the conversation she was served, its last message her reply (or the correction). The // served prompt is every earlier message through the SAME parser and template arguments @@ -6154,6 +6260,48 @@ void server_routes::init_routes() { return res; }; + // the lost-reply binding's read and its release: one execution's state, and receipts the core adopted + auto learn_receipts_task = [this](const server_http_req & req, server_task_type type, std::vector executions) { + auto res = create_response(); + auto & rd = res->rd; + { + server_task task(type); + task.id = rd.get_new_id(); + task.learn_executions = std::move(executions); + rd.post_task(std::move(task)); + } + auto result = rd.next(req.should_stop); + if (!result) { + GGML_ASSERT(req.should_stop()); + return res; + } + if (result->is_error()) { + res->error(result->to_json()); + return res; + } + res->ok(result->to_json()); + return res; + }; + this->post_learn_receipt = [this, learn_receipts_task](const server_http_req & req) { + const json body = json::parse(req.body); + const std::string execution = json_value(body, "execution", std::string()); + if (execution.empty()) { + auto res = create_response(); + res->error(format_error_response("\"execution\" (the task id a learn named) is required", ERROR_TYPE_INVALID_REQUEST)); + return res; + } + return learn_receipts_task(req, SERVER_TASK_TYPE_LEARN_RECEIPT, { execution }); + }; + this->post_learn_receipts_ack = [this, learn_receipts_task](const server_http_req & req) { + const json body = json::parse(req.body); + if (!body.contains("executions") || !body.at("executions").is_array()) { + auto res = create_response(); + res->error(format_error_response("\"executions\" (the adopted task ids) is required", ERROR_TYPE_INVALID_REQUEST)); + return res; + } + return learn_receipts_task(req, SERVER_TASK_TYPE_LEARN_RECEIPTS_ACK, body.at("executions").get>()); + }; + this->post_lora_adapters_snapshot = [this](const server_http_req & req) { auto res = create_response(); const json body = json::parse(req.body); diff --git a/tools/server/server-context.h b/tools/server/server-context.h index fd079a69cb86..10a098756605 100644 --- a/tools/server/server-context.h +++ b/tools/server/server-context.h @@ -15,6 +15,7 @@ struct server_context_impl; // private implementation struct server_context_meta { std::string build_info; + std::string engine_incarnation; std::string model_name; std::set model_aliases; std::set model_tags; @@ -170,6 +171,8 @@ struct server_routes { server_http_context::handler_t post_lora_adapters_load; server_http_context::handler_t post_lora_adapters_unload; server_http_context::handler_t post_lora_adapters_snapshot; + server_http_context::handler_t post_learn_receipt; + server_http_context::handler_t post_learn_receipts_ack; server_http_context::handler_t post_learn; server_http_context::handler_t get_train; server_http_context::handler_t post_train; diff --git a/tools/server/server-schema.cpp b/tools/server/server-schema.cpp index 828eff7285d3..11105e87f3d1 100644 --- a/tools/server/server-schema.cpp +++ b/tools/server/server-schema.cpp @@ -546,7 +546,8 @@ task_params eval_llama_cmpl_schema( f->eval(ctx, data); } - // learn a reply on this request's slot before it is released: {"reply": text, "lora_id", "lr"}; + // learn a reply on this request's slot before it is released: {"reply": text, "lora_id", "lr", + // "execution", "base_step", "engine"} (the step's identity and fence, required); // the reply is learned as the tokens the model would generate after this prompt (its text, then // the end of the turn) if (data.contains("learn")) { @@ -557,6 +558,9 @@ task_params eval_llama_cmpl_schema( learn_request L; L.lora_id = json_value(l, "lora_id", -1); L.lr = json_value(l, "lr", 1e-4f); + L.execution = json_value(l, "execution", std::string()); + L.base_step = json_value(l, "base_step", (int64_t) -1); + L.engine = json_value(l, "engine", std::string()); L.reply = common_tokenize(vocab, l.at("reply").get(), false, true); L.reply.push_back(llama_vocab_eot(vocab) != LLAMA_TOKEN_NULL ? llama_vocab_eot(vocab) : llama_vocab_eos(vocab)); params.learn = std::move(L); diff --git a/tools/server/server-task.cpp b/tools/server/server-task.cpp index 39eeee6e7060..66044eb7fe32 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp @@ -1727,8 +1727,18 @@ json server_task_result_lora_change::to_json() { json learn_outcome::to_json() const { json res = { { "success", ok }, { "loss", loss }, { "n_labels", n_labels }, { "t_ms", t_ms } }; - if (ok) { - res["step"] = step; + if (step > 0) { + res["step"] = step; // the shadow's committed step: after this step, or (refused stale) where it stands + } + if (!execution.empty()) { + res["execution"] = execution; + res["engine_incarnation"] = engine_incarnation; + } + if (!shadow.empty()) { + res["shadow"] = shadow; + } + if (replayed) { + res["replayed"] = true; } if (deferred) { res["deferred"] = true; diff --git a/tools/server/server-task.h b/tools/server/server-task.h index 584551c7801a..8fb141bf0368 100644 --- a/tools/server/server-task.h +++ b/tools/server/server-task.h @@ -61,6 +61,8 @@ enum server_task_type { SERVER_TASK_TYPE_LOAD_LORA, SERVER_TASK_TYPE_UNLOAD_LORA, SERVER_TASK_TYPE_SNAPSHOT_LORA, + SERVER_TASK_TYPE_LEARN_RECEIPT, + SERVER_TASK_TYPE_LEARN_RECEIPTS_ACK, SERVER_TASK_TYPE_LEARN, }; @@ -92,6 +94,9 @@ struct learn_request { std::vector prompt; // the served prompt, when known: her slot must hold exactly it int32_t lora_id = -1; // the trainable adapter (the shadow), loaded dormant float lr = 1e-4f; + std::string execution; // the task that owns this step: its receipt's key (required) + int64_t base_step = -1; // the shadow's committed step this step builds on (required): the fence + std::string engine; // the engine incarnation the caller dispatched to (required) }; struct learn_outcome { @@ -104,6 +109,12 @@ struct learn_outcome { double t_ms = 0.0; // the shadow's committed optimizer steps after this one (0 when it did not step) int64_t step = 0; + // the step's identity: the task that owns it, the engine incarnation that ran it, the shadow + // file it changed; replayed = this execution had already committed and nothing stepped now + std::string execution; + std::string engine_incarnation; + std::string shadow; + bool replayed = false; json to_json() const; }; @@ -263,10 +274,14 @@ struct server_task { // server loop; the loop only appends it (dormant, scale 0) to the adapter list llama_adapter_lora * load_lora = nullptr; std::string load_lora_path; + int64_t load_lora_step = 0; // a restored generation's committed step (its snapshot's), seeding its fence // used by SERVER_TASK_TYPE_UNLOAD_LORA: the adapter id to retire int32_t unload_lora = -1; + // used by SERVER_TASK_TYPE_LEARN_RECEIPT (one id) and SERVER_TASK_TYPE_LEARN_RECEIPTS_ACK (many) + std::vector learn_executions; + // used by SERVER_TASK_TYPE_SNAPSHOT_LORA: the adapter id to write, and where (confined to --train-dir) int32_t snapshot_lora = -1; std::string snapshot_path; @@ -685,6 +700,12 @@ struct server_task_result_lora_snapshot : server_task_result { } }; +// A receipt lookup or acknowledgement, answered by the loop that owns the receipts. +struct server_task_result_json : server_task_result { + json body; + virtual json to_json() override { return body; } +}; + struct server_task_result_learn : server_task_result { learn_outcome outcome; virtual json to_json() override { return outcome.to_json(); } diff --git a/tools/server/server.cpp b/tools/server/server.cpp index 1fb74cb30e2e..45810d88b846 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -220,6 +220,8 @@ int llama_server(common_params & params, int argc, char ** argv) { routes.post_lora_adapters_load = models_routes->proxy_post; routes.post_lora_adapters_unload = models_routes->proxy_post; routes.post_lora_adapters_snapshot = models_routes->proxy_post; + routes.post_learn_receipt = models_routes->proxy_post; + routes.post_learn_receipts_ack = models_routes->proxy_post; routes.post_learn = models_routes->proxy_post; routes.get_slots = models_routes->proxy_get; routes.post_slots = models_routes->proxy_post; @@ -276,6 +278,8 @@ int llama_server(common_params & params, int argc, char ** argv) { ctx_http.post("/lora-adapters/load", ex_wrapper(routes.post_lora_adapters_load)); ctx_http.post("/lora-adapters/unload", ex_wrapper(routes.post_lora_adapters_unload)); ctx_http.post("/lora-adapters/snapshot", ex_wrapper(routes.post_lora_adapters_snapshot)); + ctx_http.post("/learn/receipt", ex_wrapper(routes.post_learn_receipt)); + ctx_http.post("/learn/receipts/ack", ex_wrapper(routes.post_learn_receipts_ack)); ctx_http.post("/learn", ex_wrapper(routes.post_learn)); ctx_http.get ("/train", ex_wrapper(routes.get_train)); ctx_http.post("/train", ex_wrapper(routes.post_train)); diff --git a/tools/server/tests/learn-every-format.py b/tools/server/tests/learn-every-format.py index b0fdd9898099..f7d601b089f5 100755 --- a/tools/server/tests/learn-every-format.py +++ b/tools/server/tests/learn-every-format.py @@ -4,18 +4,30 @@ # served model), e.g.: llama-server -m MODEL --port 58199 -np 1 --jinja --train-dir DIR # what this catches: a serializer, or a request conversion (the Anthropic one dropped "learn"), that # loses the outcome, so a client learns with nothing on the wire -import json, os, sys, threading, time, urllib.request +import json, os, sys, threading, time, urllib.request, uuid U = os.environ.get("LLAMA_SERVER_URL", "http://localhost:58199") def raw(path, body): req = urllib.request.Request(U + path, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=300) as r: return r.read().decode() +with urllib.request.urlopen(U + "/props", timeout=300) as r: + props = json.load(r) if "--shadow-retirement" in sys.argv: - with urllib.request.urlopen(U + "/props", timeout=300) as r: - props = json.load(r) # No competing lane: the duty gate defers only while another slot is processing. assert props["total_slots"] == 1, "shadow-retirement requires an isolated single-slot server" +ENGINE = props["engine_incarnation"] sid = json.loads(raw("/lora-adapters/load", {"name": "shadow-05.gguf"}))["id"] -learn = {"reply": "Topeka is the capital of Kansas.", "lora_id": sid, "lr": 0.001} +_steps = {} # each shadow's committed step, as this client last saw it: the base of its next step +def fenced(lora, **extra): + # one step's identity and fence: a fresh execution, the shadow's committed step it builds on, + # the engine it is dispatched to. A retry of a deferred step re-sends this same dict. + step = dict(reply="Topeka is the capital of Kansas.", lora_id=lora, lr=0.001, execution=str(uuid.uuid4()), + base_step=_steps.get(lora, 0), engine=ENGINE) + step.update(extra) # a test may pin a stale base or engine + return step +def landed(lora, outcome): + if outcome.get("success"): + _steps[lora] = outcome["step"] + return outcome # Isolated adapter lifetime regression, with no live persona or other clients on this server. if "--shadow-retirement" in sys.argv: @@ -23,16 +35,14 @@ def raw(path, body): retired = json.loads(raw("/lora-adapters/unload", {"id": sid})) assert retired["success"] and retired["retired"] and retired["id"] == sid, retired sid = json.loads(raw("/lora-adapters/load", {"name": "shadow-05.gguf"}))["id"] - learn["lora_id"] = sid - turn = {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} - first = json.loads(raw("/completion", turn)) - assert first["learn"]["success"], first + turn = lambda: {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": fenced(sid)} + first = json.loads(raw("/completion", turn())) + assert landed(sid, first["learn"])["success"], first # a shadow that learned unloads too: its optimizer is released first (its moments go) bound = json.loads(raw("/lora-adapters/unload", {"id": sid})) assert bound["success"] and bound["retired"] and bound["id"] == sid, bound sid = json.loads(raw("/lora-adapters/load", {"name": "shadow-05.gguf"}))["id"] - learn["lora_id"] = sid - following = json.loads(raw("/completion", turn)) + following = json.loads(raw("/completion", turn())) assert following["learn"]["success"], following print(json.dumps({"unbound_retired": retired, "bound_retired": bound, "next_learn": following["learn"]})) sys.exit(0) @@ -41,8 +51,8 @@ def raw(path, body): # (fresh server; writes snap-test.gguf into --train-dir, so remove it before a rerun). if "--snapshot" in sys.argv: import urllib.error - turn = {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} - steps = [json.loads(raw("/completion", turn))["learn"] for _ in range(2)] + turn = lambda lora: {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": fenced(lora)} + steps = [landed(sid, json.loads(raw("/completion", turn(sid)))["learn"]) for _ in range(2)] assert [st["success"] for st in steps] == [True, True] and [st["step"] for st in steps] == [1, 2], steps snap = json.loads(raw("/lora-adapters/snapshot", {"id": sid, "name": "snap-test.gguf"})) assert snap["success"] and snap["steps"] == 2 and snap["bytes"] > 0, snap @@ -52,14 +62,44 @@ def raw(path, body): except urllib.error.HTTPError as error: detail = error.read().decode() assert error.code == 400 and "immutable" in detail, (error.code, detail) - third = json.loads(raw("/completion", turn))["learn"] + third = landed(sid, json.loads(raw("/completion", turn(sid)))["learn"]) # a learn's loss is its forward pass before its update: the snapshot, loaded, sees what the shadow saw loaded = json.loads(raw("/lora-adapters/load", {"name": "snap-test.gguf"}))["id"] - again = json.loads(raw("/completion", dict(turn, learn=dict(learn, lora_id=loaded))))["learn"] + again = landed(loaded, json.loads(raw("/completion", turn(loaded)))["learn"]) assert again["success"] and abs(again["loss"] - third["loss"]) < 1e-5, (third, again) print(json.dumps({"steps": steps, "snapshot": snap, "third": third, "from_snapshot": again})) sys.exit(0) +# The lost-reply binding (fresh single-slot server): a step names its execution, base step and engine; +# a repeat is answered from its receipt; a stale base or engine is refused; after the core acknowledges +# the receipt, a delayed duplicate is fenced by the shadow's step, never stepped twice. +if "--receipts" in sys.argv: + turn = lambda step: {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": step} + first_step = fenced(sid) + first = landed(sid, json.loads(raw("/completion", turn(first_step)))["learn"]) + assert first["success"] and first["step"] == 1 and first["execution"] == first_step["execution"], first + assert first["engine_incarnation"] == ENGINE and first["shadow"].endswith("shadow-05.gguf"), first + replay = json.loads(raw("/completion", turn(first_step)))["learn"] + assert replay["replayed"] is True and replay["step"] == 1 and replay["loss"] == first["loss"], replay + state = json.loads(raw("/learn/receipt", {"execution": first_step["execution"]})) + assert state["state"] == "committed" and state["engine_incarnation"] == ENGINE, state + assert json.loads(raw("/learn/receipt", {"execution": str(uuid.uuid4())}))["state"] == "unknown" + stale_base = json.loads(raw("/completion", turn(fenced(sid, base_step=0))))["learn"] + assert not stale_base["success"] and stale_base["failure"].startswith("stale_base_step") and stale_base["step"] == 1, stale_base + stale_engine = json.loads(raw("/completion", turn(fenced(sid, engine="another-run"))))["learn"] + assert not stale_engine["success"] and stale_engine["failure"].startswith("stale_engine"), stale_engine + unnamed = {k: v for k, v in fenced(sid).items() if k != "execution"} + assert "execution" in json.loads(raw("/completion", turn(unnamed)))["learn"]["failure"] + ack = json.loads(raw("/learn/receipts/ack", {"executions": [first_step["execution"]]})) + assert ack["released"] == 1 and ack["outstanding"] == 0, ack + # its receipt is gone: the delayed duplicate is refused by the shadow's step, not run again + late = json.loads(raw("/completion", turn(first_step)))["learn"] + assert not late["success"] and late["failure"].startswith("stale_base_step"), late + nxt = landed(sid, json.loads(raw("/completion", turn(fenced(sid))))["learn"]) + assert nxt["success"] and nxt["step"] == 2, nxt + print(json.dumps({"first": first, "replay": replay, "stale_base": stale_base["failure"], "late_duplicate": late["failure"], "next": nxt["step"]})) + sys.exit(0) + # Two minds on one serving context: each shadow keeps its own optimizer while the other learns in # between (needs shadow-05.gguf and a copy of it, shadow-06.gguf, in --train-dir). if "--two-minds" in sys.argv: @@ -67,9 +107,9 @@ def raw(path, body): steps = [] for lora in [sid, other, sid, other]: turn = {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, - "learn": dict(learn, lora_id=lora)} + "learn": fenced(lora)} out = json.loads(raw("/completion", turn)) - assert out["learn"]["success"], out + assert landed(lora, out["learn"])["success"], out steps.append((lora, out["learn"]["loss"])) # the two shadows start from the same file, so they are independent only if their trajectories # match: the other mind's first step is untouched by this one's step, and each second step @@ -95,7 +135,9 @@ def raw(path, body): request = {"messages": messages, "tools": tools, "id_slot": 0, "max_tokens": 1, "chat_template_kwargs": {"enable_thinking": bool(reply.get("reasoning_content"))}} raw("/v1/chat/completions", request) - experience = dict(request, messages=messages + [reply], lora_id=sid) + step = fenced(sid) + experience = dict(request, messages=messages + [reply], lora_id=sid, lr=step["lr"], + execution=step["execution"], base_step=step["base_step"], engine=ENGINE) expect_refusal = bool(reply.get("reasoning_content")) and "--expect-reasoning-refusal" in sys.argv try: outcome = json.loads(raw("/learn", experience)) @@ -113,10 +155,11 @@ def raw(path, body): time.sleep(outcome["retry_after_ms"] / 1000) # Retain and replay the whole experience; no intervening turn changes this slot. outcome = json.loads(raw("/learn", experience)) - assert outcome["success"] and outcome["n_labels"] > 1, outcome + assert landed(sid, outcome)["success"] and outcome["n_labels"] > 1, outcome print(json.dumps({"template_reply": "tool_calls" if "tool_calls" in reply else "reasoning" if reply.get("reasoning_content") else "text", "outcome": outcome})) try: - raw("/learn", {"messages": messages + [{"role": "assistant", "content": ""}], "id_slot": 0, "lora_id": sid}) + raw("/learn", {"messages": messages + [{"role": "assistant", "content": ""}], "id_slot": 0, "lora_id": sid, + "execution": str(uuid.uuid4()), "base_step": _steps.get(sid, 0), "engine": ENGINE}) raise AssertionError("empty assistant reply was accepted") except urllib.error.HTTPError as error: assert error.code == 400, error @@ -126,7 +169,7 @@ def raw(path, body): # exercises the shared admission in both directions, using the same model/adapter. if "--fairness" in sys.argv: prompt = "Q: Which city is the capital of Kansas?\nA:" - turn = {"prompt": prompt, "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} + turn = lambda: {"prompt": prompt, "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": fenced(sid)} started, stop = threading.Event(), threading.Event() progress, errors = [], [] @@ -158,25 +201,25 @@ def assert_busy(): def assert_deferred(): assert_busy() - result = json.loads(raw("/completion", turn))["learn"] + result = json.loads(raw("/completion", turn()))["learn"] assert result["success"] is False and result["deferred"] is True, result assert result["retry_after_ms"] > 0 and result["n_labels"] == 0, result return result assert_busy() - first = json.loads(raw("/completion", turn)) - assert first["learn"]["success"], first + first = json.loads(raw("/completion", turn())) + assert landed(sid, first["learn"])["success"], first completion_rest = assert_deferred() before = len(progress) t0 = time.monotonic() - post_body = dict(learn, id_slot=0, n_prefix=first["tokens_evaluated"]) + post_body = fenced(sid, id_slot=0, n_prefix=first["tokens_evaluated"]) post_deferred = json.loads(raw("/learn", post_body)) assert post_deferred["success"] is False and post_deferred["deferred"] is True, post_deferred assert post_deferred["retry_after_ms"] > 0 and post_deferred["n_labels"] == 0, post_deferred # The client owns this experience during rest; the engine holds neither request nor slot. time.sleep(post_deferred["retry_after_ms"] / 1000) assert_busy() - post = json.loads(raw("/learn", post_body)) + post = landed(sid, json.loads(raw("/learn", post_body))) post_wall_ms = (time.monotonic() - t0) * 1000 assert post["success"], post post_rest = assert_deferred() @@ -197,7 +240,7 @@ def assert_deferred(): fails = 0 for path, body in cases: for stream in (False, True): - b = dict(body, stream=stream, learn=learn, temperature=0) + b = dict(body, stream=stream, learn=fenced(sid), temperature=0) b.setdefault("max_tokens", 2); b["n_predict"] = 2 text = raw(path, b) found = [] @@ -211,7 +254,7 @@ def walk(o): elif isinstance(o, list): for v in o: walk(v) walk(json.loads(line)) - ok = len(found) == 1 and found[0].get("success") is True + ok = len(found) == 1 and landed(sid, found[0]).get("success") is True fails += not ok print(f"{'OK ' if ok else 'BAD'} {path:22s} stream={stream!s:5s} learn outcomes={len(found)} {found[0] if found else ''}") print("ALL FORMATS CARRY THE LEARN OUTCOME" if fails == 0 else f"{fails} FORMAT(S) WITHOUT IT") From 12a0de7d4170a5759918f8487fce3950f323590e Mon Sep 17 00:00:00 2001 From: joelteply Date: Sun, 1 Nov 2026 17:50:58 -0600 Subject: [PATCH 2/2] server: lost-reply binding, round 2 (Codex at 273cc3abc): a receipt replays only the same step, the shadow generation is explicit, input is bound - The engine check runs first, so it applies to a replay as well. - A learn names `input` (the revision it learns, required). A receipt stores the step's full binding: engine, generation, base_step, input. An execution re-sent with any of them changed is refused (execution_rebound) and steps nothing; only the same step replays. - The shadow generation is its adapter load id, which the server never reuses: an unload and restore is a new generation, and a request bound to a retired generation finds no loaded adapter and is refused. Every outcome reports its generation, base_step and input. - /learn/receipt's `unknown` is documented as not proof of never-committed (acknowledgement deletes receipts); redispatch rests on the fence, never on that absence. - Tests (fresh servers, Qwen2.5-0.5B, IntelMac CPU): --receipts adds the same execution re-sent with a changed input, base and generation (each refused; disabling the binding check fails it), and unload/restore: the old generation's delayed request after acknowledgement is refused, the restored generation resumes its fence at its snapshot's step (step 3 from 2). Every other mode and all ten formats pass with the new field. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FRQWzgSfo79JtnwZywHKrE --- tools/server/server-context.cpp | 46 ++++++++++++++++-------- tools/server/server-schema.cpp | 3 +- tools/server/server-task.cpp | 3 ++ tools/server/server-task.h | 4 +++ tools/server/tests/learn-every-format.py | 27 +++++++++++--- 5 files changed, 63 insertions(+), 20 deletions(-) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 74867d62ebab..0baf13fff580 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -2091,27 +2091,39 @@ struct server_context_impl { // completion's own learn before its slot is released): the slot's stable set plus the shadow on // the reply's labelled rows, then the slot back to its whole served prompt. learn_outcome learn_on_slot(server_slot & slot, const learn_request & L, int64_t t0) { - // the lost-reply binding: every step names the task that owns it, and a step already - // committed under that name is answered from its receipt, never stepped twice - if (L.execution.empty()) { + // The lost-reply binding: every step names the task that owns it, the engine it was + // dispatched to, the shadow generation (its adapter load id: never reused, so an unload and + // restore is a new generation), the committed step it builds on and the input it learns. + if (L.execution.empty() || L.input.empty()) { learn_outcome out; - out.failure = "learn needs \"execution\": the id of the task that owns this step (its receipt's key)"; + out.failure = "learn needs \"execution\" (the id of the task that owns this step) and \"input\" (the revision it learns)"; return out; } - if (const auto it = learn_receipts.find(L.execution); it != learn_receipts.end()) { - learn_outcome out = it->second; - out.replayed = true; - return out; - } - // The fence: a step builds on exactly the shadow generation and committed step the caller - // saw, on this engine. A delayed duplicate (its receipt already acknowledged and gone), or a - // redispatch racing the original under a new execution id, finds the shadow past its base and - // is refused: at most one step commits per (shadow generation, step), with no tombstones kept. if (L.engine != engine_incarnation) { learn_outcome out; out.failure = "stale_engine: dispatched to engine " + L.engine + ", this is " + engine_incarnation; return out; } + // A committed execution is answered from its receipt only when the request is the same + // step: the same generation, base and input. Anything else under that name is refused, never + // handed an unrelated success. + if (const auto it = learn_receipts.find(L.execution); it != learn_receipts.end()) { + const learn_outcome & r = it->second; + if (r.generation != L.lora_id || r.base_step != L.base_step || r.input != L.input) { + learn_outcome out; + out.failure = "execution_rebound: execution " + L.execution + " committed a different step (generation " + + std::to_string(r.generation) + ", base " + std::to_string(r.base_step) + ")"; + return out; + } + learn_outcome out = r; + out.replayed = true; + return out; + } + // The fence: a step builds on exactly the committed step the caller saw, of a generation that + // is still loaded. A delayed duplicate (its receipt already acknowledged and gone), or a + // redispatch racing the original under a new execution id, finds the shadow past its base and + // is refused: at most one step commits per (generation, step), with no tombstones kept. A + // request bound to a retired generation finds no loaded adapter and is refused below. if (L.lora_id >= 0 && (size_t) L.lora_id < params_base.lora_adapters.size() && params_base.lora_adapters[L.lora_id].ptr) { const auto at = learn_steps.find(params_base.lora_adapters[L.lora_id].ptr); const int64_t current = at == learn_steps.end() ? 0 : at->second; @@ -2140,6 +2152,9 @@ struct server_context_impl { auto out = learn_on_slot_admitted(slot, L, t0); out.execution = L.execution; out.engine_incarnation = engine_incarnation; + out.generation = L.lora_id; + out.base_step = L.base_step; + out.input = L.input; if (out.ok) { out.shadow = params_base.lora_adapters[L.lora_id].path; learn_receipts.emplace(L.execution, out); @@ -3244,7 +3259,9 @@ struct server_context_impl { case SERVER_TASK_TYPE_LEARN_RECEIPT: { // What became of one execution, from the loop that owns the receipts: committed - // (its receipt), pending (deferred, not yet run), or neither in THIS incarnation. + // (its receipt), pending (deferred, not yet run), or unknown. Unknown is NOT proof + // it never committed: acknowledgement deletes receipts. Redispatch rests on the + // fence (stale_base_step / a retired generation), never on this absence. const std::string & execution = task.learn_executions.at(0); auto res = std::make_unique(); res->id = task.id; @@ -6174,6 +6191,7 @@ void server_routes::init_routes() { task.learn.execution = json_value(body, "execution", std::string()); task.learn.base_step = json_value(body, "base_step", (int64_t) -1); task.learn.engine = json_value(body, "engine", std::string()); + task.learn.input = json_value(body, "input", std::string()); if (body.contains("messages")) { // the conversation she was served, its last message her reply (or the correction). The // served prompt is every earlier message through the SAME parser and template arguments diff --git a/tools/server/server-schema.cpp b/tools/server/server-schema.cpp index 11105e87f3d1..f48399832b4a 100644 --- a/tools/server/server-schema.cpp +++ b/tools/server/server-schema.cpp @@ -547,7 +547,7 @@ task_params eval_llama_cmpl_schema( } // learn a reply on this request's slot before it is released: {"reply": text, "lora_id", "lr", - // "execution", "base_step", "engine"} (the step's identity and fence, required); + // "execution", "base_step", "engine", "input"} (the step's identity and fence, required); // the reply is learned as the tokens the model would generate after this prompt (its text, then // the end of the turn) if (data.contains("learn")) { @@ -561,6 +561,7 @@ task_params eval_llama_cmpl_schema( L.execution = json_value(l, "execution", std::string()); L.base_step = json_value(l, "base_step", (int64_t) -1); L.engine = json_value(l, "engine", std::string()); + L.input = json_value(l, "input", std::string()); L.reply = common_tokenize(vocab, l.at("reply").get(), false, true); L.reply.push_back(llama_vocab_eot(vocab) != LLAMA_TOKEN_NULL ? llama_vocab_eot(vocab) : llama_vocab_eos(vocab)); params.learn = std::move(L); diff --git a/tools/server/server-task.cpp b/tools/server/server-task.cpp index 66044eb7fe32..42217d96b72b 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp @@ -1733,6 +1733,9 @@ json learn_outcome::to_json() const { if (!execution.empty()) { res["execution"] = execution; res["engine_incarnation"] = engine_incarnation; + res["generation"] = generation; + res["base_step"] = base_step; + res["input"] = input; } if (!shadow.empty()) { res["shadow"] = shadow; diff --git a/tools/server/server-task.h b/tools/server/server-task.h index 8fb141bf0368..40107fdf7c99 100644 --- a/tools/server/server-task.h +++ b/tools/server/server-task.h @@ -97,6 +97,7 @@ struct learn_request { std::string execution; // the task that owns this step: its receipt's key (required) int64_t base_step = -1; // the shadow's committed step this step builds on (required): the fence std::string engine; // the engine incarnation the caller dispatched to (required) + std::string input; // the input revision it learns (the experience id and cursor; required) }; struct learn_outcome { @@ -114,6 +115,9 @@ struct learn_outcome { std::string execution; std::string engine_incarnation; std::string shadow; + int32_t generation = -1; // the shadow generation: its adapter load id, never reused + int64_t base_step = -1; // the committed step this step built on + std::string input; // the input revision it learned bool replayed = false; json to_json() const; }; diff --git a/tools/server/tests/learn-every-format.py b/tools/server/tests/learn-every-format.py index f7d601b089f5..17721774c87f 100755 --- a/tools/server/tests/learn-every-format.py +++ b/tools/server/tests/learn-every-format.py @@ -21,7 +21,7 @@ def fenced(lora, **extra): # one step's identity and fence: a fresh execution, the shadow's committed step it builds on, # the engine it is dispatched to. A retry of a deferred step re-sends this same dict. step = dict(reply="Topeka is the capital of Kansas.", lora_id=lora, lr=0.001, execution=str(uuid.uuid4()), - base_step=_steps.get(lora, 0), engine=ENGINE) + base_step=_steps.get(lora, 0), engine=ENGINE, input=f"experience-{uuid.uuid4()}") step.update(extra) # a test may pin a stale base or engine return step def landed(lora, outcome): @@ -90,14 +90,30 @@ def landed(lora, outcome): assert not stale_engine["success"] and stale_engine["failure"].startswith("stale_engine"), stale_engine unnamed = {k: v for k, v in fenced(sid).items() if k != "execution"} assert "execution" in json.loads(raw("/completion", turn(unnamed)))["learn"]["failure"] + # the same execution re-sent with any bound field changed is refused, never an unrelated success + for change in ({"input": "another-experience"}, {"base_step": first_step["base_step"] + 7}, {"lora_id": sid + 1000}): + rebound = json.loads(raw("/completion", turn(dict(first_step, **change))))["learn"] + assert not rebound["success"] and rebound["failure"].startswith("execution_rebound"), (change, rebound) ack = json.loads(raw("/learn/receipts/ack", {"executions": [first_step["execution"]]})) assert ack["released"] == 1 and ack["outstanding"] == 0, ack # its receipt is gone: the delayed duplicate is refused by the shadow's step, not run again late = json.loads(raw("/completion", turn(first_step)))["learn"] assert not late["success"] and late["failure"].startswith("stale_base_step"), late nxt = landed(sid, json.loads(raw("/completion", turn(fenced(sid))))["learn"]) - assert nxt["success"] and nxt["step"] == 2, nxt - print(json.dumps({"first": first, "replay": replay, "stale_base": stale_base["failure"], "late_duplicate": late["failure"], "next": nxt["step"]})) + assert nxt["success"] and nxt["step"] == 2 and nxt["generation"] == sid, nxt + # unload and restore: a new generation (a new load id); a delayed request of the old generation, + # its receipt acknowledged, is refused and steps nothing + old_gen_step = fenced(sid) + raw("/learn/receipts/ack", {"executions": [nxt["execution"]]}) + raw("/lora-adapters/unload", {"id": sid}) + restored = json.loads(raw("/lora-adapters/load", {"name": "shadow-05.gguf", "step": 2}))["id"] + assert restored != sid, (restored, sid) + stale_gen = json.loads(raw("/completion", turn(old_gen_step)))["learn"] + assert not stale_gen["success"] and "no loaded adapter" in stale_gen["failure"], stale_gen + resumed = landed(restored, json.loads(raw("/completion", turn(dict(fenced(restored), base_step=2))))["learn"]) + assert resumed["success"] and resumed["step"] == 3 and resumed["generation"] == restored, resumed + print(json.dumps({"first": first, "replay": replay, "stale_base": stale_base["failure"], "late_duplicate": late["failure"], + "next": nxt["step"], "stale_generation": stale_gen["failure"], "resumed": resumed["step"]})) sys.exit(0) # Two minds on one serving context: each shadow keeps its own optimizer while the other learns in @@ -137,7 +153,7 @@ def landed(lora, outcome): raw("/v1/chat/completions", request) step = fenced(sid) experience = dict(request, messages=messages + [reply], lora_id=sid, lr=step["lr"], - execution=step["execution"], base_step=step["base_step"], engine=ENGINE) + execution=step["execution"], base_step=step["base_step"], engine=ENGINE, input=step["input"]) expect_refusal = bool(reply.get("reasoning_content")) and "--expect-reasoning-refusal" in sys.argv try: outcome = json.loads(raw("/learn", experience)) @@ -159,7 +175,8 @@ def landed(lora, outcome): print(json.dumps({"template_reply": "tool_calls" if "tool_calls" in reply else "reasoning" if reply.get("reasoning_content") else "text", "outcome": outcome})) try: raw("/learn", {"messages": messages + [{"role": "assistant", "content": ""}], "id_slot": 0, "lora_id": sid, - "execution": str(uuid.uuid4()), "base_step": _steps.get(sid, 0), "engine": ENGINE}) + "execution": str(uuid.uuid4()), "base_step": _steps.get(sid, 0), "engine": ENGINE, + "input": f"experience-{uuid.uuid4()}"}) raise AssertionError("empty assistant reply was accepted") except urllib.error.HTTPError as error: assert error.code == 400, error