server-models.cpp

#include "server-common.h"
#include "http.h"
#include "server-models.h"
#include "server-context.h"
#include "server-stream.h"

#include "build-info.h"
#include "preset.h"
#include "download.h"
#include "fit.h"
#include "hf-cache.h"
#include "http.h"
#include "subproc.h"

#include <cpp-httplib/httplib.h> // TODO: remove this once we use HTTP client from download.h
#include <optional>

#include <functional>
#include <optional>
#include <algorithm>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <cstring>
#include <cstdlib>
#include <atomic>
#include <chrono>
#include <deque>
#include <queue>
#include <filesystem>
#include <fstream>
#include <random>
#include <sstream>
#include <cstring>

#ifndef _WIN32
extern char **environ;
#endif

#define DEFAULT_STOP_TIMEOUT 10 // seconds

// note: SIGPIPE is ignored by the server
static void request_child_exit(server_subproc & proc) {
    FILE * stdin_file = proc.sproc.stdin_file();
    if (stdin_file) {
        fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);
        fflush(stdin_file);
    }
}

// single-threaded, watching all child processes at once
struct server_monitor {
    server_monitor(server_models & models) : models(models) {
        th = std::thread([this]() { run(); });
    }

    ~server_monitor() {
        push({ cmd_t::QUIT, {}, "", 0, false });
        th.join();
    }

    // thread-safe
    // kill_after > 0: force-kill the child after that many seconds
    void watch(const std::string & name, std::shared_ptr<server_subproc> proc, server_role_type role, int port, int kill_after = 0) {
        child_t c;
        c.name = name;
        c.proc = std::move(proc);
        c.role = role;
        c.port = port;
        if (kill_after > 0) {
            c.deadline = ggml_time_ms() + (int64_t) kill_after * 1000;
        }
        if (!c.proc->has_output()) {
            SRV_ERR("failed to get stdout/stderr of child process for name=%s\n", name.c_str());
            c.eof = true;
        }
        push({ cmd_t::WATCH, std::move(c), "", 0, false });
    }

    // thread-safe
    void stop(const std::string & name, int stop_timeout, bool send_exit) {
        push({ cmd_t::STOP, {}, name, stop_timeout, send_exit });
    }

private:
    struct child_t {
        std::string name;
        std::shared_ptr<server_subproc> proc;
        server_role_type role = SERVER_ROLE_INSTANCE;
        int port = 0;
        std::string buf[SERVER_SUBPROC_STREAMS]; // partial line of each pipe
        bool eof = false;     // output closed, waiting for the process to be reaped
        int64_t deadline = 0; // force-kill time in ms, 0 when no stop is pending
    };

    struct cmd_t {
        enum { WATCH, STOP, QUIT } type;
        child_t child;
        std::string name;
        int  stop_timeout;
        bool send_exit;
    };

    void push(cmd_t && cmd) {
        {
            std::lock_guard<std::mutex> lk(mu);
            cmds.push_back(std::move(cmd));
        }
        waiter.wake();
    }

    // returns true if the loop should exit
    bool handle_commands() {
        std::deque<cmd_t> batch;
        {
            std::lock_guard<std::mutex> lk(mu);
            batch.swap(cmds);
        }
        for (auto & cmd : batch) {
            switch (cmd.type) {
                case cmd_t::WATCH:
                    children.push_back(std::move(cmd.child));
                    break;
                case cmd_t::STOP:
                    // the newest child with this name is the one the registry knows
                    for (auto it = children.rbegin(); it != children.rend(); ++it) {
                        // a measure child has the same name, skip it
                        if (it->name != cmd.name || it->role == SERVER_ROLE_MEASURE) {
                            continue;
                        }
                        if (cmd.send_exit && !it->eof) {
                            request_child_exit(*it->proc);
                        }
                        it->deadline = ggml_time_ms() + (int64_t) cmd.stop_timeout * 1000;
                        break;
                    }
                    break;
                case cmd_t::QUIT:
                    return true;
            }
        }
        return false;
    }

    // read what the child wrote, handle its commands and forward its logs, line by line
    void read_output(child_t & c) {
        for (int i = 0; i < SERVER_SUBPROC_STREAMS; i++) {
            read_stream(c, (server_subproc_stream) i);
        }
        c.eof = c.proc->output_closed();
    }

    void read_stream(child_t & c, server_subproc_stream stream) {
        char chunk[4096];
        std::string & buf = c.buf[stream];
        bool closed = false;
        while (true) {
            int n = c.proc->read_output(stream, chunk, sizeof(chunk));
            if (n < 0) {
                closed = true;
                break;
            }
            if (n == 0) {
                break;
            }
            buf.append(chunk, (size_t) n);
            size_t start = 0;
            while (true) {
                size_t nl = buf.find('\n', start);
                if (nl == std::string::npos) {
                    break;
                }
                on_line(c, stream, buf.substr(start, nl + 1 - start));
                start = nl + 1;
            }
            buf.erase(0, start);
            if (buf.size() > max_line) {
                buf.clear(); // a child that never writes a newline must not grow this without bound
            }
        }
        if (closed && !buf.empty()) {
            on_line(c, stream, buf);
            buf.clear();
        }
    }

    void on_line(child_t & c, server_subproc_stream stream, const std::string & line) {
        if (stream == SERVER_SUBPROC_STDERR) {
            LOG("[%5d] %s", c.port, line.c_str()); // forward log
        } else if (string_starts_with(line, CMD_CHILD_TO_ROUTER_STATE)) {
            LOG_DBG("[%5d] %s", c.port, line.c_str()); // prevent spamming the log
            models.handle_child_state(c.name, line);
        } else {
            SRV_WRN("[%5d] unexpected output on the command pipe: %s", c.port, line.c_str());
        }
    }

    void run() {
        while (true) {
            if (handle_commands()) {
                return;
            }

            // wait for output, a wakeup, or the next deadline;
            // a child whose output closed is polled for its exit every 50 ms
            int64_t now     = ggml_time_ms();
            int64_t timeout = -1;
            for (const auto & c : children) {
                if (c.eof) {
                    timeout = timeout < 0 ? 50 : std::min<int64_t>(timeout, 50);
                }
                if (c.deadline) {
                    int64_t d = std::max<int64_t>(0, c.deadline - now);
                    timeout = timeout < 0 ? d : std::min(timeout, d);
                }
            }
            std::vector<server_subproc *> procs;
            std::vector<child_t *>        owners;
            for (auto & c : children) {
                if (!c.eof) {
                    procs.push_back(c.proc.get());
                    owners.push_back(&c);
                }
            }
            std::vector<bool> ready;
            waiter.wait(procs, ready, timeout);
            for (size_t i = 0; i < owners.size(); i++) {
                if (ready[i]) {
                    read_output(*owners[i]);
                }
            }

            // deadlines and exits
            now = ggml_time_ms();
            for (auto it = children.begin(); it != children.end();) {
                if (it->deadline && now >= it->deadline && !it->proc->stopped.load(std::memory_order_acquire)) {
                    SRV_WRN("force-killing child process for name=%s after timeout\n", it->name.c_str());
                    it->proc->terminate();
                    it->deadline = 0;
                }
                if (it->eof && !it->proc->is_alive()) {
                    int exit_code = it->proc->join();
                    it->proc->stopped.store(true, std::memory_order_release);
                    models.on_child_exit(it->name, it->proc, it->role, exit_code);
                    SRV_INF("instance name=%s exited with status %d\n", it->name.c_str(), exit_code);
                    it = children.erase(it);
                } else {
                    ++it;
                }
            }
        }
    }

    static constexpr size_t max_line = 1024 * 1024;

    server_models & models;
    std::mutex mu;
    std::deque<cmd_t> cmds;
    std::vector<child_t> children; // monitor thread only
    server_subproc::waiter waiter;
    std::thread th;
};

// short loopback budget for the resumable stream router to child JSON calls (probe, lookup,
// delete). distinct from params.timeout_read/write which only applies to the generation proxy
static constexpr int STREAM_LOOKUP_TIMEOUT_MS = 250;

static void unset_reserved_args(common_preset & preset, bool unset_model_args) {
    preset.unset_option("LLAMA_ARG_SSL_KEY_FILE");
    preset.unset_option("LLAMA_ARG_SSL_CERT_FILE");
    preset.unset_option("LLAMA_API_KEY");
    preset.unset_option("LLAMA_ARG_API_KEY_FILE");
    preset.unset_option("LLAMA_ARG_MODELS_DIR");
    preset.unset_option("LLAMA_ARG_MODELS_MAX");
    preset.unset_option("LLAMA_ARG_MODELS_MEMORY_MARGIN");
    preset.unset_option("LLAMA_ARG_MODELS_PRESET");
    preset.unset_option("LLAMA_ARG_MODELS_AUTOLOAD");
    preset.unset_option("LLAMA_ARG_CONNECT");
    preset.unset_option("LLAMA_ARG_CONNECT_CODE");
    if (unset_model_args) {
        preset.unset_option("LLAMA_ARG_MODEL");
        preset.unset_option("LLAMA_ARG_MMPROJ");
        preset.unset_option("LLAMA_ARG_ALIAS");
        preset.unset_option("LLAMA_ARG_HF_REPO");
    }
}

static std::vector<std::string> get_environment() {
    std::vector<std::string> env;

#ifdef _WIN32
    LPWCH env_block = GetEnvironmentStringsW();
    if (!env_block) {
        return env;
    }
    for (LPWCH e = env_block; *e; e += wcslen(e) + 1) {
        env.emplace_back(wstring_to_utf8(e));
    }
    FreeEnvironmentStringsW(env_block);
#else
    if (environ == nullptr) {
        return env;
    }
    for (char ** e = environ; *e != nullptr; e++) {
        env.emplace_back(*e);
    }
#endif

    return env;
}

void server_model_meta::update_args(common_preset_context & ctx_preset, std::string bin_path) {
    // update params
    unset_reserved_args(preset, false);
    preset.set_option(ctx_preset, "LLAMA_ARG_HOST",  CHILD_ADDR);
    preset.set_option(ctx_preset, "LLAMA_ARG_PORT",  std::to_string(port));
    preset.set_option(ctx_preset, "LLAMA_ARG_ALIAS", name);
    // the child output goes through the router to its terminal, so it follows the router colors
    preset.set_option(ctx_preset, "LLAMA_ARG_LOG_COLORS", common_log_get_colors(common_log_main()) ? "on" : "off");
    // TODO: maybe validate preset before rendering ?
    // render args
    args = preset.to_args(bin_path);

    // unified binary dispatches by subcommand, re-inject it right after the
    // binary path so the child starts as 'llama serve ...' not 'llama ...'
    const char * app_cmd = std::getenv("LLAMA_APP_CMD");
    if (app_cmd != nullptr && app_cmd[0] != '\0' && !bin_path.empty()) {
        args.insert(args.begin() + 1, app_cmd);
    }
}

void server_model_meta::update_caps(const common_params & base) {
    // reset to the default so a failed refresh cannot keep old values
    architecture = server_model_architecture_json(false, false, false, {"text"});
    n_ctx_train  = 0;

    // resolve the model file offline; do not download
    common_params params;
    params.model = base.model;
    // --no-mmproj applies to child models and blocks auto-attached projectors
    params.no_mmproj = base.no_mmproj;
    try {
        preset.apply_to_params(params, {
            "LLAMA_ARG_MODEL",
            "LLAMA_ARG_MODEL_URL",
            "LLAMA_ARG_MMPROJ",
            "LLAMA_ARG_MMPROJ_URL",
            "LLAMA_ARG_MMPROJ_AUTO",
            "LLAMA_ARG_HF_REPO",
            "LLAMA_ARG_HF_FILE",
        });
        params.offline = true;
        common_models_handler handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
        common_models_handler_apply(handler, params);
    } catch (const std::exception & e) {
        LOG_WRN("failed to resolve the model of '%s': %s\n", name.c_str(), e.what());
        return;
    }

    bool inp_image = false;
    bool inp_audio = false;
    bool gen_audio = false;
    try {
        if (!params.no_mmproj && !params.mmproj.path.empty()) {
            mtmd_caps caps = mtmd_get_cap_from_file(params.mmproj.path.c_str());
            inp_image = caps.inp_vision;
            inp_audio = caps.inp_audio;
            gen_audio = caps.gen_audio;
        }
    } catch (const std::exception & e) {
        LOG_WRN("failed to read the multimodal capabilities of '%s': %s\n", name.c_str(), e.what());
        // keep the output modalities from the GGUF metadata
    }

    // read the output modalities and the trained context from the GGUF metadata
    std::vector<std::string> output_modalities = {"text"};
    if (!params.model.path.empty()) {
        const common_gguf_info info = common_get_gguf_info(params.model.path);
        // same rule as mtmd_helper_get_capabilities()
        const bool can_chat = !gen_audio || info.has_chat_tmpl;
        output_modalities = server_model_output_modalities(info.decision_type, can_chat, gen_audio);
        n_ctx_train       = info.n_ctx_train;
    }

    // offline discovery cannot see video; a loaded model reports it
    architecture = server_model_architecture_json(inp_image, inp_audio, false, output_modalities);
}

//
// server_models
//

server_models::server_models(
        const common_params & params,
        int argc,
        char ** argv)
            : ctx_preset(LLAMA_EXAMPLE_SERVER),
              base_params(params),
              base_env(get_environment()),
              base_preset(ctx_preset.load_from_args(argc, argv)),
              sched(std::make_unique<server_models_sched>(*this)),
              monitor(std::make_unique<server_monitor>(*this)) {
    // propagate base params to child
    unset_reserved_args(base_preset, true);

    // do not propagate these options, but allow preset to explicitly set them
    base_preset.unset_option("LLAMA_ARG_LOG_FILE");

    // set binary path
    try {
        bin_path = get_server_exec_path().string();
    } catch (const std::exception & e) {
        bin_path = argv[0];
        SRV_WRN("failed to get server executable path: %s\n", e.what());
        SRV_WRN("using original argv[0] as fallback: %s\n", argv[0]);
    }

    unfinished_downloads.load();
    load_models();
    debug_fake_timing = !common_get_env("LLAMA_SERVER_DEBUG_FAKE_TIMING").empty();
}

server_models::~server_models() = default;

void server_models::instance_t::request_exit() const {
    request_child_exit(*subproc);
}

void server_models::add_model(server_model_meta && meta) {
    if (mapping.find(meta.name) != mapping.end()) {
        throw std::runtime_error(string_format("model '%s' appears multiple times", meta.name.c_str()));
    }

    // check model name does not conflict with existing aliases
    for (const auto & [key, inst] : mapping) {
        if (inst.meta.aliases.count(meta.name)) {
            throw std::runtime_error(string_format("model name '%s' conflicts with alias of model '%s'",
                meta.name.c_str(), key.c_str()));
        }
    }

    // parse aliases from preset's --alias option (comma-separated)
    std::string alias_str;
    if (meta.preset.get_option("LLAMA_ARG_ALIAS", alias_str) && !alias_str.empty()) {
        for (auto & alias : string_split<std::string>(alias_str, ',')) {
            alias = string_strip(alias);
            if (!alias.empty()) {
                meta.aliases.insert(alias);
            }
        }
    }

    // parse tags from preset's --tags option (comma-separated)
    std::string tags_str;
    if (meta.preset.get_option("LLAMA_ARG_TAGS", tags_str) && !tags_str.empty()) {
        for (auto & tag : string_split<std::string>(tags_str, ',')) {
            tag = string_strip(tag);
            if (!tag.empty()) {
                meta.tags.insert(tag);
            }
        }
    }

    // validate aliases do not conflict with existing names or aliases
    for (const auto & alias : meta.aliases) {
        if (mapping.find(alias) != mapping.end()) {
            throw std::runtime_error(string_format("alias '%s' for model '%s' conflicts with existing model name",
                alias.c_str(), meta.name.c_str()));
        }
        for (const auto & [key, inst] : mapping) {
            if (inst.meta.aliases.count(alias)) {
                throw std::runtime_error(string_format("alias '%s' for model '%s' conflicts with alias of model '%s'",
                    alias.c_str(), meta.name.c_str(), key.c_str()));
            }
        }
    }

    meta.update_args(ctx_preset, bin_path); // render args
    meta.update_caps(base_params);
    std::string name = meta.name;
    mapping[name] = instance_t{
        /* subproc */ std::make_shared<server_subproc>(),
        /* meta    */ std::move(meta)
    };
}

static bool has_sidecar(const std::vector<std::string> & sidecars, const std::string & kind) {
    return std::find(sidecars.begin(), sidecars.end(), kind) != sidecars.end();
}

// reject a sidecar kind no download can pull
static void check_sidecar(const std::string & kind) {
    std::vector<std::string> kinds;
    for (const auto & s : common_download_sidecars()) {
        kinds.emplace_back(s.name);
    }
    if (!has_sidecar(kinds, kind)) {
        throw std::invalid_argument("unknown sidecar '" + kind + "', expected one of " + string_join(kinds, ", "));
    }
}

// the sidecars a download asks for; by default the ones -hf pulls without a draft type
static std::vector<std::string> parse_sidecars(const json & body) {
    std::vector<std::string> defaults;
    for (const auto & s : common_download_sidecars()) {
        if (!s.draft) {
            defaults.emplace_back(s.name);
        }
    }
    std::vector<std::string> sidecars = json_value(body, "sidecars", defaults);
    for (const auto & kind : sidecars) {
        check_sidecar(kind);
    }
    return sidecars;
}

// the download options of a model with its sidecars, the plan a removal purges partials from
static common_download_opts sidecar_download_opts(const common_params & params, const std::vector<std::string> & sidecars) {
    common_download_opts opts;
    opts.bearer_token = params.hf_token;
    for (const auto & kind : sidecars) {
        common_download_opts_add_sidecar(opts, kind);
    }
    return opts;
}

// the HF repo a model resolves from: a preset names its own, a cached model is
// listed under it, and a local file or a models dir entry has none
static std::string model_hf_repo(const server_model_meta & meta) {
    std::string hf_repo;
    if (meta.preset.get_option("LLAMA_ARG_HF_REPO", hf_repo) && !hf_repo.empty()) {
        return hf_repo;
    }
    return meta.source == SERVER_MODEL_SOURCE_CACHE ? meta.name : "";
}

// the sidecar kinds of a HF repo present on disk, as the offline plan resolves them
static json cached_sidecars(const std::string & hf_repo) {
    common_params_model model;
    model.hf_repo = hf_repo;

    common_download_opts opts;
    opts.offline = true;
    for (const auto & s : common_download_sidecars()) {
        opts.*s.opt = true;
    }

    return common_download_plan_sidecars(common_download_get_hf_plan(model, opts));
}

// replace any inherited value of `key` in a child environment
static void set_child_env(std::vector<std::string> & env, const std::string & key, const std::string & value) {
    env.erase(std::remove_if(env.begin(), env.end(), [&](const std::string & e) {
        return e.rfind(key + "=", 0) == 0;
    }), env.end());
    env.push_back(key + "=" + value);
}

// where the unfinished downloads are remembered, inside the model cache
static std::string unfinished_downloads_path() {
    return hf_cache::get_cache_path() + "/unfinished-downloads.json";
}

void server_unfinished_downloads::load() {
    std::ifstream in(unfinished_downloads_path());
    if (!in) {
        return;
    }
    std::string raw((std::istreambuf_iterator<char>(in)), std::istreambuf_iterator<char>());
    std::lock_guard<std::mutex> lk(mutex);
    try {
        json data = json::parse(raw);
        for (const auto & d : data.value("downloads", json::array())) {
            server_unfinished_download download;
            download.model    = json_value(d, "model", std::string());
            download.sidecars = json_value(d, "sidecars", std::vector<std::string>());
            for (const auto & p : d.value("partials", json::array())) {
                download.partials.push_back({ json_value(p, "kind", std::string()), json_value(p, "path", std::string()) });
            }
            if (!download.model.empty()) {
                downloads.push_back(std::move(download));
            }
        }
    } catch (const std::exception & e) {
        LOG_WRN("failed to read unfinished downloads: %s\n", e.what());
    }
}

void server_unfinished_downloads::save() {
    json data;
    data["downloads"] = json::array();
    for (const auto & d : downloads) {
        json partials = json::array();
        for (const auto & p : d.partials) {
            partials.push_back({ {"kind", p.kind}, {"path", p.path} });
        }
        data["downloads"].push_back({ {"model", d.model}, {"sidecars", d.sidecars}, {"partials", partials} });
    }
    try {
        fs_write_atomic(std::filesystem::u8path(unfinished_downloads_path()), data.dump());
    } catch (const std::exception & e) {
        LOG_WRN("failed to record unfinished downloads: %s\n", e.what());
    }
}

void server_unfinished_downloads::delete_partials(const std::vector<common_download_partial> & parts) {
    for (const auto & p : parts) {
        const bool shared = std::any_of(downloads.begin(), downloads.end(), [&](const server_unfinished_download & d) {
            return std::any_of(d.partials.begin(), d.partials.end(), [&](const common_download_partial & q) {
                return q.path == p.path;
            });
        });
        if (shared) {
            continue;
        }
        std::error_code ec;
        std::filesystem::remove(std::filesystem::u8path(p.path), ec);
    }
}

void server_unfinished_downloads::add(const server_unfinished_download & download) {
    std::lock_guard<std::mutex> lk(mutex);
    std::vector<common_download_partial> stale;
    auto it = std::find_if(downloads.begin(), downloads.end(), [&](const auto & d) { return d.model == download.model; });
    if (it != downloads.end()) {
        stale = std::move(it->partials);
        *it = download;
    } else {
        downloads.push_back(download);
    }
    save();
    delete_partials(stale);
}

void server_unfinished_downloads::forget(const std::string & model) {
    std::lock_guard<std::mutex> lk(mutex);
    auto it = std::find_if(downloads.begin(), downloads.end(), [&](const auto & d) { return d.model == model; });
    if (it != downloads.end()) {
        downloads.erase(it);
        save();
    }
}

bool server_unfinished_downloads::remove(const std::string & model) {
    std::lock_guard<std::mutex> lk(mutex);
    auto it = std::find_if(downloads.begin(), downloads.end(), [&](const auto & d) { return d.model == model; });
    if (it == downloads.end()) {
        return false;
    }
    auto partials = std::move(it->partials);
    downloads.erase(it);
    save();
    delete_partials(partials);
    return true;
}

bool server_unfinished_downloads::remove_sidecar(const std::string & model, const std::string & kind) {
    std::lock_guard<std::mutex> lk(mutex);
    auto it = std::find_if(downloads.begin(), downloads.end(), [&](const auto & d) { return d.model == model; });
    if (it == downloads.end() || !has_sidecar(it->sidecars, kind)) {
        return false;
    }
    it->sidecars.erase(std::remove(it->sidecars.begin(), it->sidecars.end(), kind), it->sidecars.end());

    std::vector<common_download_partial> dropped;
    auto & partials = it->partials;
    auto pos = std::stable_partition(partials.begin(), partials.end(), [&](const common_download_partial & p) {
        return p.kind != kind;
    });
    dropped.assign(pos, partials.end());
    partials.erase(pos, partials.end());

    save();
    delete_partials(dropped);
    return true;
}

std::vector<server_unfinished_download> server_unfinished_downloads::get_all() {
    std::lock_guard<std::mutex> lk(mutex);
    return downloads;
}

std::vector<server_unfinished_download> server_models::unfinished_downloads_get() {
    return unfinished_downloads.get_all();
}

void server_models::unfinished_downloads_remove(const std::string & name) {
    // under the lock, so a download of the same tag cannot start while its files go
    std::lock_guard<std::mutex> lk(mutex);
    if (mapping.find(name) != mapping.end()) {
        throw std::invalid_argument("model '" + name + "' is downloading");
    }
    // a name with no record is no model: its files belong to the cached tags of the repo
    if (!unfinished_downloads.remove(name)) {
        throw std::runtime_error("model name=" + name + " is not found");
    }
    bool ok = common_download_remove(name);
    SRV_INF("removing unfinished download name=%s (%s)\n", name.c_str(), ok ? "succeeded" : "nothing cached");
}

bool server_models::remove_sidecar(const std::string & name, const std::string & kind) {
    std::lock_guard<std::mutex> lk(mutex);
    auto it = mapping.find(name);
    if (it != mapping.end() && (it->second.meta.is_running() || it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING)) {
        throw std::invalid_argument("model '" + name + "' is in use, unload it first");
    }
    // a sidecar that only ever landed as a partial is still something removed
    bool ok = unfinished_downloads.remove_sidecar(name, kind);
    ok = common_download_remove_sidecars(name, sidecar_download_opts(base_params, { kind })) || ok;
    SRV_INF("removing sidecar %s of model name=%s (%s)\n", kind.c_str(), name.c_str(), ok ? "succeeded" : "nothing cached");
    return ok;
}

void server_models::notify_sse(const std::string & event, const std::string & model_id, const json & data) {
    std::unique_ptr<server_task_result_router> result = std::make_unique<server_task_result_router>();
    result->data = {
        {"model", model_id},
        {"event", event},
    };
    if (!data.is_null()) {
        result->data["data"] = data;
    }
    SRV_DBG("notifying SSE clients about event '%s' for model '%s': %s\n", event.c_str(), model_id.c_str(), safe_json_to_str(result->data).c_str());
    sse.broadcast(std::move(result));
}

void server_models::load_models() {
    // Phase 1: load presets from all sources - pure I/O, no lock needed
    // 1. cached models
    common_presets cached_models = ctx_preset.load_from_cache();
    SRV_TRC("Loaded %zu cached model presets from %s\n", cached_models.size(), hf_cache::get_cache_path().c_str());
    // 2. local models from --models-dir
    common_presets local_models;
    if (!base_params.models_dir.empty()) {
        local_models = ctx_preset.load_from_models_dir(base_params.models_dir);
        SRV_TRC("Loaded %zu local model presets from %s\n", local_models.size(), base_params.models_dir.c_str());
    }
    // 3. custom-path models from presets
    common_preset global = {};
    common_presets custom_presets = {};
    if (!base_params.models_preset.empty()) {
        custom_presets = ctx_preset.load_from_ini(base_params.models_preset, global);
        SRV_TRC("Loaded %zu custom model presets from %s\n", custom_presets.size(), base_params.models_preset.c_str());
    }

    // cascade, apply global preset first
    cached_models  = ctx_preset.cascade(global, cached_models);
    local_models   = ctx_preset.cascade(global, local_models);
    custom_presets = ctx_preset.cascade(global, custom_presets);

    // note: if a model exists in both cached and local, local takes precedence
    common_presets final_presets;
    std::unordered_map<std::string, server_model_source> source_map;
    for (const auto & [name, preset] : cached_models) {
        final_presets[name] = preset;
        source_map[name] = SERVER_MODEL_SOURCE_CACHE;
    }
    for (const auto & [name, preset] : local_models)  {
        final_presets[name] = preset;
        source_map[name] = SERVER_MODEL_SOURCE_MODELS_DIR;
    }
    for (const auto & [name, custom] : custom_presets) {
        if (final_presets.find(name) != final_presets.end()) {
            final_presets[name].merge(custom);
        } else {
            final_presets[name] = custom;
        }
        source_map[name] = SERVER_MODEL_SOURCE_PRESET;
    }

    // overlay router's own CLI args on top of every model preset so that
    // e.g. `llama-server --temp 0` is honoured by all child processes
    for (auto & [name, preset] : final_presets) {
        preset.merge(base_preset);
    }

    auto get_source = [&](const std::string & name) {
        return source_map.count(name) ? source_map.at(name) : SERVER_MODEL_SOURCE_PRESET;
    };

    // hide cache models whose resolved file is already used by a preset with dedup-cache-models enabled
    std::set<std::string> hidden_models;
    {
        std::set<std::string> preset_paths;
        auto add_hf_path = [&preset_paths](const common_preset & preset, const char * repo_key, const char * file_key) {
            std::string hf_repo;
            if (!preset.get_option(repo_key, hf_repo) || hf_repo.empty()) {
                return;
            }
            std::string hf_file;
            preset.get_option(file_key, hf_file);
            std::string path = common_download_resolve_path(hf_repo, hf_file);
            if (!path.empty()) {
                preset_paths.insert(path);
            }
        };
        for (const auto & [name, preset] : custom_presets) {
            std::string val;
            if (!preset.get_option(COMMON_ARG_PRESET_DEDUP_CACHE_MODELS, val) || !common_arg_utils::is_truthy(val)) {
                continue;
            }
            add_hf_path(preset, "LLAMA_ARG_HF_REPO", "LLAMA_ARG_HF_FILE");
            add_hf_path(preset, "LLAMA_ARG_SPEC_DRAFT_HF_REPO", "LLAMA_ARG_SPEC_DRAFT_MODEL");
        }
        if (!preset_paths.empty()) {
            for (const auto & [name, preset] : cached_models) {
                if (get_source(name) != SERVER_MODEL_SOURCE_CACHE) {
                    continue; // merged with another source, not a pure cache entry
                }
                std::string path = common_download_resolve_path(name);
                if (!path.empty() && preset_paths.count(path)) {
                    SRV_INF("hiding cache model name=%s (deduplicated by a preset)\n", name.c_str());
                    hidden_models.insert(name);
                }
            }
        }
    }

    // Helpers that read `mapping` - must be called while holding the lock.
    auto join_set = [](const std::set<std::string> & s) {
        std::string result;
        for (const auto & v : s) {
            if (!result.empty()) result += ", ";
            result += v;
        }
        return result;
    };
    auto log_available_models = [&]() {
        SRV_INF("Available models (%zu):\n", mapping.size());
        if (mapping.empty()) {
            SRV_INF("%s", "  no models found on the system (visit https://llama.app/models for suggestions)\n");
        } else {
            for (const auto & [name, inst] : mapping) {
                const std::string source = server_model_source_to_string(inst.meta.source);

                std::string info;
                if (!inst.meta.aliases.empty()) info += " (aliases: " + join_set(inst.meta.aliases) + ")";
                if (!inst.meta.tags.empty())    info += " [tags: "    + join_set(inst.meta.tags)    + "]";

                SRV_INF("  [%10s] %s%s\n", source.c_str(), name.c_str(), info.c_str());
            }
        }
    };
    auto apply_stop_timeout = [&]() {
        for (auto & [name, inst] : mapping) {
            std::string val;
            if (inst.meta.preset.get_option(COMMON_ARG_PRESET_STOP_TIMEOUT, val)) {
                try {
                    inst.meta.stop_timeout = std::stoi(val);
                } catch (...) {
                    SRV_WRN("invalid stop-timeout value '%s' for model '%s', using default %d seconds\n",
                        val.c_str(), name.c_str(), DEFAULT_STOP_TIMEOUT);
                    inst.meta.stop_timeout = DEFAULT_STOP_TIMEOUT;
                }
            }
        }
    };
    auto apply_hidden = [&]() {
        for (auto & [name, inst] : mapping) {
            inst.meta.hidden = hidden_models.count(name) > 0;
        }
    };
    // update_args() injects HOST/PORT/ALIAS/LOG_COLORS, so strip them before comparing presets
    auto preset_options_for_compare = [](common_preset p) {
        p.unset_option("LLAMA_ARG_HOST");
        p.unset_option("LLAMA_ARG_PORT");
        p.unset_option("LLAMA_ARG_ALIAS");
        p.unset_option("LLAMA_ARG_LOG_COLORS");
        return p.options;
    };

    // Phase 2: acquire the lock once for all mapping mutations.
    // We temporarily release it only when calling functions that acquire it internally (unload)
    std::unique_lock<std::mutex> lk(mutex);

    need_reload = false;
    bool is_first_load = mapping.empty();

    if (is_first_load) {
        // FIRST LOAD: add all models, then unlock for autoloading
        for (const auto & [name, preset] : final_presets) {
            server_model_meta meta{
                /* source        */ get_source(name),
                /* preset        */ preset,
                /* name          */ name,
                /* aliases       */ {},
                /* tags          */ {},
                /* port          */ 0,
                /* status        */ SERVER_MODEL_STATUS_UNLOADED,
                /* last_used     */ 0,
                /* args          */ std::vector<std::string>(),
                /* loaded_info   */ {},
                /* progress      */ {},
                /* exit_code     */ 0,
                /* stop_timeout  */ DEFAULT_STOP_TIMEOUT,
                // /* need_download */ false,
            };
            add_model(std::move(meta));
        }
        apply_stop_timeout();
        apply_hidden();
        log_available_models();
        sched->measure_unmeasured(lk);

        // skipped on reload, see startup_models
        if (startup_models.has_value()) {
            std::vector<std::string> models_to_load;
            for (const auto & [name, inst] : mapping) {
                std::string val;
                if (inst.meta.preset.get_option(COMMON_ARG_PRESET_LOAD_ON_STARTUP, val) && common_arg_utils::is_truthy(val)) {
                    models_to_load.push_back(name);
                }
            }
            if ((int)models_to_load.size() > base_params.models_max) {
                throw std::runtime_error(string_format(
                    "number of models to load on startup (%zu) exceeds models_max (%d)",
                    models_to_load.size(), base_params.models_max));
            }

            // to be lazy-loaded after main() setup phase is completed
            startup_models = std::move(models_to_load);
        }

        lk.unlock();
    } else {
        // RELOAD: diff the new preset list against the current mapping and reconcile
        is_reloading = true;

        // find running models whose source was removed or whose preset changed
        std::vector<std::string> to_unload;
        for (const auto & [name, inst] : mapping) {
            if (!inst.meta.is_running()) continue;
            auto it = final_presets.find(name);
            if (it == final_presets.end()) {
                to_unload.push_back(name); // removed from source
            } else if (preset_options_for_compare(inst.meta.preset) != preset_options_for_compare(it->second)) {
                to_unload.push_back(name); // preset changed
            }
        }

        // unload() acquires the lock internally, so release before each call
        for (const auto & name : to_unload) {
            SRV_INF("(reload) unloading model name=%s (source updated or removed)\n", name.c_str());
            lk.unlock();
            unload(name);
            lk.lock();
        }

        // wait for all targeted models to reach UNLOADED; cv.wait handles unlock/relock
        cv.wait(lk, [&]() {
            for (const auto & name : to_unload) {
                auto it = mapping.find(name);
                if (it != mapping.end() && it->second.meta.is_running()) return false;
            }
            return true;
        });

        // erase models no longer in any source
        for (auto it = mapping.begin(); it != mapping.end(); ) {
            if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
                ++it; // download thread is still busy, skip
            } else if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADED) {
                // download finished, safe to erase
                it = mapping.erase(it);
            } else if (final_presets.find(it->first) == final_presets.end()) {
                SRV_INF("(reload) removing model name=%s (no longer in source)\n", it->first.c_str());
                it = mapping.erase(it);
            } else {
                ++it;
            }
        }

        // update presets for non-running models still in source
        for (auto & [name, inst] : mapping) {
            if (inst.meta.is_running()) continue;
            auto it = final_presets.find(name);
            if (it == final_presets.end()) continue; // erased above

            if (preset_options_for_compare(inst.meta.preset) != preset_options_for_compare(it->second)) {
                // different args, different memory: measure again
                inst.meta.memory = {};
            }
            inst.meta.preset = it->second;

            // re-parse aliases, then validate against other models
            std::set<std::string> new_aliases;
            std::string alias_str;
            if (inst.meta.preset.get_option("LLAMA_ARG_ALIAS", alias_str) && !alias_str.empty()) {
                for (auto & alias : string_split<std::string>(alias_str, ',')) {
                    alias = string_strip(alias);
                    if (!alias.empty()) new_aliases.insert(alias);
                }
            }
            inst.meta.aliases.clear();
            for (const auto & alias : new_aliases) {
                bool conflict = false;
                for (const auto & [other_name, other_inst] : mapping) {
                    if (other_name == name) continue;
                    if (other_name == alias || other_inst.meta.aliases.count(alias)) {
                        SRV_WRN("(reload) alias '%s' for model '%s' conflicts with model '%s', skipping\n",