Skip to content
2 changes: 1 addition & 1 deletion ggml/src/ggml-moe-gguf-source.h
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ class GgufMoeSource {
const int total = std::atoi(first.substr(of + pat.size(), 5).c_str());
const size_t idx0 = first.rfind('-', of - 1);
const std::string prefix = first.substr(0, idx0 + 1), suffix = first.substr(of);
for (int i = 1; i <= total; i++) { char n[8]; std::snprintf(n, sizeof(n), "%05d", i); files.push_back(prefix + n + suffix); }
for (int i = 1; i <= total; i++) { char n[16]; std::snprintf(n, sizeof(n), "%05d", i); files.push_back(prefix + n + suffix); }
}
for (const auto & p : files) {
Shard s; gguf_init_params gp{}; gp.no_alloc = true; gp.ctx = nullptr;
Expand Down
2 changes: 1 addition & 1 deletion tools/moe-fit/fit-device.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ static std::vector<std::string> shard_paths(const std::string & first) {
const int total = std::atoi(first.substr(of + pat.size(), 5).c_str());
const size_t idx0 = first.rfind('-', of - 1);
const std::string prefix = first.substr(0, idx0 + 1), suffix = first.substr(of);
for (int i = 1; i <= total; i++) { char b[8]; std::snprintf(b, sizeof(b), "%05d", i); files.push_back(prefix + b + suffix); }
for (int i = 1; i <= total; i++) { char b[16]; std::snprintf(b, sizeof(b), "%05d", i); files.push_back(prefix + b + suffix); }
return files;
}

Expand Down
73 changes: 73 additions & 0 deletions tools/server/server-context.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2473,6 +2473,42 @@ struct server_context_impl {
send_error(task, "Unable to save slot", ERROR_TYPE_SERVER);
break;
}
// Continuum fork: the CONTEXT CHECKPOINTS are part of the register file.
// On a hybrid/recurrent (or SWA) cache the next request can only reuse a
// prefix by rolling back to a checkpoint at or before the divergence —
// and even an identical prompt rolls back one token [TAG_PROMPT_LOGITS].
// A restored slot with no checkpoints therefore always hits do_reset:
// "restore ok", then a full re-prefill (measured 2026-09-04, n_restored
// 11298 → cache_n 0). The state file carries KV + tokens only, so the
// checkpoints ride in a sidecar beside it; a missing or unreadable
// sidecar degrades to the pre-fork cold restore, never to an error.
{
const std::string ckpt_path = filepath + ".ckpt";
std::ofstream out(ckpt_path, std::ios::binary | std::ios::trunc);
auto put = [&](const void * ptr, size_t n) { out.write(reinterpret_cast<const char *>(ptr), n); };
const uint32_t magic = 0x434B5054u; // 'CKPT'
const uint32_t version = 1;
const uint32_t n_ckpt = (uint32_t) slot->prompt.checkpoints.size();
put(&magic, sizeof(magic)); put(&version, sizeof(version)); put(&n_ckpt, sizeof(n_ckpt));
for (const auto & c : slot->prompt.checkpoints) {
const int64_t n_tokens = c.n_tokens;
const int32_t pos_min = c.pos_min;
const int32_t pos_max = c.pos_max;
const uint64_t sz_tgt = c.data_tgt.size();
const uint64_t sz_dft = c.data_dft.size();
const uint64_t sz_spec = c.data_spec.size();
put(&n_tokens, sizeof(n_tokens)); put(&pos_min, sizeof(pos_min)); put(&pos_max, sizeof(pos_max));
put(&sz_tgt, sizeof(sz_tgt)); put(&sz_dft, sizeof(sz_dft)); put(&sz_spec, sizeof(sz_spec));
put(c.data_tgt.data(), c.data_tgt.size());
put(c.data_dft.data(), c.data_dft.size());
put(c.data_spec.data(), c.data_spec.size());
}
if (!out) {
SLT_WRN(*slot, "checkpoint sidecar could not be written: %s (the restore will be cold)\n", ckpt_path.c_str());
} else {
SLT_INF(*slot, "saved %u context checkpoints beside the slot state: %s\n", n_ckpt, ckpt_path.c_str());
}
}

const int64_t t_end = ggml_time_us();
const double t_save_ms = (t_end - t_start) / 1000.0;
Expand Down Expand Up @@ -2533,6 +2569,43 @@ struct server_context_impl {

slot->prompt.clear();
slot->prompt.tokens = std::move(restored);
// Continuum fork: bring the context checkpoints back with the state (see
// SLOT_SAVE). Any defect in the sidecar leaves the slot exactly as a
// pre-fork restore left it — valid tokens + state, no checkpoints.
{
const std::string ckpt_path = filepath + ".ckpt";
std::ifstream in(ckpt_path, std::ios::binary);
std::list<common_prompt_checkpoint> loaded;
bool ok = (bool) in;
auto get = [&](void * ptr, size_t n) { in.read(reinterpret_cast<char *>(ptr), n); ok = ok && (bool) in; };
if (ok) {
uint32_t magic = 0, version = 0, n_ckpt = 0;
get(&magic, sizeof(magic)); get(&version, sizeof(version)); get(&n_ckpt, sizeof(n_ckpt));
ok = ok && magic == 0x434B5054u && version == 1 && n_ckpt <= 4096;
for (uint32_t i = 0; ok && i < n_ckpt; ++i) {
int64_t n_tokens = 0; int32_t pos_min = 0, pos_max = 0;
uint64_t sz_tgt = 0, sz_dft = 0, sz_spec = 0;
get(&n_tokens, sizeof(n_tokens)); get(&pos_min, sizeof(pos_min)); get(&pos_max, sizeof(pos_max));
get(&sz_tgt, sizeof(sz_tgt)); get(&sz_dft, sizeof(sz_dft)); get(&sz_spec, sizeof(sz_spec));
// a checkpoint is at most one sequence state; refuse absurd sizes
ok = ok && sz_tgt < (1ull << 36) && sz_dft < (1ull << 36) && sz_spec < (1ull << 30);
if (!ok) break;
auto & c = loaded.emplace_back();
c.update_pos(n_tokens, pos_min, pos_max);
c.id_task = -1;
c.data_tgt.resize(sz_tgt); get(c.data_tgt.data(), sz_tgt);
c.data_dft.resize(sz_dft); get(c.data_dft.data(), sz_dft);
c.data_spec.resize(sz_spec); get(c.data_spec.data(), sz_spec);
}
}
if (ok) {
slot->prompt.checkpoints = std::move(loaded);
SLT_INF(*slot, "restored %zu context checkpoints from the sidecar: %s\n", slot->prompt.checkpoints.size(), ckpt_path.c_str());
} else {
slot->prompt.checkpoints.clear();
SLT_WRN(*slot, "no usable checkpoint sidecar (%s): the restore is cold — the next divergence re-prefills from the head\n", ckpt_path.c_str());
}
}
} catch (const std::exception & err) {
slot->prompt_clear();
send_error(task, std::string("Unable to restore slot: ") + err.what(), ERROR_TYPE_INVALID_REQUEST);
Expand Down
35 changes: 24 additions & 11 deletions tools/server/tests/unit/test_slot_save.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,11 @@ def test_deferred_restore_is_bound_to_its_slot():
global server
server.server_slots = True # /slots GET to observe processing
server.n_slots = 2
server.n_predict = 512
# The long slot must OUTLAST the whole restore round-trip on a fast CI runner (a 512-token
# gen of the tiny model ends in ~2 s on hosted ubuntu/windows runners — measured 2026-09-13:
# the binding assertion then failed for timing, not for binding). 4096 keeps slot 1 busy.
server.n_predict = 4096
server.n_ctx = 8192 # 4096 per slot: room for the long slot to actually run 4096 tokens
server.start()

# A reusable page on disk: process a prompt on slot 0, save it, let slot 0 idle.
Expand All @@ -181,20 +185,23 @@ def test_deferred_restore_is_bound_to_its_slot():

done0 = threading.Event()
done1 = threading.Event()
t_done = {}

def gen(id_slot, n_predict, done):
try:
for _ in server.make_stream_request("POST", "/completion", data={
"prompt": "Tell me a very long and detailed story about a cat.",
"id_slot": id_slot, "n_predict": n_predict, "cache_prompt": True, "stream": True,
"ignore_eos": True, # the tiny model emits EOS within a few tokens: the long slot must really run long
}):
pass
finally:
t_done[id_slot] = time.time()
done.set()

# slot 0 finishes QUICKLY (short), slot 1 runs LONG. The returner wants slot 0.
t0 = threading.Thread(target=gen, args=(0, 24, done0), daemon=True)
t1 = threading.Thread(target=gen, args=(1, 512, done1), daemon=True)
# slot 0 finishes FIRST (256 tokens: long enough that a fast runner can observe it busy), slot 1 runs LONG. The returner wants slot 0.
t0 = threading.Thread(target=gen, args=(0, 128, done0), daemon=True)
t1 = threading.Thread(target=gen, args=(1, 4096, done1), daemon=True)
t1.start()
# ensure slot 1 is processing before slot 0 starts, so the slot-1 op queues first
deadline = time.time() + 20
Expand Down Expand Up @@ -233,15 +240,21 @@ def restore_slot1():

assert res.status_code == 200, f"slot-0 restore failed: {res.status_code} {res.body}"
assert res.body["n_restored"] > 0
# The binding proof: the slot-0 restore returned while slot 1 was STILL generating.
# Before the fix it starved until slot 1 finished (done1 set); with the bind it is
# serviced the moment its own slot (0) releases.
assert not done1.is_set(), \
"slot-0 restore only returned after the long slot-1 gen ended — it was not bound to its slot"
assert elapsed < 20, f"slot-0 restore took {elapsed:.1f}s — starved behind the long slot"

t_ret = time.time()
t1.join(timeout=DEFAULT_HTTP_TIMEOUT)
tr1.join(timeout=DEFAULT_HTTP_TIMEOUT)
# The binding proof, in TIMESTAMPS (2026-09-13: "returned while slot 1 was still
# generating" was a race against the runner's speed — a tiny model ends 4096
# tokens in ~4 s on Metal and ~2 s on hosted runners). With the bind, the slot-0
# restore is serviced the moment ITS slot releases: it returns within a beat of
# slot 0's own generation ending. Unbound, it waited for the LONG slot: it returns
# only after slot 1 ended. So: the restore's return sits next to done0, not done1.
after_own = t_ret - t_done[0]
gap = t_done[1] - t_done[0]
if gap < 1.0:
pytest.skip(f"runner too fast to separate the slots (slot 1 ended {gap:.2f}s after slot 0)")
assert after_own < gap / 2, \
f"slot-0 restore returned {after_own:.2f}s after its own slot freed (slot 1 ended {gap:.2f}s later) — not bound to its slot"


def test_slot_restore_legacy_token_list():
Expand Down
Loading