From e00953d115c768ff703ef67519fd5b7cf99b1b3b Mon Sep 17 00:00:00 2001 From: joelteply Date: Sun, 1 Nov 2026 17:31:46 -0600 Subject: [PATCH 1/3] 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 33f855969b9e..3df2bf251ec0 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -31,6 +31,8 @@ #include #include #include +#include +#include #include #include @@ -948,6 +950,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 @@ -2168,6 +2190,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; @@ -2178,6 +2237,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; @@ -3318,6 +3383,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 + } publish_lora_report(); 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(); @@ -3345,12 +3413,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 @@ -5026,6 +5126,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, @@ -5526,6 +5627,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 }, }; @@ -6280,6 +6382,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); @@ -6315,6 +6418,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 @@ -6401,6 +6507,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 92813c045eba..21098132aab7 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; @@ -172,6 +173,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 ce846718c7f2..4194372806fb 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp @@ -1712,8 +1712,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 44cb200da9fc..5e07c68fb725 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; @@ -687,6 +702,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 d1f67e01a149..8f7a39fcffe1 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -221,6 +221,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.get_cells = models_routes->proxy_get; @@ -279,6 +281,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 74ba3378fe2e..3474dc7b8d01 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 @@ -114,7 +154,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)) @@ -132,10 +174,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 @@ -145,7 +188,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 = [], [] @@ -177,25 +220,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() @@ -216,7 +259,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 = [] @@ -230,7 +273,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 d17ff9fa5765681ffc8cb53d3806170a0f59ccdc Mon Sep 17 00:00:00 2001 From: joelteply Date: Sun, 1 Nov 2026 17:50:58 -0600 Subject: [PATCH 2/3] 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 3df2bf251ec0..f4977a494085 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -2190,27 +2190,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; @@ -2239,6 +2251,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); @@ -3426,7 +3441,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; @@ -6421,6 +6438,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 4194372806fb..85ba1f76fb9a 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp @@ -1718,6 +1718,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 5e07c68fb725..365ba0433ba4 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 3474dc7b8d01..02d76d44fd97 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 @@ -156,7 +172,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)) @@ -178,7 +194,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 From 7bf7bf78c9a9ca13e3138de4cc04fdca794524f1 Mon Sep 17 00:00:00 2001 From: joelteply Date: Sat, 7 Nov 2026 13:12:54 -0600 Subject: [PATCH 3/3] learn: the fence and the receipt follow the optimizer's committed steps, and a failed period is discarded (Fable on #88) (1) llama_opt_learn could return false after its optimizer stepped (the step runs inside the last labelled ubatch; a failed restore after it returned false): the server then counted no step and stored no receipt, so a redispatch with the same base_step passed the fence and trained the same input twice. llama_opt_steps (ggml_opt_steps underneath) reports the committed steps; the server reads it before and after and counts the step, writes the receipt and reports success whenever the optimizer stepped, logging a step that committed before a later failure. Her slot is re-prefilled afterwards either way, as before. (2) A reply spanning batches that failed part-way through a period left its partial gradient in the accumulators, and the next learn's step applied it under another input's receipt. opt_learn now calls ggml_opt_discard_period on a failed training batch: the next backward starts a fresh period from zero gradients; the moments and iteration stand. Test: test-opt-dynamic-accum run_discarded_period (period 2, sum loss): one graph, a discard, one graph steps nothing; the next steps by exactly lr * 2x, from an untouched optimizer. Mutation-proved: with the discard a no-op the test fails (w 0.5 after one step, the carried gradient); restored, it passes. llama-server builds with the server change (CPU, BLAS). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FRQWzgSfo79JtnwZywHKrE --- ggml/include/ggml-opt.h | 8 +++ ggml/src/ggml-opt.cpp | 10 ++++ include/llama.h | 5 ++ src/llama-context-opt.cpp | 14 +++++ src/llama-context.h | 2 + tests/test-opt-dynamic-accum.cpp | 70 ++++++++++++++++++++++++ tools/server/server-context.cpp | 18 ++++-- tools/server/tests/learn-every-format.py | 6 +- 8 files changed, 126 insertions(+), 7 deletions(-) diff --git a/ggml/include/ggml-opt.h b/ggml/include/ggml-opt.h index 824b35d76b0b..305fafdfeadd 100644 --- a/ggml/include/ggml-opt.h +++ b/ggml/include/ggml-opt.h @@ -153,6 +153,14 @@ extern "C" { // set gradients to zero, initialize loss, and optionally reset the optimizer GGML_API void ggml_opt_reset(ggml_opt_context_t opt_ctx, bool optimizer); + // The optimizer steps taken (the iteration the next step uses, minus one): read before and after + // a call that may step, so a caller knows whether the weights moved even when the call failed. + GGML_API int64_t ggml_opt_steps(ggml_opt_context_t opt_ctx); + // Abandons the period in progress: the gradients accumulated since the last step are dropped (the + // next backward starts a fresh period from zero), so a failure part-way through a period never + // carries its partial gradient into the next step. The optimizer's moments and iteration stay. + GGML_API void ggml_opt_discard_period(ggml_opt_context_t opt_ctx); + GGML_API bool ggml_opt_static_graphs(ggml_opt_context_t opt_ctx); // whether the graphs are allocated_statically // get underlying tensors that store data diff --git a/ggml/src/ggml-opt.cpp b/ggml/src/ggml-opt.cpp index e00807b0328a..0558fcc5c139 100644 --- a/ggml/src/ggml-opt.cpp +++ b/ggml/src/ggml-opt.cpp @@ -958,6 +958,16 @@ struct ggml_tensor * ggml_opt_leaf_grad(ggml_opt_context_t opt_ctx, struct ggml_ return it == opt_ctx->leaf_grads.end() ? nullptr : it->second; } +int64_t ggml_opt_steps(ggml_opt_context_t opt_ctx) { + return opt_ctx->iter - 1; +} + +void ggml_opt_discard_period(ggml_opt_context_t opt_ctx) { + // the next ggml_opt_alloc zeroes the accumulators when a period starts fresh at opt_i 0 + opt_ctx->period_fresh = true; + opt_ctx->opt_i = 0; +} + void ggml_opt_reset(ggml_opt_context_t opt_ctx, bool optimizer) { opt_ctx->checkpoint_ready = false; if (optimizer) { diff --git a/include/llama.h b/include/llama.h index 752bffc03abf..ac3f93758abd 100644 --- a/include/llama.h +++ b/include/llama.h @@ -1860,6 +1860,11 @@ extern "C" { // A reply within one batch is one exact step; a longer one steps per batch, its gradient // stopped at each batch boundary. false: refused or failed by name (llama_opt_failure), // with the sequence restored. + // The optimizer steps the active shadow's optimizer has committed (0 before llama_opt_init). Read + // before and after llama_opt_learn: a learn whose optimizer stepped committed an update even when + // it returned false (its restore failed), so the caller's fence and receipt follow this count. + LLAMA_API int64_t llama_opt_steps(struct llama_context * lctx); + LLAMA_API bool llama_opt_learn( struct llama_context * lctx, llama_seq_id seq, diff --git a/src/llama-context-opt.cpp b/src/llama-context-opt.cpp index eb4b28b771c5..ce67bab852e1 100644 --- a/src/llama-context-opt.cpp +++ b/src/llama-context-opt.cpp @@ -1015,6 +1015,12 @@ bool llama_context::opt_learn(llama_seq_id seq, llama_pos n_cached, const llama_ opt_outputs_internal = false; opt_learn_labels = 0; opt_learn_labels_done = 0; + // A batch that failed part-way through a period leaves its partial gradient in the accumulators: + // drop the period, so the next learn's step never applies this input's gradient under another + // input's receipt (Fable on fork #88). Steps already taken by earlier batches stand. + if (!ok && opt_ctx != nullptr) { + ggml_opt_discard_period(opt_ctx); + } // RESTORE: the reply's cells and the recurrent state decoded again under her stable set bool restored = attn == nullptr || attn->seq_rm(seq, n_cached, -1); if (recr != nullptr) { @@ -1034,6 +1040,14 @@ bool llama_context::opt_learn(llama_seq_id seq, llama_pos n_cached, const llama_ return ok; } +int64_t llama_context::opt_steps() const { + return opt_ctx != nullptr ? ggml_opt_steps(opt_ctx) : 0; +} + +int64_t llama_opt_steps(struct llama_context * ctx) { + return ctx->opt_steps(); +} + bool llama_opt_learn( struct llama_context * ctx, llama_seq_id seq, diff --git a/src/llama-context.h b/src/llama-context.h index 8f5a610ddf66..d096ca397093 100644 --- a/src/llama-context.h +++ b/src/llama-context.h @@ -209,6 +209,8 @@ struct llama_context { bool opt_select(const llama_adapter_lora * adapter, struct llama_opt_params lopt_params); // forget `adapter`'s optimizer (its moments go); false when it has none bool opt_release(const llama_adapter_lora * adapter); + // the active shadow's committed optimizer steps (0 before llama_opt_init) + int64_t opt_steps() const; // TODO: more flexible combinations of logical/physical batch size and context size // set by llama_opt_stop from any thread; opt_epoch returns at the next data item diff --git a/tests/test-opt-dynamic-accum.cpp b/tests/test-opt-dynamic-accum.cpp index fc8d1f0c1190..534a3867683f 100644 --- a/tests/test-opt-dynamic-accum.cpp +++ b/tests/test-opt-dynamic-accum.cpp @@ -513,11 +513,81 @@ static void run_refuses_misshaped_gradient() { ggml_free(ctx); } +// what this catches (Fable on fork #88): a learn whose reply spans batches failing part-way through +// a period left that batch's partial gradient in the accumulators, and the next learn's step applied +// it under another input's receipt. ggml_opt_discard_period drops the period. With a period of two +// graphs on loss = sum(w * x) (gradient x per graph): one graph, a discard, then one graph is the +// first half of a NEW period, so nothing steps; the next graph ends it and moves w by exactly +// -lr * 2x, as from an untouched optimizer. Without the discard the second graph already ends the +// period. ggml_opt_steps counts the committed steps the caller fences on. +static void run_discarded_period() { + ggml_backend_t cpu = ggml_backend_init_by_type(GGML_BACKEND_DEVICE_TYPE_CPU, nullptr); + GGML_ASSERT(cpu != nullptr); + ggml_backend_t backends[] = { cpu }; + ggml_backend_sched_t sched = ggml_backend_sched_new(backends, nullptr, 1, GGML_DEFAULT_GRAPH_SIZE, false, true); + + ggml_init_params sp = { 8 * ggml_tensor_overhead(), nullptr, true }; + ggml_context * ctx_static = ggml_init(sp); + ggml_tensor * w = ggml_new_tensor_1d(ctx_static, GGML_TYPE_F32, 1); + ggml_set_param(w); + ggml_backend_buffer_t buf = ggml_backend_alloc_ctx_tensors(ctx_static, cpu); + const float w0 = 1.0f; + ggml_backend_tensor_set(w, &w0, 0, sizeof(float)); + + ggml_opt_params params = ggml_opt_default_params(sched, GGML_OPT_LOSS_TYPE_SUM); + params.optimizer = GGML_OPT_OPTIMIZER_TYPE_SGD; + params.get_opt_pars = sgd_pars; + params.opt_period = 2; + ggml_opt_context_t opt_ctx = ggml_opt_init(params); + ggml_opt_result_t result = ggml_opt_result_init(); + CHECK(ggml_opt_steps(opt_ctx) == 0); + + auto eval = [&]() { + ggml_init_params cp = { GGML_DEFAULT_GRAPH_SIZE * ggml_tensor_overhead() + 4 * ggml_graph_overhead_custom(GGML_DEFAULT_GRAPH_SIZE, true), nullptr, true }; + ggml_context * ctx_compute = ggml_init(cp); + ggml_tensor * x = ggml_new_tensor_1d(ctx_compute, GGML_TYPE_F32, 1); + ggml_tensor * out = ggml_mul(ctx_compute, w, x); + ggml_cgraph * gf = ggml_new_graph_custom(ctx_compute, GGML_DEFAULT_GRAPH_SIZE, true); + ggml_build_forward_expand(gf, out); + ggml_opt_prepare_alloc(opt_ctx, ctx_compute, gf, x, out); + CHECK(ggml_opt_alloc(opt_ctx, true)); + ggml_backend_tensor_set(x, &X, 0, sizeof(float)); + ggml_opt_eval(opt_ctx, result); + ggml_free(ctx_compute); + }; + auto w_now = [&]() { + float v; + ggml_backend_tensor_get(w, &v, 0, sizeof(float)); + return v; + }; + + eval(); // the failed learn's first graph: half a period, a gradient + ggml_opt_discard_period(opt_ctx); // its next graph failed: the period is dropped + eval(); // the next learn's first graph: half of a fresh period + if (std::fabs(w_now() - w0) > 1e-7f || ggml_opt_steps(opt_ctx) != 0) { + fprintf(stderr, "a discarded period still counted: w %f (from %f), %lld steps\n", w_now(), w0, (long long) ggml_opt_steps(opt_ctx)); + exit(1); + } + eval(); // ends the fresh period: its two graphs, never three + if (std::fabs(w_now() - (w0 - LR * 2 * X)) > 1e-6f || ggml_opt_steps(opt_ctx) != 1) { + fprintf(stderr, "after a discarded period w moved to %f, expected %f\n", w_now(), w0 - LR * 2 * X); + exit(1); + } + + ggml_opt_result_free(result); + ggml_opt_free(opt_ctx); + ggml_backend_buffer_free(buf); + ggml_free(ctx_static); + ggml_backend_sched_free(sched); + ggml_backend_free(cpu); +} + int main() { ggml_backend_load_all(); run(1); run(2); run_manual(); + run_discarded_period(); run_refuses_graph_without_backward(); run_trains_through_broadcast_divisor(); run_refuses_misshaped_gradient(); diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index f4977a494085..15aaf35e55a0 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -2348,9 +2348,15 @@ struct server_context_impl { cp->load_tgt(ctx_tgt, slot.id, LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY); } ggml_opt_result_t result = ggml_opt_result_init(); + const int64_t steps_before = llama_opt_steps(ctx_tgt); const bool ok = llama_opt_learn(ctx_tgt, slot.id, n0, toks.data(), labels.data(), (int32_t) toks.size(), result); + // COMMITTED is whether the optimizer stepped, not whether the call returned true: a learn whose + // restore failed after the step has moved her shadow, so it is counted and receipted, and a + // redispatch is answered from that receipt instead of training the same input twice (Fable + // on fork #88). Her slot is re-prefilled below either way. + const bool committed = llama_opt_steps(ctx_tgt) > steps_before; double loss = 0.0, unc = 0.0; - if (ok) { + if (committed) { ggml_opt_result_loss(result, &loss, &unc); } ggml_opt_result_free(result); @@ -2393,9 +2399,13 @@ struct server_context_impl { slot.prompt.tokens.keep_first(n_keep); slot.prompt.checkpoints.remove_if([&](const common_prompt_checkpoint & c) { return c.n_tokens > n_keep; }); learn_outcome out; - out.ok = ok; - out.step = ok ? ++learn_steps[loras[L.lora_id].ptr] : 0; - out.failure = ok ? "" : llama_opt_failure(ctx_tgt); + out.ok = committed; + out.step = committed ? ++learn_steps[loras[L.lora_id].ptr] : 0; + out.failure = committed ? "" : llama_opt_failure(ctx_tgt); + if (committed && !ok) { + SRV_WRN("learn on slot %d committed its step, then failed after it (%s): the slot was re-prefilled\n", slot.id, + llama_opt_failure(ctx_tgt)); + } out.loss = loss; out.n_labels = (int32_t) toks.size() - 1; out.t_ms = (ggml_time_us() - t0) / 1000.0; diff --git a/tools/server/tests/learn-every-format.py b/tools/server/tests/learn-every-format.py index 02d76d44fd97..a1d9711ef32c 100755 --- a/tools/server/tests/learn-every-format.py +++ b/tools/server/tests/learn-every-format.py @@ -142,15 +142,15 @@ def landed(lora, outcome): if "--non-f32-refused" in sys.argv: f16 = json.loads(raw("/lora-adapters/load", {"name": "shadow-f16.gguf"}))["id"] turn = lambda lora: {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, - "learn": dict(learn, lora_id=lora)} + "learn": fenced(lora)} first = json.loads(raw("/completion", turn(sid))) - assert first["learn"]["success"], first + assert landed(sid, first["learn"])["success"], first refused = json.loads(raw("/completion", turn(f16))) assert not refused["learn"]["success"] and "non-F32" in refused["learn"]["failure"], refused with urllib.request.urlopen(U + "/health", timeout=30) as r: assert json.load(r)["status"] == "ok", "the server still answers after the refusal" again = json.loads(raw("/completion", turn(sid))) - assert again["learn"]["success"], again + assert landed(sid, again["learn"])["success"], again print(json.dumps({"refused": refused["learn"]["failure"], "after": again["learn"]["loss"]})) sys.exit(0)