From 97f2d878324806f7a6857b2ed700f442f0156936 Mon Sep 17 00:00:00 2001 From: joelteply Date: Tue, 10 Nov 2026 15:21:08 -0600 Subject: [PATCH 1/2] tests: one learn fixture owns custody, grants and release for every learn script Every learn now runs inside custody the core registers (weights, optimizer and workspace residencies from the engine's own preflight) under a ledger grant, and every lifecycle verb names the engine incarnation. Seven scripts predated that; each copied its own transport and none could run. tools/server/tests/learn_fixture.py is the one owner: - Engine(url, name): the incarnation, an idle-only grant, request/post/ok/get with the engine added on bound routes; - fresh()/load(): dry-run plan, registered weights residency, seat; - learn()/experience(): preflight THIS example, optimizer residency once per adapter, a workspace per example, split exactly as admit_learning_residency splits rows; - shared_grant(measured, ...): a contended share whose latency allowance covers the measured chunks and boundary, max(entry, restoration); - filling(): the longest reply the slot window holds, counted by the server's tokenizer; - close(), also at exit: retire adapters, release every reference, revoke grants. Without it the engine-wide workspace refused the next script's owner (owner_mismatch, reproduced here). Migrated: dream-adapter-routes.py (CI), learn-past-example.py, learn-unload-during-learn.py, learn-every-format.py (all six modes), learn-cancel-custody.py, learn-chunk-yields.py, learn-refuses-repacked.py. No script needs a hand-placed shadow-05.gguf any more: shadows are minted fresh (two from one seed are the "copy" two-minds needs); the F16 copy is still made by make-f16-shadow.py, from a fresh shadow. Measured on a CPU llama-server built from this branch (IntelMac, stories260K, --no-repack): dream-adapter-routes twice then learn-past-example twice on ONE server: pass (this failed with owner_mismatch before close existed); learn-every-format default, --shadow-retirement, --two-minds, --non-f32-refused: pass; learn-cancel-custody: pass; learn-refuses-repacked (server without --no-repack): pass. Not passing, reported, not hidden: - learn-unload-during-learn.py aborts the engine 2 of 3 runs: ggml-opt.cpp:969 GGML_ASSERT(!allocation_pending && !graph_pending) in ggml_opt_free <- opt_release, freeing idle B while A learns (confirmed from source by Codex; the native fix is assigned). This script is the regression for it. - --fairness and learn-chunk-yields learn under contention: a contended graph chunk (1.5 s on chunk-yields) exceeds an allowance only idle steps can measure, so the step is refused mid-way (chunk_exceeds_latency). How a contended learn obtains its allowance is the ledger owner's open question. - --template-replies needs a model whose template renders tool_calls; stories260K's does not. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FRQWzgSfo79JtnwZywHKrE --- tools/server/tests/dream-adapter-routes.py | 92 ++------- tools/server/tests/learn-cancel-custody.py | 32 ++- tools/server/tests/learn-chunk-yields.py | 38 ++-- tools/server/tests/learn-every-format.py | 166 ++++++++++------ tools/server/tests/learn-past-example.py | 64 +----- tools/server/tests/learn-refuses-repacked.py | 33 +-- .../server/tests/learn-unload-during-learn.py | 69 ++++--- tools/server/tests/learn_fixture.py | 188 ++++++++++++++++++ 8 files changed, 406 insertions(+), 276 deletions(-) create mode 100644 tools/server/tests/learn_fixture.py diff --git a/tools/server/tests/dream-adapter-routes.py b/tools/server/tests/dream-adapter-routes.py index 4b8d7a88126b..69919d81b450 100644 --- a/tools/server/tests/dream-adapter-routes.py +++ b/tools/server/tests/dream-adapter-routes.py @@ -7,32 +7,18 @@ import concurrent.futures import hashlib import json +import os from pathlib import Path import sys import time -import urllib.error -import urllib.request -import uuid - -url = sys.argv[1].rstrip("/") -train_dir = Path(sys.argv[2]) -prefix = "d1-" + uuid.uuid4().hex - - -def post(route, body): - req = urllib.request.Request(url + route, json.dumps(body).encode(), - {"Content-Type": "application/json"}) - try: - with urllib.request.urlopen(req, timeout=120) as response: - return response.status, json.load(response) - except urllib.error.HTTPError as error: - return error.code, json.load(error) +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine -def ok(route, body): - status, result = post(route, body) - assert status == 200, (route, status, result) - return result +engine = Engine(sys.argv[1], "d1", timeout=120) +train_dir = Path(sys.argv[2]) +prefix = engine.prefix +post, ok, get = engine.post, engine.ok, engine.get def complete(adapter=None): @@ -42,29 +28,6 @@ def complete(adapter=None): return ok("/completion", body)["content"] -def get(route): - with urllib.request.urlopen(url + route, timeout=120) as response: - return json.load(response) - - -ledger = get("/learn/grants") -def register(suffix, subject, rows): - # The engine seats an adapter only inside a residency the core registered for its exact - # subject, with the leases that charge its rows (server-learn-residency.h). - rid = prefix + suffix - leases = [{"kind": kind, "lease_id": rid + kind, - "bytes": sum(row["bytes"] for row in rows if row["host"] == host)} - for kind, host in (("vram", False), ("ram", True))] - assert ok("/learn/residency", {"id": rid, "engine": ledger["engine"], "owner": prefix, - "subject": subject, "rows": rows, "leases": leases})["refusal"] == "none" - return rid -grant = prefix + "-grant" -# An idle-only grant (no share, no period): the one authority the ledger lets learn before this -# geometry's chunk cost is measured (server-learn-ledger.h, pending_calibration). A shared grant -# needs a measured max_chunk_us first. Nothing else decodes while this test learns, so it is never -# contended. -assert ok("/learn/grants", {"id": grant, "engine": ledger["engine"], - "period_ms": 0, "share_ppm": 0, "expires_in_ms": 120000})["refusal"] == "none" baseline = complete() fresh_name = prefix + "-fresh.gguf" # Dry-run uses the real loader's placement/alignment before any adapter is seated or file written. @@ -73,8 +36,7 @@ def register(suffix, subject, rows): assert not plan["reserved"] and plan["weights"] assert not (train_dir / fresh_name).exists(), "dry run published an adapter" assert complete() == baseline, "dry run changed serving" -fresh_weights = register("-fresh", {"fresh": fresh_spec}, [dict(row, kind="weights") for row in plan["weights"]]) -fresh = ok("/lora-adapters/fresh", dict(fresh_spec, residency=fresh_weights)) +fresh = engine.fresh(**fresh_spec) assert fresh["weights"] == plan["weights"], ("allocated weights differ from dry plan", fresh, plan) file_plan = ok("/lora-adapters/load", {"name": fresh_name, "dry_run": True}) assert file_plan["weights"] == plan["weights"] @@ -88,42 +50,17 @@ def register(suffix, subject, rows): repeat_name = prefix + "-repeat.gguf" ok("/lora-adapters/save", {"id": fresh["id"], "name": repeat_name}) assert hashlib.sha256((train_dir / repeat_name).read_bytes()).hexdigest() == before, "save bytes are unstable without learning" -optimizer_rid = None -def learning(reply, suffix): - # A learn runs only inside custody the core registered from the engine's own preflight of this - # example: her optimizer (persistent per shadow, registered once) and the engine-wide step - # workspace sized to this reply (server-context-learn-residency.inc admit_learning_residency). - global optimizer_rid - plan = ok("/learn/preflight", {"engine": ledger["engine"], "lora_id": fresh["id"], "learn": {"reply": reply}}) - assert plan["refusal"] == "none", plan - rows = [row for row in plan["requirements"] if row["kind"] != "shared_compute_held"] - if optimizer_rid is None: - optimizer_rid = register("-optimizer", {"loaded": {"lora_id": fresh["id"], "path": fresh["path"]}}, - [row for row in rows if row["kind"] != "shared_compute_required"]) - workspace = register("-workspace" + suffix, {"workspace": True}, - [row for row in rows if row["kind"] == "shared_compute_required"]) - return {"reply": reply, "lora_id": fresh["id"], "grant": grant, "lr": 0.001, - "optimizer_residency": optimizer_rid, "workspace_residency": workspace} learned = ok("/completion", {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, - "learn": learning("Topeka is the capital of Kansas.", "-short")}) + "learn": engine.learn(fresh, "Topeka is the capital of Kansas.")}) assert learned["learn"]["success"], learned -# The long reply spans as many resumable chunks as this engine's window holds: sized from its own -# slot window and tokenizer (the reply's tokens plus its end of turn, after the served prompt), -# never a fixed repeat count, since preflight refuses an example past the served window. -def rows(text, special): - return len(ok("/tokenize", {"content": text, "add_special": special})["tokens"]) +# The long reply spans as many resumable chunks as this engine's window holds. long_prompt = "The capital of Kansas is" -room = get("/slots")[0]["n_ctx"] - rows(long_prompt, True) - 1 -unit = "A room contains a card and a window. " -repeats = room // rows(unit, False) -while repeats > 1 and rows(unit * repeats, False) > room: - repeats -= 1 -assert repeats > 1, ("the served window holds no multi-sentence reply", room) +long_reply = engine.filling(long_prompt, "A room contains a card and a window. ") # A queued save must refuse the active shadow even between resumable chunks. with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(ok, "/completion", {"prompt": long_prompt, "id_slot": 0, - "n_predict": 1, "temperature": 0, "learn": learning(unit * repeats, "-long")}) + "n_predict": 1, "temperature": 0, "learn": engine.learn(fresh, long_reply)}) deadline = time.monotonic() + 120 while time.monotonic() < deadline: if get("/slots")[0]["is_learning"]: @@ -141,9 +78,7 @@ def rows(text, special): assert snapshot["steps"] == 2, ("two whole replies learned are two committed steps", snapshot) after = hashlib.sha256((train_dir / snapshot_name).read_bytes()).hexdigest() assert after != before, "successful learn did not change saved adapter weights" -snapshot_plan = ok("/lora-adapters/load", {"name": snapshot_name, "dry_run": True}) -snapshot_weights = register("-snapshot", {"load": {"name": snapshot_name}}, [dict(row, kind="weights") for row in snapshot_plan["weights"]]) -loaded = ok("/lora-adapters/load", {"name": snapshot_name, "residency": snapshot_weights}) +loaded = engine.load(snapshot_name) assert complete(loaded["id"]) == trained_text, "trained snapshot reload changed output" assert post("/lora-adapters/save", {"id": fresh["id"], "name": snapshot_name})[0] >= 400 assert post("/lora-adapters/fresh", {"name": fresh_name})[0] >= 400 @@ -157,3 +92,4 @@ def rows(text, special): assert sorted(status == 200 for status in results) == [False, True], results print(json.dumps({"gate": "D1", "fresh": fresh, "snapshot": snapshot, "before_sha256": before, "after_sha256": after, "passed": True})) +engine.close() diff --git a/tools/server/tests/learn-cancel-custody.py b/tools/server/tests/learn-cancel-custody.py index f2dfa510525e..7d6e5c618486 100644 --- a/tools/server/tests/learn-cancel-custody.py +++ b/tools/server/tests/learn-cancel-custody.py @@ -2,33 +2,26 @@ """Disconnect during learning, then reuse the slot through the real server queue. What this catches: HTTP cancellation releasing a sequence while its training or -stable restoration still owns KV. Run on an isolated two-slot server with ---slots, batch 128, and a fresh shadow-05.gguf under --train-dir. +stable restoration still owns KV. Usage: learn-cancel-custody.py URL, on an isolated +two-slot server with --slots, batch 128, --no-repack and --train-dir. """ import json import os import socket +import sys import time import urllib.parse -import urllib.request -url = os.environ["LLAMA_SERVER_URL"] +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine - -def request(path, body=None): - data = None if body is None else json.dumps(body).encode() - req = urllib.request.Request(url + path, data=data, - headers={"Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=120) as response: - return json.load(response) - - -sid = request("/lora-adapters/load", {"name": "shadow-05.gguf"})["id"] -target = urllib.parse.urlsplit(url) -body = json.dumps({"prompt": "Remember these observed facts:", "id_slot": 0, - "n_predict": 8, "temperature": 0, - "learn": {"lora_id": sid, "lr": 0.001, - "reply": "A room contains a card and a window. " * 55}}).encode() +engine = Engine(sys.argv[1], "cancel", timeout=120) +request = lambda path, body=None: engine.get(path) if body is None else engine.ok(path, body) +shadow = engine.fresh() +prompt = "Remember these observed facts:" +target = urllib.parse.urlsplit(engine.url) +body = json.dumps({"prompt": prompt, "id_slot": 0, "n_predict": 8, "temperature": 0, + "learn": engine.learn(shadow, engine.filling(prompt, "A room contains a card and a window. "))}).encode() conn = socket.create_connection((target.hostname, target.port), timeout=120) try: conn.sendall(("POST /completion HTTP/1.1\r\nHost: " + target.netloc @@ -56,3 +49,4 @@ def request(path, body=None): assert not after["is_learning"] and not after["is_processing"], after print(json.dumps({"cancelled_task": held["id_task"], "next_task": after["id_task"], "next_tokens": following["tokens_predicted"], "slot_restored": True})) +engine.close() diff --git a/tools/server/tests/learn-chunk-yields.py b/tools/server/tests/learn-chunk-yields.py index ab60067dd336..9e4cd890aeac 100644 --- a/tools/server/tests/learn-chunk-yields.py +++ b/tools/server/tests/learn-chunk-yields.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Real two-slot server test; requires an isolated server and shadow-05.gguf. +"""Real two-slot server test; requires an isolated server. Usage: learn-chunk-yields.py URL What this catches: a reply loop that blocks the event loop until learning ends, or releases its slot before stable restoration. A neighbour must emit tokens @@ -8,22 +8,28 @@ """ import json import os +import sys import threading import time import urllib.request -URL = os.environ["LLAMA_SERVER_URL"] +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine - -def request(path, body=None): - data = None if body is None else json.dumps(body).encode() - req = urllib.request.Request(URL + path, data=data, - headers={"Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=120) as response: - return json.load(response) - - -shadow = request("/lora-adapters/load", {"name": "shadow-05.gguf"})["id"] +engine = Engine(sys.argv[1], "yields", timeout=120) +URL = engine.url +request = lambda path, body=None: engine.get(path) if body is None else engine.ok(path, body) +shadow = engine.fresh() +prompt = "Remember these observed facts:" +reply = engine.filling(prompt, "A room contains a card and a window. ") +# The learn runs while a neighbour decodes, so it needs a share of a contended engine: calibrated +# by two idle steps on this geometry (cold and warm entries), then a whole-engine grant. +idle = {"prompt": prompt, "id_slot": 0, "n_predict": 8, "temperature": 0} +cold = request("/completion", dict(idle, learn=engine.learn(shadow, reply)))["learn"] +warm = request("/completion", dict(idle, learn=engine.learn(shadow, reply)))["learn"] +whole = engine.shared_grant(max(cold, warm, key=lambda o: o["max_chunk_ms"]), + period_ms=engine.max_expiry_ms, share_ppm=1000000) +learn_body = dict(idle, learn=engine.learn(shadow, reply, grant=whole)) frames, errors, result = [], [], [] started, stop = threading.Event(), threading.Event() @@ -48,12 +54,7 @@ def neighbour(): def learn(): try: - result.append(request("/completion", { - "prompt": "Remember these observed facts:", "id_slot": 0, - "n_predict": 8, "temperature": 0, - "learn": {"lora_id": shadow, "lr": 0.001, - "reply": "A room contains a card and a window. " * 55}, - })) + result.append(request("/completion", learn_body)) except Exception as error: errors.append(str(error)) @@ -86,3 +87,4 @@ def learn(): "neighbour_frames_during_learning": max(observed) - min(observed)})) finally: stop.set() +engine.close() diff --git a/tools/server/tests/learn-every-format.py b/tools/server/tests/learn-every-format.py index 92d0940c09d3..671c83471399 100755 --- a/tools/server/tests/learn-every-format.py +++ b/tools/server/tests/learn-every-format.py @@ -1,84 +1,99 @@ #!/usr/bin/env python3 # Every final-response format x stream mode carries exactly one learn outcome for a learn-carrying -# request. Run against a server started with --train-dir holding shadow-05.gguf (a fresh LoRA for the -# served model), e.g.: llama-server -m MODEL --port 58199 -np 1 --jinja --train-dir DIR +# request. Run against an isolated server with --train-dir and --no-repack on CPU, e.g.: +# llama-server -m MODEL --port 58199 -np 1 --jinja --no-repack --train-dir DIR +# learn-every-format.py URL TRAIN_DIR [--shadow-retirement | --two-minds | --non-f32-refused | +# --template-replies | --fairness] +# Every shadow is minted fresh through learn_fixture (its custody, grants and release), never a +# hand-placed shadow-05.gguf. # 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 -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() +import json, os, subprocess, sys, threading, time +from pathlib import Path + +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine + +engine = Engine(sys.argv[1], "formats") +train_dir = Path(sys.argv[2]) +U = engine.url +raw = lambda path, body: engine.request(path, body)[1] +reply = "Topeka is the capital of Kansas." +shadow = engine.fresh() +learn = engine.learn(shadow, reply) 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" -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} + assert engine.get("/props")["total_slots"] == 1, "shadow-retirement requires an isolated single-slot server" # Isolated adapter lifetime regression, with no live persona or other clients on this server. if "--shadow-retirement" in sys.argv: - import urllib.error - 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)) + status, retired = engine.unload(shadow) + assert status == 200 and retired["success"] and retired["retired"] and retired["id"] == shadow["id"], retired + shadow = engine.fresh() + turn = lambda learn: {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} + first = engine.ok("/completion", turn(engine.learn(shadow, reply))) assert 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)) + status, bound = engine.unload(shadow) + assert status == 200 and bound["success"] and bound["retired"] and bound["id"] == shadow["id"], bound + shadow = engine.fresh() + following = engine.ok("/completion", turn(engine.learn(shadow, reply))) assert following["learn"]["success"], following print(json.dumps({"unbound_retired": retired, "bound_retired": bound, "next_learn": following["learn"]})) + engine.close() 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). +# between. Both are minted from the same seed, so they start as the same weights. if "--two-minds" in sys.argv: - other = json.loads(raw("/lora-adapters/load", {"name": "shadow-06.gguf"}))["id"] + other = engine.fresh() steps = [] - for lora in [sid, other, sid, other]: + for adapter in [shadow, other, shadow, other]: turn = {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, - "learn": dict(learn, lora_id=lora)} - out = json.loads(raw("/completion", turn)) + "learn": engine.learn(adapter, reply)} + out = engine.ok("/completion", turn) assert 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 + steps.append((adapter["id"], out["learn"]["loss"])) + # the two shadows start from the same weights, 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 # continues its own optimizer (a shared optimizer or shared weights would split them) assert abs(steps[0][1] - steps[1][1]) < 1e-6 and abs(steps[2][1] - steps[3][1]) < 1e-6, steps assert steps[2][1] < steps[0][1], steps print(json.dumps({"steps": steps})) + engine.close() sys.exit(0) -# A learn naming an adapter that cannot train is refused by name, and serving survives it (needs -# shadow-05.gguf and an F16 copy of it, shadow-f16.gguf, in --train-dir: make-f16-shadow.py writes -# one). What this catches (Fable on fork #85, 2026-11-07): the F32 shadow learned, then a learn on -# its F16 copy hit a GGML_ASSERT in opt_select and killed the serving process (exit 134). +# A learn naming an adapter that cannot train is refused by name, and serving survives it: an F16 +# copy of a fresh shadow (make-f16-shadow.py). What this catches (Fable on fork #85, 2026-11-07): +# the F32 shadow learned, then a learn on its F16 copy hit a GGML_ASSERT in opt_select and killed +# the serving process (exit 134). 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)} - first = json.loads(raw("/completion", turn(sid))) + f16_name = engine.prefix + "-f16.gguf" + subprocess.run([sys.executable, os.path.join(os.path.dirname(__file__), "make-f16-shadow.py"), + shadow["path"], str(train_dir / f16_name)], check=True) + f16 = engine.load(f16_name) + turn = lambda learn: {"prompt": "The capital of Kansas is", "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} + first = engine.ok("/completion", turn(learn)) assert 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))) + # Refused by name at the first gate that sees the adapter: its preflight, or the learn itself. + status, plan = engine.post("/learn/preflight", {"engine": engine.engine, "lora_id": f16["id"], "learn": {"reply": reply}}) + if status == 200 and plan["refusal"] == "none": + refused = engine.ok("/completion", turn(engine.learn(f16, reply)))["learn"] + assert not refused["success"] and "non-F32" in refused["failure"], refused + why = refused["failure"] + else: + why = json.dumps(plan) + assert "F32" in why, (status, plan) + assert engine.get("/health")["status"] == "ok", "the server still answers after the refusal" + again = engine.ok("/completion", turn(engine.learn(shadow, reply))) assert again["learn"]["success"], again - print(json.dumps({"refused": refused["learn"]["failure"], "after": again["learn"]["loss"]})) + print(json.dumps({"refused": why, "after": again["learn"]["loss"]})) + engine.close() sys.exit(0) # Isolated template/label regression; never run this shadow-adapter fixture on a persona lane. # Add --expect-reasoning-refusal only for a known non-reasoning template; the exact refusal must match. if "--template-replies" in sys.argv: - import urllib.error messages = [{"role": "user", "content": "Inspect the requested source file."}] tools = [{"type": "function", "function": {"name": "read_file", "description": "Read a file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}}] @@ -91,38 +106,44 @@ 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) expect_refusal = bool(reply.get("reasoning_content")) and "--expect-reasoning-refusal" in sys.argv - try: - outcome = json.loads(raw("/learn", experience)) - except urllib.error.HTTPError as error: - detail = error.read().decode() - assert expect_refusal and error.code == 400, (error.code, detail) + if expect_refusal: + # The template refuses before any custody is consulted: rendering comes first. + status, detail = engine.request("/learn", dict(request, messages=messages + [reply], lora_id=shadow["id"])) + assert status == 400, (status, detail) assert "the served template does not render the reply's reasoning_content" in detail, detail print(json.dumps({"template_reply": "reasoning", "expected_refusal": detail})) continue - assert not expect_refusal, ("template was expected to refuse reasoning", outcome) + experience = engine.experience(shadow, messages + [reply], + **{k: v for k, v in request.items() if k != "messages"}) + outcome = engine.ok("/learn", experience) deadline = time.monotonic() + 300 while outcome.get("deferred"): assert time.monotonic() < deadline, ("deferred fixture deadline", outcome) assert not outcome["success"] and outcome["n_labels"] == 0, outcome 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)) + outcome = engine.ok("/learn", experience) assert 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}) - raise AssertionError("empty assistant reply was accepted") - except urllib.error.HTTPError as error: - assert error.code == 400, error + status, _ = engine.request("/learn", {"messages": messages + [{"role": "assistant", "content": ""}], "id_slot": 0, "lora_id": shadow["id"]}) + assert status == 400, ("empty assistant reply was accepted", status) + engine.close() sys.exit(0) # Run with --fairness against an isolated two-slot server with --slots. This # exercises the shared admission in both directions, using the same model/adapter. if "--fairness" in sys.argv: + import math, urllib.request prompt = "Q: Which city is the capital of Kansas?\nA:" - turn = {"prompt": prompt, "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} + idle = {"prompt": prompt, "id_slot": 0, "n_predict": 1, "temperature": 0, "learn": learn} + # Calibrate under the contention this test applies: entry and restoration are measured per + # geometry, cold and warm, and a contended step costs several idle ones. Two idle steps measure + # the entries (the ledger's bootstrap), then one step under a whole-engine grant measures a + # contended step once the busy client runs (below). + cold = engine.ok("/completion", idle)["learn"] + warm = engine.ok("/completion", idle)["learn"] + measured = max(cold, warm, key=lambda o: o["max_chunk_ms"]) started, stop = threading.Event(), threading.Event() progress, errors = [], [] @@ -148,31 +169,41 @@ def busy_client(): try: assert started.wait(300) and not errors, errors def assert_busy(): - with urllib.request.urlopen(U + "/slots", timeout=300) as response: - slots = json.load(response) + slots = engine.get("/slots") assert any(s["id"] == 1 and s["is_processing"] for s in slots), slots def assert_deferred(): assert_busy() - result = json.loads(raw("/completion", turn))["learn"] + result = engine.ok("/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)) + whole = engine.shared_grant(measured, period_ms=engine.max_expiry_ms, share_ppm=1000000) + contended = engine.ok("/completion", dict(idle, learn=engine.learn(shadow, reply, grant=whole)))["learn"] + assert contended["success"], ("a whole-engine grant learns under contention", contended) + # A step is admitted on its measured entry and debited its measured time when it settles: + # 1.5 contended steps per period admits one and refuses the next until the period rolls. + # The period spans 100 contended steps so this test's own round trips cannot roll it early. + period_ms = math.ceil(100 * contended["t_ms"]) + tight = engine.shared_grant(contended, period_ms=period_ms, + share_ppm=math.ceil(1.5 * contended["t_ms"] / period_ms * 1000000)) + learn = engine.learn(shadow, reply, grant=tight) + turn = dict(idle, learn=learn) + first = engine.ok("/completion", turn) assert 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_deferred = json.loads(raw("/learn", post_body)) + post_deferred = engine.ok("/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 = engine.ok("/learn", post_body) post_wall_ms = (time.monotonic() - t0) * 1000 assert post["success"], post post_rest = assert_deferred() @@ -182,6 +213,8 @@ def assert_deferred(): "other_slot_frames": len(progress) - before})) finally: stop.set() + worker.join(timeout=30) + engine.close() raise SystemExit(0) msgs = [{"role": "user", "content": "Which city is the capital of Kansas?"}] @@ -211,4 +244,5 @@ def walk(o): 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") +engine.close() raise SystemExit(1 if fails else 0) diff --git a/tools/server/tests/learn-past-example.py b/tools/server/tests/learn-past-example.py index 044b6f48c1e2..c6452a4b8ea0 100644 --- a/tools/server/tests/learn-past-example.py +++ b/tools/server/tests/learn-past-example.py @@ -9,63 +9,21 @@ Usage: learn-past-example.py URL TRAIN_DIR (server started with --train-dir TRAIN_DIR, --no-repack on CPU) """ -import json, sys, urllib.error, urllib.request, uuid +import json, os, sys -url = sys.argv[1].rstrip("/") -prefix = "past-" + uuid.uuid4().hex +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine - -def post(route, body): - if route in ('/lora-adapters/unload', '/learn/residency/release', '/learn/residency/revoke'): - body = dict(body, engine=ledger['engine']) - request = urllib.request.Request(url + route, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}) - try: - with urllib.request.urlopen(request, timeout=600) as response: - return response.status, json.load(response) - except urllib.error.HTTPError as error: - return error.code, json.loads(error.read() or b"{}") - - -def ok(route, body): - status, result = post(route, body) - assert status == 200, (route, status, result) - return result - - -with urllib.request.urlopen(url + "/learn/grants", timeout=60) as response: - ledger = json.load(response) -with urllib.request.urlopen(url + "/props", timeout=60) as response: - props = json.load(response)["learn_ledger"] +engine = Engine(sys.argv[1], "past") +props = engine.get("/props")["learn_ledger"] +ledger = engine.get("/learn/grants") assert props["max_period_ms"] == ledger["max_period_ms"] assert props["max_expiry_ms"] == ledger["max_expiry_ms"] -grant = prefix + "-grant" -assert ok("/learn/grants", {"id": grant, "engine": ledger["engine"], "period_ms": 0, - "share_ppm": 0, "expires_in_ms": ledger["max_expiry_ms"]})["refusal"] == "none" -def register(suffix, subject, rows): - rid = prefix + suffix - leases = [{"kind": kind, "lease_id": rid + kind, - "bytes": sum(row["bytes"] for row in rows if row["host"] == host)} - for kind, host in (("vram", False), ("ram", True))] - assert ok("/learn/residency", {"id": rid, "engine": ledger["engine"], "owner": prefix, - "subject": subject, "rows": rows, "leases": leases})["refusal"] == "none" - return rid - -fresh = {"name": prefix + "-shadow.gguf", "rank": 8, "alpha": 16, - "targets": "attn_q,attn_v", "top_layers": 1, "seed": 7} -plan = ok("/lora-adapters/fresh", dict(fresh, dry_run=True)) -weights = register("-weights", {"fresh": fresh}, [dict(row, kind="weights") for row in plan["weights"]]) -shadow = ok("/lora-adapters/fresh", dict(fresh, residency=weights)) +shadow = engine.fresh() example = [{"role": "user", "content": "What is the capital of Kansas?"}, {"role": "assistant", "content": "Topeka."}] -plan = ok("/learn/preflight", {"engine": ledger["engine"], "lora_id": shadow["id"], "messages": example}) -assert plan["refusal"] == "none", plan -optimizer = register("-optimizer", {"loaded": {"lora_id": shadow["id"], "path": shadow["path"]}}, - [row for row in plan["requirements"] if row["kind"].startswith("optimizer_")]) -workspace = register("-workspace", {"workspace": True}, - [row for row in plan["requirements"] if row["kind"] == "shared_compute_required"]) - -learn = {"messages": example, "id_slot": 0, "lora_id": shadow["id"], "grant": grant, "lr": 0.001, - "optimizer_residency": optimizer, "workspace_residency": workspace} +learn = dict(engine.experience(shadow, example), id_slot=0) +ok = engine.ok # her slot serves something else first, so it no longer holds the example's prompt ok("/completion", {"prompt": "A card and a window are in the room.", "id_slot": 0, "n_predict": 1, "temperature": 0}) @@ -77,6 +35,4 @@ def register(suffix, subject, rows): assert hit["loss"] < missed["loss"], ("a second step on the same example lowers its loss", missed["loss"], hit["loss"]) print(json.dumps({"missed": missed, "hit": hit})) -assert ok("/lora-adapters/unload", {"id": shadow["id"]})["retired"] -for reference in (weights, optimizer, workspace): - assert ok("/learn/residency/release", {"id": reference})["refusal"] == "none" +engine.close() diff --git a/tools/server/tests/learn-refuses-repacked.py b/tools/server/tests/learn-refuses-repacked.py index 6469c98f9bb1..5e08798bb8a5 100644 --- a/tools/server/tests/learn-refuses-repacked.py +++ b/tools/server/tests/learn-refuses-repacked.py @@ -1,24 +1,33 @@ #!/usr/bin/env python3 # A learn-carrying completion on a server that may hold repacked CPU weights still answers its turn, # and its learn outcome says why it did not learn. Run against a CPU server WITHOUT --no-repack and a -# --train-dir holding shadow-05.gguf, e.g.: +# --train-dir, e.g.: # llama-server -m MODEL --port 58199 -np 2 -c 8192 -ngl 0 --jinja --train-dir DIR +# learn-refuses-repacked.py http://127.0.0.1:58199 # what this catches (continuum card 5b497852): a learn on repacked weights failed in the backward on # x86 and crashed the engine in ggml_backend_tensor_copy on a Metal host at -ngl 0; it must refuse # the way /train does, before any training graph is built, and leave the turn served. -import json, os, urllib.request -U = os.environ.get("LLAMA_SERVER_URL", "http://localhost:58199") -def post(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 json.loads(r.read().decode()) -sid = post("/lora-adapters/load", {"name": "shadow-05.gguf"})["id"] -res = post("/v1/chat/completions", { +import json, os, sys + +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine + +engine = Engine(sys.argv[1], "repacked") +shadow = engine.fresh() +reply = "Topeka is the capital of Kansas." +# Refused by name at the first gate that sees repacked weights: the example's preflight (before any +# custody is registered) or, past it, the learn itself; either way the turn is still served. +status, plan = engine.post("/learn/preflight", {"engine": engine.engine, "lora_id": shadow["id"], "learn": {"reply": reply}}) +learn = engine.learn(shadow, reply) if status == 200 and plan["refusal"] == "none" else {"reply": reply, "lora_id": shadow["id"], "grant": engine.grant, "lr": 0.001} +res = engine.ok("/v1/chat/completions", { "messages": [{"role": "user", "content": "Which city is the capital of Kansas?"}], "max_tokens": 8, "temperature": 0, "chat_template_kwargs": {"enable_thinking": False}, - "learn": {"reply": "Topeka is the capital of Kansas.", "lora_id": sid, "lr": 0.001}}) + "learn": learn}) answer = res["choices"][0]["message"]["content"] -learn = res.get("learn") or {} -ok = bool(answer) and learn.get("success") is False and "--no-repack" in learn.get("failure", "") -print(f"answer={answer!r} learn={learn}") +outcome = res.get("learn") or {} +why = outcome.get("failure", "") + json.dumps(plan) +ok = bool(answer) and outcome.get("success") is not True and "--no-repack" in why +print(f"answer={answer!r} learn={outcome} preflight={plan}") print("REFUSED WITH THE REASON, TURN SERVED" if ok else "WRONG") +engine.close() raise SystemExit(0 if ok else 1) diff --git a/tools/server/tests/learn-unload-during-learn.py b/tools/server/tests/learn-unload-during-learn.py index 6402bed60467..b9763ca5213f 100644 --- a/tools/server/tests/learn-unload-during-learn.py +++ b/tools/server/tests/learn-unload-during-learn.py @@ -1,29 +1,40 @@ -# Unload custody while another mind learns. What this catches (Codex on #85, s4-unload-other-shadow): -# with mind A's live reply in flight, unloading idle mind B ignored a refused llama_opt_release and -# freed B's tensors while its optimizer stayed in opt_minds. Now B's optimizer is released (its own -# buffers only) and B unloads; A's own shadow is refused until its restoration finishes. -# Run against: llama-server -m MODEL --port 58199 -np 2 --jinja --no-repack --train-dir DIR (shadow-05 + its copy shadow-06). -import json, os, threading, time, urllib.request, urllib.error -U=os.environ.get("LLAMA_SERVER_URL","http://localhost:58199") -def post(path, body): - req=urllib.request.Request(U+path, data=json.dumps(body).encode(), headers={"Content-Type":"application/json"}) - try: - with urllib.request.urlopen(req, timeout=600) as r: return r.status, json.load(r) - except urllib.error.HTTPError as e: return e.code, json.loads(e.read() or b"{}") -a=post("/lora-adapters/load",{"name":"shadow-05.gguf"})[1]["id"]; b=post("/lora-adapters/load",{"name":"shadow-06.gguf"})[1]["id"] -turn=lambda lora, reply, slot: {"prompt":"The capital of Kansas is","id_slot":slot,"n_predict":1,"temperature":0,"learn":{"reply":reply,"lora_id":lora,"lr":0.001}} -print("B first learn", post("/completion", turn(b," Topeka.",1))[1]["learn"]["success"]) -long_reply=" Topeka, the capital city of the state, sits on the Kansas River." * 40 -res={} -t=threading.Thread(target=lambda: res.update(a=post("/completion", turn(a,long_reply,0)))) -t.start(); time.sleep(4) -ub=post("/lora-adapters/unload",{"id":b}); ua=post("/lora-adapters/unload",{"id":a}) -t.join() -la=res["a"][1]["learn"] -assert ub[0] == 200 and ub[1].get("success"), ("an idle shadow unloads while another mind learns", ub) -assert ua[0] == 503 and "restoration" in ua[1]["error"]["message"], ("the learning shadow is refused", ua) -assert la["success"], ("the live reply completes", la) -b2=post("/lora-adapters/load",{"name":"shadow-06.gguf"})[1]["id"] -again=post("/completion", turn(b2," Topeka.",1))[1]["learn"] -assert again["success"], ("a reloaded B learns again", again) -print(json.dumps({"unload_b": ub[0], "unload_a": ua[0], "a_labels": la["n_labels"], "b_again": again["loss"]})) +#!/usr/bin/env python3 +"""Unload custody while another mind learns. + +What this catches (Codex on #85, s4-unload-other-shadow): with mind A's live reply in flight, +unloading idle mind B ignored a refused llama_opt_release and freed B's tensors while its optimizer +stayed in opt_minds. Now B's optimizer is released (its own buffers only) and B unloads; A's own +shadow is refused until its restoration finishes. + +Usage: learn-unload-during-learn.py URL (an isolated server with -np 2 --no-repack --train-dir DIR) +""" +import json, os, sys, threading, time + +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine + +with Engine(sys.argv[1], "unload") as engine: + prompt = "The capital of Kansas is" + a, b = engine.fresh(), engine.fresh() + turn = lambda learn, slot: {"prompt": prompt, "id_slot": slot, "n_predict": 1, "temperature": 0, "learn": learn} + first = engine.ok("/completion", turn(engine.learn(b, " Topeka."), 1))["learn"] + assert first["success"], ("B learns first", first) + # Custody is taken before A starts: preflight refuses while a learn retains the context. + long_learn = engine.learn(a, engine.filling(prompt, " Topeka, the capital city of the state, sits on the Kansas River.")) + res = {} + t = threading.Thread(target=lambda: res.update(a=engine.post("/completion", turn(long_learn, 0)))) + t.start() + deadline = time.monotonic() + 120 + while not engine.get("/slots")[0]["is_learning"]: + assert time.monotonic() < deadline and t.is_alive(), ("A's learn was never observed", res) + time.sleep(0.005) + ub, ua = engine.unload(b), engine.unload(a) + t.join() + la = res["a"][1]["learn"] + assert ub[0] == 200 and ub[1].get("success"), ("an idle shadow unloads while another mind learns", ub) + assert ua[0] == 503 and "restoration" in ua[1]["error"]["message"], ("the learning shadow is refused", ua) + assert la["success"], ("the live reply completes", la) + b2 = engine.fresh() + again = engine.ok("/completion", turn(engine.learn(b2, " Topeka."), 1))["learn"] + assert again["success"], ("a reloaded B learns again", again) + print(json.dumps({"unload_b": ub[0], "unload_a": ua[0], "a_labels": la["n_labels"], "b_again": again["loss"]})) diff --git a/tools/server/tests/learn_fixture.py b/tools/server/tests/learn_fixture.py new file mode 100644 index 000000000000..9b4fbf8d0178 --- /dev/null +++ b/tools/server/tests/learn_fixture.py @@ -0,0 +1,188 @@ +"""The one owner of the learn lifecycle every learn script drives against a live server. + +A learn runs only inside custody the core registered (server-learn-residency.h) under a grant +the ledger admitted (server-learn-ledger.h), and every lifecycle verb names the engine +incarnation it was decided against. Each script opens an Engine, takes its adapters, grants and +custody through it, and closes it: closing retires its adapters and releases everything it +registered, so the next script on the same server starts with no custody bound (the engine-wide +learning workspace refuses a second owner while the first still holds it). + + with Engine(url, "past") as engine: + shadow = engine.fresh() + outcome = engine.ok("/completion", {..., "learn": engine.learn(shadow, "Topeka.")}) +""" +import atexit +import json +import math +import urllib.error +import urllib.request +import uuid + +# Routes whose body must carry the engine incarnation. +BOUND = ("/lora-adapters/unload", "/learn/residency/release", "/learn/residency/revoke", + "/learn/grants/revoke") + +# A shadow small enough to train on any CPU fixture model: rank 8 on the top layer's q and v. +SHADOW = {"rank": 8, "alpha": 16, "targets": "attn_q,attn_v", "top_layers": 1, "seed": 7} + + +class Engine: + def __init__(self, url, name, timeout=600): + self.url = url.rstrip("/") + self.prefix = name + "-" + uuid.uuid4().hex + self.timeout = timeout + ledger = self.get("/learn/grants") + self.engine = ledger["engine"] + self.max_expiry_ms = ledger["max_expiry_ms"] + self.references, self.grants, self.shadows = [], [], [] + self.optimizers = {} + self.minted = 0 # adapter names are immutable files: never reuse one + # A failed assertion must not leave custody bound for the next script on this server. + atexit.register(self.close) + self.grant = self.idle_grant() + + def __enter__(self): + return self + + def __exit__(self, *_): + self.close() + + # ── transport ────────────────────────────────────────────────────────────────────────────── + def request(self, route, body=None): + if body is not None and route in BOUND: + body = dict(body, engine=self.engine) + data = None if body is None else json.dumps(body).encode() + req = urllib.request.Request(self.url + route, data=data, + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as response: + return response.status, response.read().decode() + except urllib.error.HTTPError as error: + return error.code, error.read().decode() + + def post(self, route, body): + status, text = self.request(route, body) + return status, json.loads(text or "{}") + + def ok(self, route, body): + status, result = self.post(route, body) + assert status == 200, (route, status, result) + return result + + def get(self, route): + status, text = self.request(route) + assert status == 200, (route, status, text) + return json.loads(text) + + # ── grants ───────────────────────────────────────────────────────────────────────────────── + def register_grant(self, suffix, **terms): + gid = self.prefix + suffix + body = dict({"id": gid, "engine": self.engine, "period_ms": 0, "share_ppm": 0, + "expires_in_ms": self.max_expiry_ms}, **terms) + assert self.ok("/learn/grants", body)["refusal"] == "none", body + self.grants.append(gid) + return gid + + def idle_grant(self): + # Idle-only authority (no share, no period): the one grant the ledger lets learn before a + # geometry's chunk cost is measured. It is refused while another slot decodes. + return self.register_grant("-idle") + + def shared_grant(self, measured, period_ms, share_ppm): + # A share of a contended engine needs a measured latency allowance covering every segment + # the engine admits against it for a learn already run on this geometry: its chunks, and + # its boundary, priced as max(entry, restoration) (server-context-learn.inc learn_on_slot). + assert measured["success"] and measured["max_chunk_ms"] > 0, measured + allowance_ms = max(measured["max_chunk_ms"], measured["entry_ms"], measured["restoration_ms"]) + return self.register_grant("-shared-%d" % len(self.grants), period_ms=period_ms, share_ppm=share_ppm, + max_chunk_us=math.ceil(allowance_ms * 1000)) + + # ── residency custody ────────────────────────────────────────────────────────────────────── + def register(self, suffix, subject, rows): + rid = self.prefix + suffix + leases = [{"kind": kind, "lease_id": rid + kind, + "bytes": sum(row["bytes"] for row in rows if row["host"] == host)} + for kind, host in (("vram", False), ("ram", True))] + assert self.ok("/learn/residency", {"id": rid, "engine": self.engine, "owner": self.prefix, + "subject": subject, "rows": rows, "leases": leases})["refusal"] == "none", rid + self.references.append(rid) + return rid + + def fresh(self, name=None, **spec): + # Measure with the real loader, register its exact rows, then seat it. + self.minted += 1 + spec = dict(SHADOW, name=name or self.prefix + "-%d.gguf" % self.minted, **spec) + plan = self.ok("/lora-adapters/fresh", dict(spec, dry_run=True)) + weights = self.register("-weights-" + spec["name"], {"fresh": spec}, + [dict(row, kind="weights") for row in plan["weights"]]) + return self.seated(self.ok("/lora-adapters/fresh", dict(spec, residency=weights))) + + def load(self, name): + plan = self.ok("/lora-adapters/load", {"name": name, "dry_run": True}) + weights = self.register("-weights-" + name, {"load": {"name": name}}, + [dict(row, kind="weights") for row in plan["weights"]]) + return self.seated(self.ok("/lora-adapters/load", {"name": name, "residency": weights})) + + def seated(self, adapter): + self.shadows.append(adapter["id"]) + return adapter + + def custody(self, adapter, example): + # Preflight THIS example, then register her optimizer once per adapter (it persists across + # her steps) and a workspace sized to this example (engine-wide, replaced per example). + # Rows split exactly as admit_learning_residency splits them. + plan = self.ok("/learn/preflight", dict(example, engine=self.engine, lora_id=adapter["id"])) + assert plan["refusal"] == "none", plan + rows = [row for row in plan["requirements"] if row["kind"] != "shared_compute_held"] + if adapter["id"] not in self.optimizers: + self.optimizers[adapter["id"]] = self.register( + "-optimizer-%d" % adapter["id"], + {"loaded": {"lora_id": adapter["id"], "path": adapter["path"]}}, + [row for row in rows if row["kind"] != "shared_compute_required"]) + workspace = self.register("-workspace-%d" % len(self.references), {"workspace": True}, + [row for row in rows if row["kind"] == "shared_compute_required"]) + return {"optimizer_residency": self.optimizers[adapter["id"]], "workspace_residency": workspace} + + def learn(self, adapter, reply, grant=None, lr=0.001): + """A completion's `learn` object, inside custody for this reply.""" + learn = {"reply": reply, "lora_id": adapter["id"], "grant": grant or self.grant, "lr": lr} + return dict(learn, **self.custody(adapter, {"learn": learn})) + + def experience(self, adapter, messages, grant=None, lr=0.001, **request): + """A /learn body for a served conversation ending in the reply to learn.""" + body = dict(request, messages=messages, lora_id=adapter["id"], grant=grant or self.grant, lr=lr) + return dict(body, **self.custody(adapter, {"messages": messages, **request})) + + def filling(self, prompt, unit, slot=0): + """The longest repetition of `unit` her slot's window holds as a reply after `prompt`: the + reply's tokens and its end of turn, counted by the server's own tokenizer with the reply's + semantics (no specials added, specials parsed). Preflight refuses an example past the window, + so a long learn is sized from it, never a fixed repeat count.""" + def rows(text, special): + return len(self.ok("/tokenize", {"content": text, "add_special": special})["tokens"]) + room = self.get("/slots")[slot]["n_ctx"] - rows(prompt, True) - 1 + repeats = room // rows(unit, False) + while repeats > 1 and rows(unit * repeats, False) > room: + repeats -= 1 + assert repeats > 1, ("the served window holds no multi-sentence reply", room) + return unit * repeats + + # ── retirement ───────────────────────────────────────────────────────────────────────────── + def unload(self, adapter): + """Retire an adapter; the script asserts on the returned (status, body).""" + outcome = self.post("/lora-adapters/unload", {"id": adapter["id"]}) + if outcome[0] == 200: + self.shadows.remove(adapter["id"]) + self.optimizers.pop(adapter["id"], None) + return outcome + + def close(self): + # Idempotent (an explicit close, then atexit). Adapters first (each settles its learning residency), then the references, then grants. + for sid in list(self.shadows): + status, body = self.post("/lora-adapters/unload", {"id": sid}) + assert status == 200 and body.get("retired"), ("close could not retire", sid, status, body) + for rid in reversed(self.references): + assert self.ok("/learn/residency/release", {"id": rid})["refusal"] == "none", rid + for gid in self.grants: + self.ok("/learn/grants/revoke", {"id": gid}) + self.shadows, self.references, self.grants = [], [], [] From c94de1d846afdd238c58c6e82a1bb2106bfc6fb4 Mon Sep 17 00:00:00 2001 From: joelteply Date: Tue, 10 Nov 2026 21:05:09 -0600 Subject: [PATCH 2/2] tests: a learn's workspace custody covers the previous registration, what is held, and its own need The step workspace is engine-wide and its custody passes from one reference to the next; claim_plan refuses ('uncovered') a replacement that does not cover every row the previous reference was registered with. Measured on CPU (stories260K, -np 2): A's long learn registered 7,026,688 bytes, the scheduler then held 4,907,008, and the reloaded B's short reply needed 2,573,344; sized to its own need (or even to need plus held) B's workspace was refused every time. The fixture now remembers the rows it last registered and covers the union of those, shared_compute_held and the example's own requirement, each buffer at the largest. learn-unload-during-learn.py then passes 10/10 on the fixed engine (fork #108). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FRQWzgSfo79JtnwZywHKrE --- tools/server/tests/learn_fixture.py | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/tools/server/tests/learn_fixture.py b/tools/server/tests/learn_fixture.py index 9b4fbf8d0178..2188f561b7e2 100644 --- a/tools/server/tests/learn_fixture.py +++ b/tools/server/tests/learn_fixture.py @@ -37,6 +37,7 @@ def __init__(self, url, name, timeout=600): self.references, self.grants, self.shadows = [], [], [] self.optimizers = {} self.minted = 0 # adapter names are immutable files: never reuse one + self.workspace_rows = [] # the rows the current workspace reference was registered with # A failed assertion must not leave custody bound for the next script on this server. atexit.register(self.close) self.grant = self.idle_grant() @@ -139,8 +140,20 @@ def custody(self, adapter, example): "-optimizer-%d" % adapter["id"], {"loaded": {"lora_id": adapter["id"], "path": adapter["path"]}}, [row for row in rows if row["kind"] != "shared_compute_required"]) - workspace = self.register("-workspace-%d" % len(self.references), {"workspace": True}, - [row for row in rows if row["kind"] == "shared_compute_required"]) + # The step workspace is engine-wide and its custody passes from one reference to the + # next: a replacement must cover every row the previous reference was registered with + # (claim_plan), which can exceed what the scheduler holds now. So it covers the union of + # the previous registration, what is held (shared_compute_held) and this example's need, + # each buffer at the largest of them. + workspace_rows = {} + for row in self.workspace_rows + plan["requirements"]: + if row["kind"] not in ("shared_compute_required", "shared_compute_held"): + continue + key = (row["device"], row["buffer_type"], row["host"]) + bytes_ = max(row["bytes"], workspace_rows[key]["bytes"]) if key in workspace_rows else row["bytes"] + workspace_rows[key] = dict(row, kind="shared_compute_required", bytes=bytes_) + self.workspace_rows = [row for row in workspace_rows.values() if row["bytes"] > 0] + workspace = self.register("-workspace-%d" % len(self.references), {"workspace": True}, self.workspace_rows) return {"optimizer_residency": self.optimizers[adapter["id"]], "workspace_residency": workspace} def learn(self, adapter, reply, grant=None, lr=0.001):