diff --git a/tools/server/tests/dream-adapter-routes.py b/tools/server/tests/dream-adapter-routes.py
index 4b8d7a88126..69919d81b45 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 f2dfa510525..7d6e5c61848 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 ab60067dd33..9e4cd890aea 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 92d0940c09d..671c8347139 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 044b6f48c1e..c6452a4b8ea 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 6469c98f9bb..5e08798bb8a 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 6402bed6046..b9763ca5213 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 00000000000..2188f561b7e
--- /dev/null
+++ b/tools/server/tests/learn_fixture.py
@@ -0,0 +1,201 @@
+"""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
+        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()
+
+    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"])
+        # 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):
+        """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 = [], [], []