#include "ling3/chat_protocol.h"
#include "ling3/chat_sampling.h"
#include "ling3/flow_gate.h"
#include "ling3/context_options.h"
#include "ling3/decoder.h"
#include "ling3/model_package.h"
#include "ling3/rknn_backend.h"
#include "ling3/runtime_requirements.h"
#include "ling3/tokenizer.h"
#include "ling3/w4_linear.h"
#include "ling3/state_cache.h"
#include "mla_npu.h"

#include <arpa/inet.h>
#include <sys/ioctl.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/utsname.h>
#include <fcntl.h>
#include <poll.h>
#include <sched.h>
#include <unistd.h>

#include <atomic>
#include <charconv>
#include <chrono>
#include <condition_variable>
#include <csignal>
#include <cstring>
#include <deque>
#include <filesystem>
#include <fstream>
#include <functional>
#include <iostream>
#include <mutex>
#include <numeric>
#include <random>
#include <sstream>
#include <thread>

namespace {
using namespace ling3;
using namespace ling3::chat;
using Clock = std::chrono::steady_clock;
volatile std::sig_atomic_t quitting = 0;
void OnSignal(int) { quitting = 1; }
double Ms(Clock::time_point a, Clock::time_point b = Clock::now()) {
    return std::chrono::duration<double, std::milli>(b - a).count();
}
std::string Read(const std::filesystem::path & p) {
    std::ifstream f(p, std::ios::binary);
    return {std::istreambuf_iterator<char>(f), {}};
}
double MemoryMiB(std::string_view field, const char * file = "/proc/meminfo") {
    std::istringstream s(Read(file));
    std::string line;
    while (std::getline(s, line)) if (line.starts_with(field)) {
        std::istringstream value(line.substr(field.size()));
        double kb = 0; value >> kb; return kb / 1024;
    }
    return 0;
}
void Configure() {
    // A deployed engine always uses the measured FP32-state + NPU W4A8 path.
    // Ignore ambient experiment settings; no external graph or tokenizer files.
    for (auto key : {"LING3_PREFILL_W4A4", "LING3_EXPERT_ALL_CORES", "LING3_GDN_PREFILL_DIR",
                     "LING3_DISABLE_BATCH_INPUT_REUSE", "LING3_DISABLE_PARALLEL_GATHER"}) unsetenv(key);
    for (auto key : {"LING3_PREWARM_EXPERTS", "LING3_EXPERT_BALANCED", "LING3_EXPERT_ZERO_COPY",
                     "LING3_GDN_CPU_PREFILL", "LING3_GDN_CPU_FP32_STATE", "LING3_GDN_CPU_DECODE",
                     "LING3_GDN_FULL_FP32", "LING3_MLA_SIMD", "LING3_VECTOR_MATH"}) setenv(key, "1", 1);
    cpu_set_t mask; CPU_ZERO(&mask);
    for (int i = 4; i < 8; ++i) CPU_SET(i, &mask);
    if (sched_setaffinity(0, sizeof(mask), &mask) != 0)
        throw std::runtime_error("cannot use A76 CPU4-7: " + std::string(std::strerror(errno)));
    rlimit r {};
    if (getrlimit(RLIMIT_NOFILE, &r) != 0) throw std::runtime_error("getrlimit failed");
    r.rlim_cur = std::min<rlim_t>(262144, r.rlim_max);
    if (setrlimit(RLIMIT_NOFILE, &r) != 0 || r.rlim_cur < 65536)
        std::cerr << "resource_warning: file descriptor limit is below the qualified budget; continuing, actual opens may fail\n";
}
struct DrmVersion {
    int major, minor, patch;
    std::size_t name_len; char * name;
    std::size_t date_len; char * date;
    std::size_t desc_len; char * desc;
};
Json CheckDevice() {
#if !LING3_WITH_ICU
    throw std::runtime_error("non-production build: exact ICU tokenizer support is required");
#endif
    utsname u {}; uname(&u);
    const auto compatible = Read("/proc/device-tree/compatible");
    if (std::string(u.machine) != "aarch64" || compatible.find("rockchip,rk3588") == std::string::npos)
        throw std::runtime_error("unsupported device: requires RK3588/RK3588S, 64-bit Linux");
    const auto memory = MemoryMiB("MemTotal:");
    if (memory < 14000)
        std::cerr << "memory_warning: physical RAM " << memory
                  << " MiB is below the previously tested 16 GB class; continuing, available memory may be insufficient\n";
    Json report {{"soc", "RK3588"}, {"kernel", u.release}, {"memory_mib", memory},
                 {"minimum_rknpu_driver", "0.9.8"}, {"cpu_affinity", "4-7"}};
    bool found = false;
    std::error_code ec;
    for (const auto & entry : std::filesystem::directory_iterator("/dev/dri", ec)) {
        if (!entry.path().filename().string().starts_with("renderD")) continue;
        const int fd = open(entry.path().c_str(), O_RDWR | O_CLOEXEC);
        if (fd < 0) continue;
        char name[128] {};
        DrmVersion v {}; v.name = name; v.name_len = sizeof(name) - 1;
        const int rc = ioctl(fd, _IOWR('d', 0x00, DrmVersion), &v);
        close(fd);
        if (rc != 0 || std::string(name) != "rknpu") continue;
        report["npu_node"] = entry.path().string();
        report["npu_driver"] = std::to_string(v.major) + "." + std::to_string(v.minor) + "." + std::to_string(v.patch);
        // Conservative qualification floor, not a claim about every older SDK.
        if (!QualifiedRknpuDriver(v.major, v.minor, v.patch))
            std::cerr << "driver_warning: RKNPU " << report["npu_driver"].get<std::string>()
                << " is below qualified 0.9.8; continuing to real hardware capability checks\n";
        found = true; break;
    }
    if (!found) throw std::runtime_error("no accessible RKNPU DRM device; check rknpu kernel driver and render/video group permissions (Panthor GPU is not NPU)");
    Configure();
    if (!RknnBackendAvailable()) throw std::runtime_error("this binary was built without RKNN");
    // Exercise the actual native INT4 matmul path on each core, not just version text.
    std::vector<std::byte> w(256 * 64 / 2, std::byte{0x11});
    std::vector<float> scales(64, 1), input(256, 1), out(64);
    std::vector<float> batch_input(128 * 256), batch_output(128 * 64);
    std::vector<std::int32_t> correction(64, 8 * 256);
    for (int core = 0; core < 3; ++core) {
        DynamicW4Linear linear({256, 64, 1, {core}, 0}, w, scales, correction);
        linear.Run(input, out);
        for (auto x : out) if (!std::isfinite(x) || std::abs(x - 256) > 2)
            throw std::runtime_error("native INT4 numerical probe failed on NPU core " + std::to_string(core));
        // Check the FD-view capability required by shared batch A/C before
        // allocating the full model. Large-to-small transitions catch stale IO.
        float value = 1;
        for (std::size_t rows : {128, 1, 3, 16, 2}) {
            std::fill(batch_input.begin(), batch_input.end(), value);
            linear.RunBatch(std::span<const float>(batch_input).first(rows * 256), rows,
                            std::span<float>(batch_output).first(rows * 64));
            for (std::size_t i = 0; i < rows * 64; ++i)
                if (!std::isfinite(batch_output[i]) || std::abs(batch_output[i] - 256 * value) > 2)
                    throw std::runtime_error("shared batch A/C FD-view probe failed on NPU core " + std::to_string(core));
            value = -value;
        }
    }
    report["w4a8_native_int4_three_core_probe"] = "passed";
    report["shared_batch_ac_fd_view_probe"] = "passed";
    report["mla_backend"] = std::getenv("LING3_MLA_BACKEND") ? std::getenv("LING3_MLA_BACKEND") : "auto";
    {
        MlaNpu attention;
        std::vector<float> q(16*16*192,0), values(16*16*128);
        std::vector<std::uint16_t> k(272*16*192,0), v(272*16*128,0x3e80); // BF16 0.25.
        bool ok=attention.Run(q,k,v,16,272,values);
        if(ok)for(float x:values)if(!std::isfinite(x)||std::abs(x-0.25F)>0.0001F)
            throw std::runtime_error("MLA FP16 dynamic input probe returned incorrect values");
        if(ok){
            std::fill(v.begin(),v.end(),0xbf00); // BF16 -0.5; detect stale B after rebinding.
            ok=attention.Run(q,k,v,16,272,values);
            if(ok)for(float x:values)if(!std::isfinite(x)||std::abs(x+0.5F)>0.0001F)
                throw std::runtime_error("MLA FP16 dynamic update probe returned incorrect values");
        }
        report["mla_fp16_dynamic_probe"] = ok ? "passed" : "CPU_attention_fallback_or_selected";
    }
    report["available_memory_mib"] = MemoryMiB("MemAvailable:");
    std::cout << "device_check=" << report.dump() << std::endl;
    return report;
}

struct Result { std::string text, reason = "length"; Json usage, metrics; };
class Engine {
    ModelPackage package_;
    std::size_t context_;
    Tokenizer tokenizer_;
    Decoder decoder_;
    std::vector<float> logits_;
    std::mutex mutex_;
    std::ofstream log_;
    FlowGate flow_;
    std::atomic<std::uint64_t> evaluated_steps_ {0};
    struct PrefixCache {
        std::vector<std::uint32_t> tokens;
        std::string user;
        std::size_t aligned_position = 0;
        std::shared_ptr<DecoderCheckpoint> aligned, exact;
        std::shared_ptr<const DecoderState> aligned_state, exact_state;
        std::vector<std::uint32_t> continuation_tokens;
        std::shared_ptr<DecoderCheckpoint> continuation;
        std::shared_ptr<const DecoderState> continuation_state;
        bool generated_lineage = false;
        std::vector<float> logits;
        std::size_t bytes() const {
            std::size_t n = tokens.size() * sizeof(std::uint32_t) + logits.size() * sizeof(float);
            if (aligned) n += Decoder::CheckpointBytes(*aligned);
            if (exact && exact != aligned) n += Decoder::CheckpointBytes(*exact);
            if (aligned_state) n += aligned_state->bytes();
            if (exact_state && exact_state != aligned_state) n += exact_state->bytes();
            n += continuation_tokens.size() * sizeof(std::uint32_t);
            if (continuation) n += Decoder::CheckpointBytes(*continuation);
            if (continuation_state) n += continuation_state->bytes();
            return n;
        }
        void ClearContinuation() {
            continuation_tokens.clear(); continuation.reset(); continuation_state.reset();
        }
    } prefix_cache_;
    std::string resident_key_; // Empty for the anonymous resident slot.
    StateCache<PrefixCache> sessions_ {256ULL * 1024 * 1024};
    static std::string SessionKey(const std::string & user, const std::string & session) {
        return Json::array({user, session}).dump();
    }
    void ClearCacheUnlocked() { prefix_cache_ = {}; resident_key_.clear(); }
    static Tokenizer MakeTokenizer(const ModelPackage & p) {
        const auto & t = p.tensor("tokenizer");
        return Tokenizer({t.data, static_cast<std::size_t>(t.entry->data_bytes)});
    }
public:
    double initialization_ms = 0, warmup_ms = 0;
    Json WeightFormat() const {
        if(package_.header().flags & ling3::kPackageMixedW4W8){
            const auto& metadata=package_.tensor("precision.recipe");
            auto recipe=Json::parse(reinterpret_cast<const char*>(metadata.data),
                reinterpret_cast<const char*>(metadata.data)+metadata.entry->data_bytes);
            const char* shared=std::getenv("LING3_BRIDGE_SHARED_STAGE");
            return {{"source","self_contained_selective_w4_w8"},{"execution","mixed_W4A8_W8A8"},
                {"w8_families",recipe.at("w8_families")},
                {"calibrated_w4_families",recipe.at("calibrated_w4_families")},
                {"linear_counts",recipe.at("linear_counts")},
                {"shared_stage",!shared || std::string(shared)=="1"},{"experimental",true}};
        }
        const char* families=std::getenv("LING3_W8_FAMILIES");
        const char* calibrated=std::getenv("LING3_CALIBRATED_W4_FAMILIES");
        const char* shared=std::getenv("LING3_BRIDGE_SHARED_STAGE");
        if((families && *families)||(calibrated && *calibrated)) return {{"source","original_W4_with_selective_overrides"},
            {"execution","mixed_W4A8_W8A8"},{"w8_families",families?families:""},
            {"calibrated_w4_families",calibrated?calibrated:""},
            {"shared_stage",shared && std::string(shared)=="1"},{"experimental",true}};
        if (!(package_.header().flags & 0x100)) return {{"source","custom_per_channel_w4"},{"execution","W4A8"}};
        const std::string mode=std::getenv("LING3_OFFICIAL_EXECUTION")?std::getenv("LING3_OFFICIAL_EXECUTION"):"w8";
        return {{"source","official_int4_group32_with_high_precision_layers"},
                {"execution",mode=="w8"?"W8A8_bridge_requantized":"FP16_bridge"},
                {"experimental",true}};
    }
    explicit Engine(const std::string & model, const std::string & log, std::size_t context)
        : package_(model), context_(context), tokenizer_(MakeTokenizer(package_)), decoder_(package_, context),
          logits_(package_.header().vocab_size), log_(log, std::ios::app) {
        if (!log_) throw std::runtime_error("cannot open metrics log: " + log);
        for (std::size_t rows = 1; rows <= 128; rows *= 2) decoder_.PrepareBatch(rows);
    }
    bool active() const { return flow_.Active(); }
    bool Cancel(const std::string & id = {}) { return flow_.Cancel(id); }
    Json FlowStatus() const {
        auto status = flow_.Status();
        status["evaluated_steps"] = evaluated_steps_.load();
        return status;
    }
    void Ack(const std::string & id, std::uint64_t sequence) { flow_.Ack(id, sequence); }
    void Pause(const std::string & id) { flow_.Pause(id); }
    void Resume(const std::string & id) { flow_.Resume(id); }
    void ClearCache(const std::string & session = {}, const std::string & user = {}) {
        std::unique_lock lock(mutex_, std::try_to_lock);
        if (!lock.owns_lock()) throw Error(429, "engine_busy", "cannot clear cache during inference");
        if (session.empty()) { sessions_.Clear(); ClearCacheUnlocked(); decoder_.Reset(); }
        else {
            const auto key = SessionKey(user, session);
            sessions_.Erase(key);
            if (resident_key_ == key) { ClearCacheUnlocked(); decoder_.Reset(); }
        }
    }
    void SetCacheBudget(std::size_t bytes) { sessions_ = StateCache<PrefixCache>(bytes); }
    Json CacheStatus() {
        std::unique_lock lock(mutex_, std::try_to_lock);
        if (!lock.owns_lock()) throw Error(429, "engine_busy", "cache status requires an idle engine");
        return {{"sessions", sessions_.size()}, {"snapshot_bytes", sessions_.bytes()},
            {"budget_bytes", sessions_.limit()}, {"evictions", sessions_.evictions()},
            {"default_prefix_bytes", resident_key_.empty() ? prefix_cache_.bytes() : 0},
            {"resident_cache_bytes", prefix_cache_.bytes()}, {"resident_named", !resident_key_.empty()},
            {"persistent", false}};
    }
    void ForkCache(const std::string & user, const std::string & source, const std::string & target) {
        std::unique_lock lock(mutex_, std::try_to_lock);
        if (!lock.owns_lock()) throw Error(429, "engine_busy", "cannot fork during inference");
        const auto key = SessionKey(user, source);
        const auto * entry = sessions_.Get(key);
        if (!entry) throw Error(404, "session_not_found", "source cache is not resident");
        const auto copy = *entry;
        if (!sessions_.Put(SessionKey(user, target), copy, copy.bytes()))
            throw Error(409, "cache_budget_exceeded", "snapshot does not fit cache budget");
        if (resident_key_ == SessionKey(user, target)) { ClearCacheUnlocked(); decoder_.Reset(); }
    }
    std::size_t context() const { return context_; }
    Json WeightMemory() const {
        return {{"source_weight_discard_advised_mib", package_.discarded_weight_bytes() / 1048576.0},
                {"source_weight_pagecache_resident_mib", package_.resident_linear_weight_bytes() / 1048576.0},
                {"source_weight_cache_advice_failures", package_.cache_advice_failures()}};
    }
    void Warmup() {
        const auto start = Clock::now();
        auto r = Parse({{"model", kModel}, {"messages", Json::array({{{"role", "user"}, {"content", "你好"}}})}, {"max_tokens", 2}});
        // Warm exactly 128 input positions, independent of tokenizer text length.
        const auto ids = tokenizer_.Encode(r.prompt);
        std::vector<std::uint32_t> tokens(128);
        for (std::size_t i = 0; i < tokens.size(); ++i) tokens[i] = ids[i % ids.size()];
        decoder_.EvalBatch(tokens, logits_);
        decoder_.Eval(std::max_element(logits_.begin(), logits_.end()) - logits_.begin(), logits_);
        decoder_.Reset();
        warmup_ms = Ms(start);
    }
    Result Run(const Request & r, const std::string & id,
               const std::function<bool()> & begin,
               const std::function<bool(std::string_view)> & emit,
               const std::function<bool()> & connected) {
        const auto start = Clock::now();
        std::unique_lock lock(mutex_, std::try_to_lock);
        if (!lock.owns_lock()) throw Error(429, "engine_busy", "one inference at a time; retry after the active request finishes");
        rusage faults_before {}; getrusage(RUSAGE_SELF, &faults_before);
        const auto attention_before = decoder_.AttentionStats();
        const auto tokens = tokenizer_.Encode(r.prompt);
        const auto output_budget = OutputBudget(tokens.size(), context(), r.max_tokens);
        const bool named_session = !r.session_id.empty();
        const auto session_key = SessionKey(r.cache_user, r.session_id);
        // The active named session, like the anonymous slot, retains lightweight
        // checkpoints even when a full historical snapshot cannot fit the budget.
        if (named_session) {
            if (resident_key_ != session_key) {
                const auto * saved = r.cache_prompt ? sessions_.Get(session_key) : nullptr;
                prefix_cache_ = saved ? *saved : PrefixCache {};
            }
            if (!r.cache_prompt) sessions_.Erase(session_key);
        } else if (!resident_key_.empty()) ClearCacheUnlocked();
        // A strict request must never inherit the numerical lineage of an
        // incremental decode path, even through an exact prompt cache hit.
        if (!r.reuse_generated_state && prefix_cache_.generated_lineage) ClearCacheUnlocked();
        const std::size_t aligned_position = (tokens.size()/128)*128;
        const auto state_cost = [&](std::size_t position) {
            return 29638656ULL + decoder_.StateSignature().size() + position*61440ULL;
        };
        const std::size_t snapshot_cost = state_cost(tokens.size()) +
            ((aligned_position && aligned_position != tokens.size()) ? state_cost(aligned_position) : 0) +
            tokens.size()*sizeof(std::uint32_t) + logits_.size()*sizeof(float);
        const bool cache_named = named_session && r.cache_prompt && snapshot_cost <= sessions_.limit();
        if (output_budget < r.max_tokens)
            std::cerr << "request_info: output_budget=" << output_budget
                      << " remaining_context=" << context() - tokens.size()
                      << "; generation may continue until EOS, stop, cancellation or context capacity\n";
        flow_.Begin(id, r.flow_ack);
        evaluated_steps_ = 0;
        struct Guard {
            Engine & e; bool preserve = false;
            ~Guard() {
                if (!preserve) { e.ClearCacheUnlocked(); e.decoder_.Reset(); }
                e.flow_.End();
            }
        } guard {*this};
        const auto cache_begin = Clock::now();
        std::size_t cached_tokens = 0;
        bool exact_hit = false;
        std::size_t generated_cached_tokens = 0;
        std::string cache_status = r.cache_prompt ? "miss" : "disabled";
        if (r.cache_prompt && prefix_cache_.user == r.cache_user && (prefix_cache_.exact || prefix_cache_.exact_state) &&
            prefix_cache_.tokens == tokens) {
            cached_tokens = prefix_cache_.exact ? decoder_.RestoreCheckpoint(*prefix_cache_.exact)
                : decoder_.RestoreState(*prefix_cache_.exact_state);
            if (!prefix_cache_.exact) prefix_cache_.exact = decoder_.SaveCheckpoint();
            logits_ = prefix_cache_.logits;
            exact_hit = true;
            cache_status = "exact_prompt_hit";
        } else if (r.cache_prompt && r.reuse_generated_state && prefix_cache_.user == r.cache_user &&
                   (prefix_cache_.continuation || prefix_cache_.continuation_state) &&
                   tokens.size() > prefix_cache_.continuation_tokens.size() &&
                   std::equal(prefix_cache_.continuation_tokens.begin(), prefix_cache_.continuation_tokens.end(), tokens.begin())) {
            cached_tokens = prefix_cache_.continuation ? decoder_.RestoreCheckpoint(*prefix_cache_.continuation)
                : decoder_.RestoreState(*prefix_cache_.continuation_state);
            generated_cached_tokens = cached_tokens - prefix_cache_.tokens.size();
            prefix_cache_.generated_lineage = true;
            prefix_cache_.exact.reset(); prefix_cache_.exact_state.reset(); prefix_cache_.logits.clear();
            cache_status = "generated_prefix_hit";
        } else if (r.cache_prompt && r.reuse_generated_state && prefix_cache_.user == r.cache_user &&
                   (prefix_cache_.exact || prefix_cache_.exact_state) && tokens.size() > prefix_cache_.tokens.size() &&
                   std::equal(prefix_cache_.tokens.begin(), prefix_cache_.tokens.end(), tokens.begin())) {
            cached_tokens = prefix_cache_.exact ? decoder_.RestoreCheckpoint(*prefix_cache_.exact)
                : decoder_.RestoreState(*prefix_cache_.exact_state);
            prefix_cache_.generated_lineage = true;
            prefix_cache_.exact.reset(); prefix_cache_.exact_state.reset(); prefix_cache_.logits.clear();
            cache_status = "exact_prefix_hit";
        } else if (r.cache_prompt && prefix_cache_.user == r.cache_user && (prefix_cache_.aligned || prefix_cache_.aligned_state) &&
                   prefix_cache_.aligned_position > 0 && tokens.size() > prefix_cache_.aligned_position &&
                   std::equal(prefix_cache_.tokens.begin(),
                              prefix_cache_.tokens.begin() + prefix_cache_.aligned_position, tokens.begin())) {
            cached_tokens = prefix_cache_.aligned ? decoder_.RestoreCheckpoint(*prefix_cache_.aligned)
                : decoder_.RestoreState(*prefix_cache_.aligned_state);
            prefix_cache_.exact.reset();
            prefix_cache_.exact_state.reset();
            prefix_cache_.logits.clear();
            cache_status = "prefix_hit";
        } else {
            ClearCacheUnlocked(); decoder_.Reset();
        }
        prefix_cache_.ClearContinuation();
        resident_key_ = named_session ? session_key : "";
        const double cache_restore_ms = Ms(cache_begin);
        double cache_save_ms = 0;
        const auto aligned_target = (tokens.size() / 128) * 128;
        if (r.cache_prompt && !exact_hit) {
            prefix_cache_.tokens = tokens;
            prefix_cache_.user = r.cache_user;
        }
        Result result;
        std::size_t generated = 0, eval_count = 0, prompt_evaluated = 0;
        double decode_eval_ms = 0, prefill_eval_ms = 0, first_ms = 0;
        auto first = Clock::now(), last = first;
        auto alive = [&] { return !quitting && flow_.Alive() && connected(); };
        auto aborted = [&] { result.reason = "canceled"; flow_.Cancel(id); };
        bool began = begin();
        const auto prefill_start = Clock::now();
        auto progress_at = prefill_start;
        if (began) for (std::size_t offset = cached_tokens; offset < tokens.size();) {
            if (!alive() || !flow_.Wait([&] { return !quitting && connected(); })) { aborted(); break; }
            // Generated prefixes can end between block boundaries. Preserve
            // the aligned checkpoint even when the first resumed block is short.
            auto block = std::span<const std::uint32_t>(tokens).subspan(offset,
                std::min<std::size_t>(128 - offset % 128, tokens.size() - offset));
            const bool final = offset + block.size() == tokens.size();
            if (block.size() == 1) prefill_eval_ms += decoder_.Eval(block[0], logits_).total_ms;
            else prefill_eval_ms += (final ? decoder_.EvalBatch(block, logits_) : decoder_.EvalBatchState(block)).total_ms;
            ++evaluated_steps_;
            offset += block.size(); prompt_evaluated = offset - cached_tokens;
            if (r.cache_prompt && offset == aligned_target) {
                const auto saved_at = Clock::now();
                prefix_cache_.aligned = decoder_.SaveCheckpoint();
                prefix_cache_.aligned_state = cache_named ? decoder_.SaveState() : nullptr;
                prefix_cache_.aligned_position = offset;
                cache_save_ms += Ms(saved_at);
            }
            const auto now = Clock::now();
            if (Ms(progress_at, now) >= 10000) {
                const Json progress {{"id", id}, {"timestamp", std::time(nullptr)},
                    {"prompt_tokens", tokens.size()}, {"prompt_evaluated_tokens", prompt_evaluated},
                    {"cached_tokens", cached_tokens}, {"processed_position", offset},
                    {"elapsed_ms", Ms(start, now)}, {"prefill_eval_ms", prefill_eval_ms},
                    {"rss_mib", MemoryMiB("VmRSS:", "/proc/self/status")},
                    {"peak_rss_mib", MemoryMiB("VmHWM:", "/proc/self/status")},
                    {"swap_mib", MemoryMiB("VmSwap:", "/proc/self/status")},
                    {"available_memory_mib", MemoryMiB("MemAvailable:")}};
                std::cerr << "prefill_progress=" << progress.dump() << std::endl;
                progress_at = now;
            }
        } else aborted();
        if (result.reason != "canceled" && r.cache_prompt && !exact_hit) {
            const auto saved_at = Clock::now();
            prefix_cache_.exact = aligned_target == tokens.size() ? prefix_cache_.aligned : decoder_.SaveCheckpoint();
            prefix_cache_.exact_state = cache_named ? (aligned_target == tokens.size()
                ? prefix_cache_.aligned_state : decoder_.SaveState()) : nullptr;
            prefix_cache_.logits = logits_;
            cache_save_ms += Ms(saved_at);
        }
        const auto prefill_end = Clock::now();
        const bool retain_generated = r.cache_prompt && r.reuse_generated_state;
        std::vector<std::uint32_t> evaluated_tokens = retain_generated ? tokens : std::vector<std::uint32_t>{};
        TextFilter filter(r.stops);
        std::mt19937 random(r.seed);
        std::vector<bool> seen(logits_.size(), false);
        for (auto token : tokens) if (token < seen.size()) seen[token] = true;
        double flow_wait_ms = 0;
        std::size_t flow_wait_count = 0;
        auto output = [&](std::string text) {
            result.text += text;
            if (!text.empty()) {
                flow_.Prepare();
                if (!emit(text)) { aborted(); return false; }
                if (r.flow_ack) {
                    const auto wait_start = Clock::now();
                    const bool resumed = flow_.Wait([&] { return !quitting && connected(); });
                    flow_wait_ms += Ms(wait_start); ++flow_wait_count;
                    if (!resumed) { aborted(); return false; }
                }
            }
            return true;
        };
        if (result.reason != "canceled") for (std::size_t i = 0; i < output_budget; ++i) {
            if (!alive() || !flow_.Wait([&] { return !quitting && connected(); })) { aborted(); break; }
            const auto token = SampleToken(logits_, r, seen, random);
            seen[token] = true;
            last = Clock::now();
            if (!generated) { first = last; first_ms = Ms(start, first); }
            ++generated; // includes EOS and stop tokens, like completion usage.
            if (token == package_.header().eos_token) { result.reason = "stop"; break; }
            if (!output(filter.Push(tokenizer_.Piece(token)))) break;
            if (filter.stopped) { result.reason = "stop"; break; }
            if (i + 1 < output_budget) {
                if (!flow_.Wait([&] { return !quitting && connected(); })) { aborted(); break; }
                decode_eval_ms += decoder_.Eval(token, logits_).total_ms; ++eval_count;
                if (retain_generated) evaluated_tokens.push_back(token);
                ++evaluated_steps_;
            }
        }
        if (result.reason != "canceled") output(filter.Push({}, true));
        if (result.reason != "canceled" && !alive()) aborted();
        bool continuation_stored = false, continuation_skipped_budget = false;
        double continuation_save_ms = 0;
        if (result.reason != "canceled" && retain_generated && eval_count) {
            const auto cost = state_cost(decoder_.position()) + evaluated_tokens.size()*sizeof(std::uint32_t);
            {
                const auto save_begin = Clock::now();
                // Do not evaluate the final sampled token just for caching.
                // The stored prefix contains only tokens actually consumed.
                prefix_cache_.continuation = decoder_.SaveCheckpoint();
                // Historical backup cost excludes borrowed resident checkpoints.
                auto backup = prefix_cache_;
                backup.aligned.reset(); backup.exact.reset(); backup.continuation.reset();
                if (named_session) {
                    if (!cache_named || cost > sessions_.limit() || backup.bytes() > sessions_.limit() - cost)
                        continuation_skipped_budget = true;
                    else prefix_cache_.continuation_state = decoder_.SaveState();
                }
                prefix_cache_.continuation_tokens = std::move(evaluated_tokens);
                continuation_stored = true;
                continuation_save_ms = Ms(save_begin);
            }
        }
        if (result.reason != "canceled" && !alive()) aborted();
        if (result.reason != "canceled" && !flow_.TryCommit()) aborted();
        if (result.reason == "canceled") { prefix_cache_.ClearContinuation(); continuation_stored = false; }
        result.usage = {{"prompt_tokens", tokens.size()}, {"completion_tokens", generated},
                        {"total_tokens", tokens.size() + generated}};
        if (r.cache_prompt) result.usage["prompt_tokens_details"] = {{"cached_tokens", cached_tokens}};
        const auto decode_ms = generated > 1 ? Ms(first, last) : 0;
        result.metrics = {{"id", id}, {"model", kModel}, {"timestamp", std::time(nullptr)},
            {"finish_reason", result.reason}, {"usage", result.usage},
            {"ttft_ms", generated ? Json(first_ms) : Json(nullptr)}, {"prefill_ms", Ms(prefill_start, prefill_end)},
            {"prefill_eval_ms", prefill_eval_ms}, {"prompt_evaluated_tokens", prompt_evaluated},
            {"prefill_tokens_per_second", prefill_eval_ms > 0 ? 1000 * prompt_evaluated / prefill_eval_ms : 0},
            {"decode_ms", decode_ms}, {"decode_eval_ms", decode_eval_ms}, {"decode_eval_count", eval_count},
            {"decode_tokens_per_second", decode_ms > 0 ? 1000 * (generated - 1) / decode_ms : 0},
            {"total_ms", Ms(start)}, {"rss_mib", MemoryMiB("VmRSS:", "/proc/self/status")},
            {"peak_rss_mib", MemoryMiB("VmHWM:", "/proc/self/status")},
            {"queue_ms", 0}, {"cached_tokens", cached_tokens}, {"warmed", true},
            {"cache_status", cache_status}, {"cache_restore_ms", cache_restore_ms},
            {"cache_save_ms", cache_save_ms}, {"prefix_cache_mib", prefix_cache_.bytes() / 1048576.0}};
        result.metrics["context_length"] = context_;
        result.metrics["reuse_generated_state"] = r.reuse_generated_state;
        result.metrics["generated_state_lineage"] = prefix_cache_.generated_lineage;
        result.metrics["generated_cached_tokens"] = generated_cached_tokens;
        result.metrics["continuation_stored"] = continuation_stored;
        result.metrics["continuation_skipped_budget"] = continuation_skipped_budget;
        result.metrics["continuation_save_ms"] = continuation_save_ms;
        result.metrics["continuation_tokens"] = continuation_stored ? prefix_cache_.continuation_tokens.size() : 0;
        result.metrics["flow_wait_ms"] = flow_wait_ms;
        result.metrics["flow_wait_count"] = flow_wait_count;
        result.metrics["flow_control"] = r.flow_ack ? "ack" : "none";
        result.metrics["sampling"] = {{"temperature", r.temperature}, {"top_p", r.top_p},
            {"top_k", r.top_k}, {"repeat_penalty", r.repeat_penalty}, {"seed", r.seed},
            {"enable_thinking", r.enable_thinking}};
        if (named_session) {
            bool stored = false;
            if (result.reason != "canceled" && r.cache_prompt) {
                if (cache_named && prefix_cache_.exact_state) {
                    auto backup = prefix_cache_;
                    backup.aligned.reset(); backup.exact.reset(); backup.continuation.reset();
                    if (!backup.continuation_state) backup.continuation_tokens.clear();
                    stored = sessions_.Put(session_key, backup, backup.bytes());
                }
                else sessions_.Erase(session_key);
            }
            result.metrics["session_id"] = r.session_id;
            result.metrics["session_cache_stored"] = stored;
            result.metrics["session_cache_skipped_budget"] = r.cache_prompt && !cache_named;
            result.metrics["session_cache_bytes"] = sessions_.bytes();
            result.metrics["session_cache_evictions"] = sessions_.evictions();
        }
        result.metrics["decode_compute_tokens_per_second"] = decode_eval_ms > 0 ? 1000 * eval_count / decode_eval_ms : 0;
        result.metrics["requested_max_tokens"] = r.max_tokens == std::numeric_limits<std::size_t>::max()
            ? Json(nullptr) : Json(r.max_tokens);
        result.metrics["effective_max_tokens"] = output_budget;
        const bool context_reached = result.reason == "length" && generated == context() - tokens.size();
        result.metrics["context_capacity_reached"] = context_reached;
        if (context_reached)
            std::cerr << "request_info: context_capacity_reached id=" << id
                      << " capacity=" << context() << "; finished normally with reason=length\n";
        const auto attention_after = decoder_.AttentionStats();
        result.metrics["mla_npu_calls"] = attention_after.npu_calls - attention_before.npu_calls;
        result.metrics["mla_cpu_calls"] = attention_after.cpu_calls - attention_before.cpu_calls;
        result.metrics["mla_fallbacks"] = attention_after.fallbacks - attention_before.fallbacks;
        result.metrics["mla_npu_ms"] = attention_after.npu_ms - attention_before.npu_ms;
        rusage faults_after {}; getrusage(RUSAGE_SELF, &faults_after);
        result.metrics["minor_page_faults"] = faults_after.ru_minflt - faults_before.ru_minflt;
        result.metrics["major_page_faults"] = faults_after.ru_majflt - faults_before.ru_majflt;
        log_ << result.metrics.dump() << '\n'; log_.flush();
        std::cerr << "reply_metrics=" << result.metrics.dump() << std::endl;
        guard.preserve = result.reason != "canceled" && r.cache_prompt;
        return result;
    }
};

bool Send(int fd, std::string_view data) {
    while (!data.empty()) {
        const auto n = send(fd, data.data(), data.size(), MSG_NOSIGNAL);
        if (n < 0 && errno == EINTR) continue;
        if (n <= 0) return false;
        data.remove_prefix(n);
    }
    return true;
}
bool Connected(int fd) {
    pollfd p {fd, POLLIN | POLLRDHUP, 0};
    if (poll(&p, 1, 0) < 0) return false;
    if (p.revents & (POLLERR | POLLHUP | POLLRDHUP | POLLNVAL)) return false;
    return true;
}
void Response(int fd, int code, const Json & body) {
    const auto s = body.dump();
    Send(fd, "HTTP/1.1 " + std::to_string(code) + (code == 200 ? " OK" : " Error") +
        "\r\nContent-Type: application/json; charset=utf-8\r\nConnection: close\r\nContent-Length: " +
        std::to_string(s.size()) + "\r\n\r\n" + s);
}
struct Http { std::string method, path, body; };
Http ReadRequest(int fd) {
    std::string data; char buf[4096]; std::size_t end;
    const auto started = Clock::now();
    auto receive = [&] {
        if (Ms(started) > 10000 || quitting) throw Error(408, "request_timeout", "request read timed out");
        const auto n = recv(fd, buf, sizeof(buf), 0);
        if (n <= 0) throw Error(400, "invalid_http", "incomplete HTTP request");
        data.append(buf, n);
    };
    while ((end = data.find("\r\n\r\n")) == std::string::npos) {
        receive(); if (data.size() > 65536) throw Error(413, "too_large", "headers too large");
    }
    Http h;
    std::istringstream headers(data.substr(0, end));
    std::string line, version; std::getline(headers, line);
    std::istringstream request_line(line); request_line >> h.method >> h.path >> version;
    if (version != "HTTP/1.1" && version != "HTTP/1.0") throw Error(400, "invalid_http", "invalid HTTP request line");
    std::size_t length = 0; bool have_length = false;
    while (std::getline(headers, line)) {
        if (!line.empty() && line.back() == '\r') line.pop_back();
        auto colon = line.find(':');
        if (colon == std::string::npos) throw Error(400, "invalid_http", "invalid header");
        std::string name = line.substr(0, colon), value = line.substr(colon + 1);
        std::transform(name.begin(), name.end(), name.begin(), [](unsigned char c) { return std::tolower(c); });
        value.erase(0, value.find_first_not_of(" \t"));
        if (name == "transfer-encoding") throw Error(400, "invalid_http", "request Transfer-Encoding unsupported; send Content-Length");
        if (name == "content-length") {
            auto [p, ec] = std::from_chars(value.data(), value.data() + value.size(), length);
            if (have_length || ec != std::errc{} || p != value.data() + value.size())
                throw Error(400, "invalid_http", "invalid/duplicate Content-Length");
            have_length = true;
        }
    }
    if (length > data.max_size()-end-4) throw Error(400,"invalid_http","Content-Length exceeds addressable string size");
    if (length > 1024 * 1024)
        std::cerr << "request_warning: body_bytes=" << length << " exceeds previous 1 MiB budget; continuing\n";
    end += 4;
    while (data.size() - end < length) receive();
    h.body = data.substr(end, length);
    return h;
}
std::string Id() {
    static std::atomic_uint64_t count {0};
    return "chatcmpl-" + std::to_string(std::time(nullptr)) + "-" + std::to_string(++count);
}
void Handle(int fd, Engine & engine, const Json & device) {
    struct Guard { int fd; ~Guard() { close(fd); } } guard {fd};
    timeval read_timeout {2, 0}, write_timeout {5, 0};
    setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &read_timeout, sizeof(read_timeout));
    setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &write_timeout, sizeof(write_timeout));
    bool streaming_started = false;
    auto error = [&](const Error & e) {
        if (streaming_started) Send(fd, "data: " + ErrorBody(e).dump() + "\n\ndata: [DONE]\n\n");
        else Response(fd, e.status, ErrorBody(e));
    };
    try {
        const auto http = ReadRequest(fd);
        if (http.method == "GET" && http.path == "/health") {
            Response(fd, 200, {{"status", "ok"}, {"active", engine.active()}, {"device", device},
                {"context_length", engine.context()}, {"initialization_ms", engine.initialization_ms},
                {"warmup_ms", engine.warmup_ms}, {"weight_memory", engine.WeightMemory()},
                {"weight_format",engine.WeightFormat()}, {"mindnano_integration_version", 5},
                {"capabilities", {{"flow_control_ack", true}, {"pause_resume", true}, {"top_k", true},
                    {"repeat_penalty", true}, {"enable_thinking", true}, {"session_cache", true},
                    {"generated_state_reuse", true}, {"qa_cache_default", true}, {"resident_session_cache", true}}},
                {"generation", engine.FlowStatus()}}); return;
        }
        auto cache_field = [](const Json & body, const char * field, bool required) {
            if (!body.is_object() || (required && !body.contains(field)) ||
                (body.contains(field) && !body[field].is_string()))
                throw Error(400, "invalid_parameter", std::string(field)+" must be a string");
            auto value = body.value(field, std::string{});
            if (required && (value.empty() || value.size() > 256))
                throw Error(400, "invalid_parameter", std::string(field)+" must contain 1..256 bytes");
            return value;
        };
        auto cache_fields = [](const Json & body, std::initializer_list<std::string_view> allowed) {
            if (!body.is_object()) throw Error(400, "invalid_parameter", "cache request must be an object");
            for (auto it = body.begin(); it != body.end(); ++it)
                if (std::find(allowed.begin(), allowed.end(), it.key()) == allowed.end())
                    throw Error(400, "invalid_parameter", "unsupported cache field: " + it.key());
        };
        if (http.method == "GET" && http.path == "/v1/cache/status") {
            Response(fd, 200, engine.CacheStatus()); return;
        }
        if (http.method == "POST" && http.path == "/v1/cache/clear") {
            const auto body = Json::parse(http.body);
            cache_fields(body, {"user", "session_id"});
            const auto user = cache_field(body, "user", false);
            const auto session = cache_field(body, "session_id", body.contains("session_id"));
            if (!user.empty() && session.empty())
                throw Error(400, "invalid_parameter", "user-scoped clear requires session_id");
            engine.ClearCache(session, user); Response(fd, 200, {{"cleared", true}}); return;
        }
        if (http.method == "POST" && http.path == "/v1/cache/fork") {
            const auto body = Json::parse(http.body);
            cache_fields(body, {"user", "source_session_id", "target_session_id"});
            engine.ForkCache(cache_field(body, "user", false), cache_field(body, "source_session_id", true),
                             cache_field(body, "target_session_id", true));
            Response(fd, 200, {{"forked", true}}); return;
        }
        if (http.method == "GET" && http.path == "/v1/models") {
            Response(fd, 200, {{"object", "list"}, {"data", Json::array({{{"id", kModel}, {"object", "model"},
                {"created", 0}, {"owned_by", "mindnano"}, {"context_length", engine.context()}}})}}); return;
        }
        if (http.method == "POST" && http.path == "/v1/cancel") {
            const auto body = Json::parse(http.body);
            const auto canceled = engine.Cancel(body.value("request_id", std::string{}));
            Response(fd, 200, {{"cancel_requested", canceled}}); return;
        }
        if (http.method == "POST" && http.path == "/v1/flow/ack") {
            const auto body = Json::parse(http.body);
            if (!body.contains("sequence") || !body["sequence"].is_number_unsigned() ||
                !body.contains("request_id") || !body["request_id"].is_string())
                throw Error(400, "invalid_flow_control", "request_id and positive sequence are required");
            engine.Ack(body["request_id"].get<std::string>(), body["sequence"].get<std::uint64_t>());
            Response(fd, 200, {{"acknowledged", body["sequence"]}}); return;
        }
        if (http.method == "GET" && http.path == "/v1/generation/status") {
            Response(fd, 200, engine.FlowStatus()); return;
        }
        if (http.method == "POST" && (http.path == "/v1/generation/pause" || http.path == "/v1/generation/resume")) {
            const auto body = Json::parse(http.body);
            const auto id = body.at("request_id").get<std::string>();
            if (http.path == "/v1/generation/pause") engine.Pause(id);
            else engine.Resume(id);
            Response(fd, 200, engine.FlowStatus()); return;
        }
        if (http.method != "POST" || http.path != "/v1/chat/completions")
            throw Error(404, "not_found", "endpoint not found");
        const auto request = Parse(Json::parse(http.body));
        const auto id = Id(); const auto created = std::time(nullptr);
        auto chunk = [&](Json delta, Json finish = nullptr) {
            Json out {{"id", id}, {"object", "chat.completion.chunk"}, {"created", created}, {"model", kModel},
                {"choices", Json::array({{{"index", 0}, {"delta", delta}, {"finish_reason", finish}, {"logprobs", nullptr}}})}};
            if (request.include_usage) out["usage"] = nullptr;
            if (request.flow_ack && delta.contains("content") && delta["content"] != "")
                out["mindnano_flow"] = {{"request_id", id}, {"sequence", engine.FlowStatus()["sequence"]}};
            return Send(fd, "data: " + out.dump() + "\n\n");
        };
        const auto result = engine.Run(request, id, [&] {
            if (!request.stream) return Connected(fd);
            streaming_started = true;
            return Send(fd, "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream; charset=utf-8\r\n"
                "Cache-Control: no-cache\r\nX-Accel-Buffering: no\r\nConnection: close\r\n\r\n") &&
                chunk({{"role", "assistant"}, {"content", ""}});
        }, [&](std::string_view text) { return !request.stream || chunk({{"content", text}}); },
           [&] { return Connected(fd); });
        if (result.reason == "canceled") {
            error(Error(409, "request_canceled", "generation canceled; partial text is not a completed reply")); return;
        }
        if (request.stream) {
            chunk(Json::object(), result.reason);
            Json final {{"id", id}, {"object", "chat.completion.chunk"}, {"created", created}, {"model", kModel},
                {"choices", Json::array()}, {"mindnano_metrics", result.metrics}};
            if (request.include_usage) final["usage"] = result.usage;
            Send(fd, "data: " + final.dump() + "\n\ndata: [DONE]\n\n");
        } else Response(fd, 200, {{"id", id}, {"object", "chat.completion"}, {"created", created}, {"model", kModel},
            {"choices", Json::array({{{"index", 0}, {"message", {{"role", "assistant"}, {"content", result.text}}},
                                     {"finish_reason", result.reason}, {"logprobs", nullptr}}})},
            {"usage", result.usage}, {"mindnano_metrics", result.metrics}});
    } catch (const Error & e) { error(e); }
      catch (const Json::exception & e) { error(Error(400, "invalid_json", e.what())); }
      catch (const std::exception & e) { error(Error(500, "inference_error", e.what())); }
}
void Console(Engine & engine) {
    Json messages = Json::array();
    bool reuse_generated = true;
    std::cout << "\n输入中文即可对话。/reset 清空历史，/reuse on|off 切换生成状态复用，/quit 退出。\n> " << std::flush;
    while (!quitting) {
        pollfd p {STDIN_FILENO, POLLIN, 0};
        if (poll(&p, 1, 100) <= 0) continue;
        std::string line;
        if (!std::getline(std::cin, line)) return;
        if (line == "/quit") { quitting = 1; engine.Cancel(); return; }
        if (line == "/reset") {
            try { engine.ClearCache(); messages = Json::array(); }
            catch (const std::exception & e) { std::cerr << "\nerror: " << e.what() << '\n'; }
        }
        else if (line == "/cancel") engine.Cancel();
        else if (line == "/reuse on" || line == "/reuse off") {
            reuse_generated = line == "/reuse on";
            std::cout << "生成状态复用：" << (reuse_generated ? "开启（增量数值路径）" : "关闭（严格前缀路径）") << '\n';
        }
        else if (!line.empty()) {
            auto next = messages;
            next.push_back({{"role", "user"}, {"content", line}});
            try {
                auto r = Parse({{"model", kModel}, {"messages", next}, {"reuse_generated_state", reuse_generated}});
                const auto result = engine.Run(r, Id(), [] { return true; }, [](auto text) {
                    std::cout << text << std::flush; return true;
                }, [] { return true; });
                if (result.reason != "canceled") {
                    next.push_back({{"role", "assistant"}, {"content", result.text}}); messages = next;
                }
                std::cout << "\n[TTFT " << result.metrics["ttft_ms"] << " ms; decode "
                    << result.metrics["decode_tokens_per_second"] << " tokens/s; " << result.reason << "]\n";
            } catch (const std::exception & e) { std::cerr << "\nerror: " << e.what() << '\n'; }
        }
        std::cout << "> " << std::flush;
    }
}
} // namespace

int main(int argc, char ** argv) {
    try {
        std::string model = "ling3-tiny-w4.l3r", host = "127.0.0.1", log = "metrics.jsonl";
        int port = 9091; bool check = false, console = isatty(STDIN_FILENO);
        std::string mla_backend = "auto";
        std::size_t context = 4096;
        std::size_t session_cache_mib = 256;
        for (int i = 1; i < argc; ++i) {
            const std::string arg = argv[i];
            if (arg == "--help") {
                std::cout << "mindnano-infer [--model FILE] [--host IPv4] [--port 9091] [--log metrics.jsonl]\n"
                    "               [--context 4K|8K|16K|32K|64K|128K|256K] [--list-contexts]\n"
                    "               [--no-console] [--check] [--mla-backend auto|cpu|npu]\n"
                    "               [--session-cache-mib 256] (0 disables historical snapshots, not resident KV)\n"
                    "RK3588 / tested RKNPU 0.9.8; qualification and memory estimates are advisory. Text-only OpenAI chat completions.\n"; return 0;
            }
            if (arg == "--check") { check = true; continue; }
            if (arg == "--allow-experimental-context") { continue; } // Compatibility: no longer needed.
            if (arg == "--list-contexts") {
                for (auto n : kContexts) std::cout << n / 1024 << "K: estimated_model_rss_gib="
                    << EstimateContextMiB(n) / 1024 << " suggested_available_memory_gib="
                    << (EstimateContextMiB(n) + 1536) / 1024
                    << (n > 131072 ? " experimental_extrapolation_beyond_native_128K" : "") << '\n';
                return 0;
            }
            if (arg == "--no-console") { console = false; continue; }
            if (i + 1 == argc) throw std::runtime_error("missing argument after " + arg);
            if (arg == "--model") model = argv[++i];
            else if (arg == "--context") context = ParseContext(argv[++i]);
            else if (arg == "--mla-backend") mla_backend = argv[++i];
            else if (arg == "--host") host = argv[++i];
            else if (arg == "--log") log = argv[++i];
            else if (arg == "--session-cache-mib") {
                const std::string value = argv[++i];
                const auto [p, ec] = std::from_chars(value.data(), value.data()+value.size(), session_cache_mib);
                if (ec != std::errc{} || p != value.data()+value.size() ||
                    session_cache_mib > std::numeric_limits<std::size_t>::max()/(1024*1024))
                    throw std::invalid_argument("session cache budget must be a non-negative MiB integer");
            }
            else if (arg == "--port") {
                std::string value = argv[++i]; auto [p, ec] = std::from_chars(value.data(), value.data() + value.size(), port);
                if (ec != std::errc{} || p != value.data() + value.size() || port < 1 || port > 65535)
                    throw std::runtime_error("port must be 1..65535");
            } else throw std::runtime_error("unknown option " + arg);
        }
        std::signal(SIGPIPE, SIG_IGN); std::signal(SIGINT, OnSignal); std::signal(SIGTERM, OnSignal);
        if (mla_backend != "auto" && mla_backend != "cpu" && mla_backend != "npu")
            throw std::invalid_argument("--mla-backend must be auto, cpu or npu");
        setenv("LING3_MLA_BACKEND", mla_backend.c_str(), 1);
        std::cerr << "mla_backend=" << mla_backend << " tile=512 shared_by_six_layers=1\n";
        const auto device = CheckDevice();
        if (check) return 0;
        if (context > 131072)
            std::cerr << "context_warning: 256K exceeds native 128K; continuing with unvalidated position extrapolation\n";
        const auto required = EstimateContextMiB(context) + 1536;
        const auto available = MemoryMiB("MemAvailable:");
        const Json budget {{"context_length", context}, {"policy", "warn_only"},
            {"estimate_basis", "legacy_before_memory_optimizations"},
            {"estimated_model_rss_mib", EstimateContextMiB(context)},
            {"reserve_mib", 1536}, {"suggested_available_mib", required},
            {"available_memory_mib", available},
            {"below_estimate", available < required}};
        std::cerr << "memory_budget=" << budget.dump() << std::endl;
        if (available < required)
            std::cerr << "memory_warning: available RAM may be insufficient; legacy estimate including reserve is "
                      << required << " MiB, available " << available
                      << " MiB; continuing without a memory admission limit\n";
        if (context > 4096) std::cerr << "notice: contexts above 4K have not passed long-context quality/performance qualification\n";
        // Bind before expensive initialization; conflicting ports fail immediately.
        const int server = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
        if (server < 0) throw std::runtime_error("socket failed");
        struct CloseServer { int fd; ~CloseServer() { close(fd); } } close_server {server};
        int reuse = 1; setsockopt(server, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
        sockaddr_in addr {}; addr.sin_family = AF_INET; addr.sin_port = htons(port);
        if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) throw std::runtime_error("host must be IPv4");
        if (bind(server, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) != 0)
            throw std::runtime_error("bind failed: " + std::string(std::strerror(errno)));
        const auto start = Clock::now();
        std::cout << "initializing_model=" << model << std::endl;
        Engine engine(model, log, context);
        engine.SetCacheBudget(session_cache_mib*1024*1024);
        engine.initialization_ms = Ms(start);
        std::cout << "weight_format=" << engine.WeightFormat().dump() << std::endl;
        engine.Warmup();
        std::cout << "weight_memory=" << engine.WeightMemory().dump() << std::endl;
        if (listen(server, 16) != 0) throw std::runtime_error("listen failed");
        std::cout << "service_ready=http://" << host << ':' << port << "/v1\n"
                  << "initialization_ms=" << engine.initialization_ms << " warmup_ms=" << engine.warmup_ms
                  << " context_length=" << engine.context() << " metrics_log=" << log << std::endl;
        std::mutex mutex; std::condition_variable cv; std::deque<int> queue;
        bool stop_workers = false;
        std::vector<std::thread> workers;
        for (int i = 0; i < 4; ++i) workers.emplace_back([&] {
            while (true) {
                int fd;
                { std::unique_lock lock(mutex); cv.wait(lock, [&] { return stop_workers || !queue.empty(); });
                  if (queue.empty()) return; fd = queue.front(); queue.pop_front(); }
                Handle(fd, engine, device);
            }
        });
        std::thread terminal;
        if (console) terminal = std::thread([&] { Console(engine); });
        while (!quitting) {
            pollfd p {server, POLLIN, 0};
            if (poll(&p, 1, 100) <= 0) continue;
            const int fd = accept4(server, nullptr, nullptr, SOCK_CLOEXEC);
            if (fd < 0) continue;
            std::lock_guard lock(mutex);
            if (queue.size() >= 16) { close(fd); continue; }
            queue.push_back(fd); cv.notify_one();
        }
        engine.Cancel();
        { std::lock_guard lock(mutex); stop_workers = true; for (int fd : queue) close(fd); queue.clear(); }
        cv.notify_all();
        for (auto & worker : workers) worker.join();
        if (terminal.joinable()) terminal.join();
        return 0;
    } catch (const std::exception & e) {
        std::cerr << "startup_error: " << e.what() << std::endl; return 1;
    }
}