From dd2738d952df8f0e6f8e6fe4ca4f16224a4b7eef Mon Sep 17 00:00:00 2001 From: Graffioh <93008765+Graffioh@users.noreply.github.com> Date: Wed, 26 Aug 2026 14:44:39 +0000 Subject: [PATCH] experimental: integrate luce_odistill bridge --- server/CMakeLists.txt | 1 + server/src/server/http_server.cpp | 111 ++++++ server/src/server/http_server.h | 13 + server/src/server/luce_odistill_bridge.cpp | 392 +++++++++++++++++++++ server/src/server/luce_odistill_bridge.h | 94 +++++ server/src/server/scheduler.cpp | 10 + server/src/server/server_main.cpp | 60 ++++ server/test/test_server_unit.cpp | 89 +++++ 8 files changed, 770 insertions(+) create mode 100644 server/src/server/luce_odistill_bridge.cpp create mode 100644 server/src/server/luce_odistill_bridge.h diff --git a/server/CMakeLists.txt b/server/CMakeLists.txt index 925d5839d..1658e066f 100644 --- a/server/CMakeLists.txt +++ b/server/CMakeLists.txt @@ -516,6 +516,7 @@ add_library(dflash_common STATIC src/server/reasoning.cpp src/server/tool_memory.cpp src/server/sse_emitter.cpp + src/server/luce_odistill_bridge.cpp src/server/prefix_cache.cpp src/server/pin_friendly_prompt.cpp src/server/disk_prefix_cache.cpp diff --git a/server/src/server/http_server.cpp b/server/src/server/http_server.cpp index a909344e8..ef333adb9 100644 --- a/server/src/server/http_server.cpp +++ b/server/src/server/http_server.cpp @@ -794,6 +794,26 @@ json build_props_body(const ServerConfig & config, // benchmarks to silently run at temp=0 (degenerate-decode collapse) // when the model card specifies temp=1.0/top_p=0.95/top_k=64. const auto & smp = config.sampler_defaults; + const auto & selection = config.luce_odistill_selection; + const json odistill = { + {"enabled", selection.configured}, + {"capture_enabled", !config.luce_odistill_capture_socket.empty()}, + {"status", selection.configured ? json("experimental") : json(nullptr)}, + {"selection_state", selection.configured + ? json(selection.selection_state) : json(nullptr)}, + {"release_id", selection.release_id.empty() + ? json(nullptr) : json(selection.release_id)}, + {"source_artifact_sha256", selection.source_artifact_sha256.empty() + ? json(nullptr) : json(selection.source_artifact_sha256)}, + {"runtime_artifact_sha256", selection.runtime_artifact_sha256.empty() + ? json(nullptr) : json(selection.runtime_artifact_sha256)}, + {"target_profile_sha256", selection.target_profile_sha256.empty() + ? json(nullptr) : json(selection.target_profile_sha256)}, + {"drafter_profile_sha256", selection.drafter_profile_sha256.empty() + ? json(nullptr) : json(selection.drafter_profile_sha256)}, + {"evaluation_report_sha256", selection.evaluation_report_sha256.empty() + ? json(nullptr) : json(selection.evaluation_report_sha256)}, + }; json body = { {"default_generation_settings", { {"n_ctx", config.max_ctx}, @@ -804,6 +824,7 @@ json build_props_body(const ServerConfig & config, {"repeat_penalty", smp.has_repetition_penalty ? smp.repetition_penalty : 1.0f}, }}, {"model_alias", config.model_name}, + {"experimental", {{"luce_odistill", odistill}}}, {"model_path", config.model_path}, {"build_info", std::string(kServerName) + " v" DFLASH_SERVER_VERSION " props_schema=" + std::to_string(kPropsSchema)}, @@ -1141,6 +1162,10 @@ HttpServer::HttpServer(ModelBackend & backend, } disk_cache_.init(); status_html_path_ = resolve_status_html(); + if (!config_.luce_odistill_capture_socket.empty()) { + luce_odistill_trace_sink_ = std::make_unique( + config_.luce_odistill_capture_socket); + } // PPP env overrides (operator-facing; no CLI flags required). auto env_truthy = [](const char * v) -> bool { @@ -2131,6 +2156,7 @@ bool HttpServer::route_request(SocketHandle fd, const HttpRequest & hr) { hr.path.c_str(), hr.body.size()); ParsedRequest req; + req.luce_odistill_capture = hr.luce_odistill_capture; bool count_tokens_only = false; try { const json body = json::parse(hr.body); @@ -2574,6 +2600,65 @@ bool is_continuation_request(const json & messages) { } // namespace +void HttpServer::enqueue_luce_odistill_trace( + const ParsedRequest & req, const GenerateResult & result, + const GenTimings & timings, const SseEmitter & emitter, + int completion_tokens, double latency_seconds) { + if (!luce_odistill_trace_sink_ || + !config_.luce_odistill_selection.configured || + !req.luce_odistill_capture.granted || + req.model != config_.model_name || + !result.ok()) { + return; + } + + json request = req.raw_body; + request["model"] = req.model; + request["messages"] = req.messages; + if (!req.tools.is_null()) request["tools"] = req.tools; + if (!req.tool_choice.is_null()) request["tool_choice"] = req.tool_choice; + + json message = { + {"role", "assistant"}, + {"content", emitter.accumulated_text()}, + }; + if (!emitter.reasoning_text().empty()) { + message["reasoning_content"] = emitter.reasoning_text(); + } + if (!emitter.tool_calls().empty()) { + json tool_calls = json::array(); + for (const auto & tool_call : emitter.tool_calls()) { + tool_calls.push_back({ + {"id", tool_call.id}, + {"type", "function"}, + {"function", { + {"name", tool_call.name}, + {"arguments", tool_call.arguments}, + }}, + }); + } + message["tool_calls"] = std::move(tool_calls); + } + const json choice = { + {"message", message}, + {"finish_reason", emitter.finish_reason()}, + }; + const json response = { + {"id", req.response_id}, + {"model", req.model}, + {"choices", json::array({choice})}, + {"usage", { + {"completion_tokens", completion_tokens}, + {"timings", build_timings_json(timings, completion_tokens)}, + {"accept_rate", result.accept_rate}, + {"spec_decode_ran", result.spec_decode_ran}, + }}, + }; + luce_odistill_trace_sink_->enqueue(build_luce_odistill_native_event( + config_.model_name, request, response, req.luce_odistill_capture, + config_.luce_odistill_selection, latency_seconds)); +} + void HttpServer::apply_flowkv_compression( const ParsedRequest & req, PreparedPrompt & prepared) { int hot_window = 2; @@ -4200,6 +4285,10 @@ void HttpServer::process_job(ServerJob * job) { const std::string finish = client_disconnected ? "client_disconnect" : (result.ok() ? emitter.finish_reason() : "error"); + if (!client_disconnected) { + enqueue_luce_odistill_trace( + req, result, gen_timings, emitter, completion_tokens, elapsed_s); + } std::fprintf(stderr, "[server] chat DONE %s ok=%s in=%zu effective_in=%zu out=%d " @@ -4363,6 +4452,28 @@ bool HttpServer::read_http_request(SocketHandle fd, HttpRequest & out) { } out.query = std::move(query_string); + // Preserve only the four internal opt-in headers. They never enter the + // prompt or access logs; duplicate values fail closed in the bridge parser. + std::vector> request_headers; + size_t header_cursor = line_end + 1; + while (header_cursor < (size_t)hend) { + size_t next = buf.find("\n", header_cursor); + if (next == std::string::npos || next > (size_t)hend) break; + std::string header_line = buf.substr(header_cursor, next - header_cursor); + if (!header_line.empty() && header_line.back() == '\r') { + header_line.pop_back(); + } + if (header_line.empty()) break; + const size_t colon = header_line.find(':'); + if (colon != std::string::npos) { + request_headers.emplace_back( + header_line.substr(0, colon), header_line.substr(colon + 1)); + } + header_cursor = next + 1; + } + out.luce_odistill_capture = + parse_luce_odistill_capture_headers(request_headers); + // Find Content-Length. long content_length = 0; { diff --git a/server/src/server/http_server.h b/server/src/server/http_server.h index 52c36473b..5b88a84aa 100644 --- a/server/src/server/http_server.h +++ b/server/src/server/http_server.h @@ -30,6 +30,7 @@ #include "adaptive_keep_ratio.h" #include "server_status.h" #include "sse_emitter.h" +#include "luce_odistill_bridge.h" #include #include @@ -235,6 +236,11 @@ struct ServerConfig { // Routing data collection (--collect-routing ): write binary per-token // routing data (hidden states + expert selections) for predictor training. std::string collect_routing_path; + + // Experimental local luce_odistill bridge. The selection is verified once + // before backend construction; trace emission is optional and best-effort. + LuceODistillSelection luce_odistill_selection; + std::string luce_odistill_capture_socket; }; namespace http_detail { @@ -310,6 +316,7 @@ struct ParsedRequest { DiskPrefixCachePolicy disk_cache_policy; // PPP: stable pin cut for tool-heavy requests (0 = use default boundary). int pin_end_token = 0; + LuceODistillConsent luce_odistill_capture; }; // Parse request sampler fields, applying model-card defaults where present. @@ -459,6 +466,10 @@ class HttpServer { void configure_generation_io( ServerJob * job, const ParsedRequest & req, SseEmitter & emitter, GenerationOutputState & output, DaemonIO & io); + void enqueue_luce_odistill_trace( + const ParsedRequest & req, const GenerateResult & result, + const GenTimings & timings, const SseEmitter & emitter, + int completion_tokens, double latency_seconds); // Worker thread, concurrent mode (the backend exposes a SeqEngine): // iteration-level scheduler. Admission is claim-only; this baseline @@ -496,6 +507,7 @@ class HttpServer { std::string path; std::string query; // raw query string (after '?') std::string body; + LuceODistillConsent luce_odistill_capture; }; bool read_http_request(SocketHandle fd, HttpRequest & out); @@ -549,6 +561,7 @@ class HttpServer { ToolMemory tool_memory_; PrefixCache prefix_cache_; DiskPrefixCache disk_cache_; + std::unique_ptr luce_odistill_trace_sink_; // Per-session adaptive keep_ratio bandit state. HttpServerSessions sessions_; diff --git a/server/src/server/luce_odistill_bridge.cpp b/server/src/server/luce_odistill_bridge.cpp new file mode 100644 index 000000000..adc9f77d9 --- /dev/null +++ b/server/src/server/luce_odistill_bridge.cpp @@ -0,0 +1,392 @@ +#include "luce_odistill_bridge.h" + +#include "common/gguf_inspect.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +#include +#include +#include +#include +#include +#endif + +namespace dflash::common { + +namespace { + +constexpr size_t kMaxManifestBytes = 1024 * 1024; +constexpr size_t kMaxEventBytes = 32 * 1024 * 1024; +constexpr size_t kMaxQueuedBytes = 64 * 1024 * 1024; +constexpr size_t kMaxQueuedEvents = 64; + +bool is_sha256(const nlohmann::json & value) { + if (!value.is_string()) return false; + const std::string text = value.get(); + return text.size() == 64 && std::all_of(text.begin(), text.end(), [](char c) { + return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f'); + }); +} + +bool has_exact_keys(const nlohmann::json & value, + std::initializer_list keys) { + if (!value.is_object() || value.size() != keys.size()) return false; + for (const char * key : keys) { + if (!value.contains(key)) return false; + } + return true; +} + +std::string lower_ascii(std::string value) { + std::transform(value.begin(), value.end(), value.begin(), [](unsigned char c) { + return (char)std::tolower(c); + }); + return value; +} + +std::string trim_ascii(std::string value) { + auto is_space = [](unsigned char c) { return std::isspace(c) != 0; }; + while (!value.empty() && is_space((unsigned char)value.front())) { + value.erase(value.begin()); + } + while (!value.empty() && is_space((unsigned char)value.back())) { + value.pop_back(); + } + return value; +} + +bool truthy(const std::string & value) { + const std::string normalized = lower_ascii(trim_ascii(value)); + return normalized == "1" || normalized == "true" || + normalized == "yes" || normalized == "on"; +} + +} // namespace + +bool parse_luce_odistill_selection(const nlohmann::json & value, + LuceODistillSelection & out, + std::string & error) { + out = LuceODistillSelection{}; + if (!has_exact_keys(value, { + "schema_version", "kind", "experimental", "selection_state", + "release_id", "source_artifact_sha256", "runtime_artifact", + "target_profile_sha256", "drafter_profile_sha256", + "evaluation_report_sha256"})) { + error = "selection must contain the exact versioned fields"; + return false; + } + if (value["schema_version"] != 1 || + value["kind"] != "luce_odistill_lucebox_selection" || + value["experimental"] != true) { + error = "selection schema is not the experimental Lucebox contract"; + return false; + } + if (!value["selection_state"].is_string()) { + error = "selection_state must be a string"; + return false; + } + const std::string state = value["selection_state"].get(); + if (state == "baseline") { + if (!value["release_id"].is_null() || + !value["evaluation_report_sha256"].is_null()) { + error = "baseline selection cannot name a release or report"; + return false; + } + } else if (state == "promoted") { + if (!value["release_id"].is_string() || + value["release_id"].get().rfind("release-", 0) != 0 || + !is_sha256(value["evaluation_report_sha256"])) { + error = "promoted selection requires a release and report digest"; + return false; + } + } else { + error = "selection_state must be baseline or promoted"; + return false; + } + for (const char * key : { + "source_artifact_sha256", "target_profile_sha256", + "drafter_profile_sha256"}) { + if (!is_sha256(value[key])) { + error = std::string(key) + " must be a lowercase SHA-256 digest"; + return false; + } + } + const auto & runtime = value["runtime_artifact"]; + if (!has_exact_keys(runtime, {"format", "path", "sha256"}) || + runtime["format"] != "gguf" || !runtime["path"].is_string() || + !is_sha256(runtime["sha256"])) { + error = "runtime_artifact must bind one absolute GGUF path and digest"; + return false; + } + const std::filesystem::path runtime_path(runtime["path"].get()); + if (!runtime_path.is_absolute()) { + error = "runtime_artifact.path must be absolute"; + return false; + } + + out.configured = true; + out.selection_state = state; + if (value["release_id"].is_string()) { + out.release_id = value["release_id"].get(); + } + out.source_artifact_sha256 = value["source_artifact_sha256"].get(); + out.runtime_artifact_path = runtime_path.string(); + out.runtime_artifact_sha256 = runtime["sha256"].get(); + out.target_profile_sha256 = value["target_profile_sha256"].get(); + out.drafter_profile_sha256 = value["drafter_profile_sha256"].get(); + if (value["evaluation_report_sha256"].is_string()) { + out.evaluation_report_sha256 = + value["evaluation_report_sha256"].get(); + } + return true; +} + +bool load_luce_odistill_selection(const std::string & manifest_path, + LuceODistillSelection & out, + std::string & error) { +#if !defined(_WIN32) + struct stat metadata{}; + if (lstat(manifest_path.c_str(), &metadata) != 0 || + !S_ISREG(metadata.st_mode) || S_ISLNK(metadata.st_mode) || + (metadata.st_mode & (S_IWGRP | S_IWOTH)) != 0) { + error = "selection manifest must be a real non-shared-writable regular file"; + return false; + } +#endif + std::ifstream stream(manifest_path, std::ios::binary); + if (!stream) { + error = "cannot open selection manifest"; + return false; + } + std::string raw((std::istreambuf_iterator(stream)), + std::istreambuf_iterator()); + if (raw.empty() || raw.size() > kMaxManifestBytes) { + error = "selection manifest size is invalid"; + return false; + } + nlohmann::json value; + try { + value = nlohmann::json::parse(raw); + } catch (const std::exception &) { + error = "selection manifest is not valid JSON"; + return false; + } + LuceODistillSelection parsed; + if (!parse_luce_odistill_selection(value, parsed, error)) return false; + +#if !defined(_WIN32) + struct stat runtime_metadata{}; + if (lstat(parsed.runtime_artifact_path.c_str(), &runtime_metadata) != 0 || + !S_ISREG(runtime_metadata.st_mode) || S_ISLNK(runtime_metadata.st_mode) || + (runtime_metadata.st_mode & (S_IWGRP | S_IWOTH)) != 0) { + error = "selected runtime artifact must be a real non-shared-writable regular file"; + return false; + } +#endif + const GgufMetadata runtime = + read_gguf_metadata(parsed.runtime_artifact_path, /*compute_sha256=*/true); + if (!runtime.ok || runtime.sha256.empty()) { + error = "selected runtime artifact is not a readable GGUF"; + return false; + } + if (runtime.sha256 != parsed.runtime_artifact_sha256) { + error = "selected runtime artifact SHA-256 does not match the manifest"; + return false; + } + out = std::move(parsed); + return true; +} + +nlohmann::json luce_odistill_runtime_drafter_json( + const LuceODistillSelection & selection) { + return { + {"selection_state", selection.selection_state}, + {"release_id", selection.release_id.empty() + ? nlohmann::json(nullptr) : nlohmann::json(selection.release_id)}, + {"source_artifact_sha256", selection.source_artifact_sha256}, + {"runtime_artifact_sha256", selection.runtime_artifact_sha256}, + {"evaluation_report_sha256", selection.evaluation_report_sha256.empty() + ? nlohmann::json(nullptr) + : nlohmann::json(selection.evaluation_report_sha256)}, + {"target_profile_sha256", selection.target_profile_sha256}, + {"drafter_profile_sha256", selection.drafter_profile_sha256}, + }; +} + +LuceODistillConsent parse_luce_odistill_capture_headers( + const std::vector> & headers) { + LuceODistillConsent result; + std::unordered_map selected; + std::unordered_set duplicates; + for (const auto & [raw_name, raw_value] : headers) { + const std::string name = lower_ascii(trim_ascii(raw_name)); + if (name != "x-luce-odistill-consent" && + name != "x-luce-odistill-subject" && + name != "x-luce-odistill-conversation" && + name != "x-luce-odistill-profile") { + continue; + } + if (!selected.emplace(name, trim_ascii(raw_value)).second) { + duplicates.insert(name); + } + } + const auto consent = selected.find("x-luce-odistill-consent"); + result.requested = consent != selected.end() && truthy(consent->second); + if (!result.requested || !duplicates.empty()) return result; + const auto subject = selected.find("x-luce-odistill-subject"); + if (subject == selected.end() || subject->second.empty() || + subject->second.size() > 512) { + return result; + } + result.subject = subject->second; + const auto conversation = selected.find("x-luce-odistill-conversation"); + if (conversation != selected.end()) { + if (conversation->second.empty() || conversation->second.size() > 512) { + return result; + } + result.conversation = conversation->second; + } + const auto profile = selected.find("x-luce-odistill-profile"); + if (profile != selected.end()) { + if (profile->second.empty() || profile->second.size() > 128) return result; + result.profile = profile->second; + } + result.granted = true; + return result; +} + +nlohmann::json build_luce_odistill_native_event( + const std::string & served_model, + const nlohmann::json & request, + const nlohmann::json & response, + const LuceODistillConsent & consent, + const LuceODistillSelection & selection, + double latency_seconds) { + return { + {"schema_version", 1}, + {"kind", "luce_odistill_native_capture"}, + {"served_model", served_model}, + {"request", request}, + {"response", response}, + {"consent", { + {"granted", consent.granted}, + {"subject", consent.subject}, + {"conversation", consent.conversation.empty() + ? nlohmann::json(nullptr) : nlohmann::json(consent.conversation)}, + {"profile", consent.profile}, + {"source", "x-luce-odistill-consent"}, + }}, + {"runtime_drafter", luce_odistill_runtime_drafter_json(selection)}, + {"observation", {{"latency_seconds", latency_seconds}}}, + }; +} + +LuceODistillTraceSink::LuceODistillTraceSink(std::string socket_path) + : socket_path_(std::move(socket_path)) { + if (!socket_path_.empty()) { + worker_ = std::thread(&LuceODistillTraceSink::worker_loop, this); + } +} + +LuceODistillTraceSink::~LuceODistillTraceSink() { + { + std::lock_guard lock(mutex_); + stopping_ = true; + queue_.clear(); + queued_bytes_ = 0; + } + cv_.notify_all(); + if (worker_.joinable()) worker_.join(); +} + +bool LuceODistillTraceSink::enqueue(const nlohmann::json & event) { + if (socket_path_.empty()) return false; + std::string payload; + try { + payload = event.dump(); + } catch (const std::exception &) { + return false; + } + if (payload.empty() || payload.size() > kMaxEventBytes) return false; + { + std::lock_guard lock(mutex_); + if (stopping_ || queue_.size() >= kMaxQueuedEvents || + queued_bytes_ + payload.size() > kMaxQueuedBytes) { + return false; + } + queued_bytes_ += payload.size(); + queue_.push_back(std::move(payload)); + } + cv_.notify_one(); + return true; +} + +void LuceODistillTraceSink::worker_loop() { + while (true) { + std::string payload; + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [&] { return stopping_ || !queue_.empty(); }); + if (stopping_) return; + payload = std::move(queue_.front()); + queue_.pop_front(); + queued_bytes_ -= payload.size(); + } + (void)send_payload(payload); + } +} + +bool LuceODistillTraceSink::send_payload(const std::string & payload) const { +#if defined(_WIN32) + (void)payload; + return false; +#else + sockaddr_un address{}; + if (socket_path_.size() >= sizeof(address.sun_path) || + payload.size() > std::numeric_limits::max()) { + return false; + } + const int fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (fd < 0) return false; + timeval timeout{}; + timeout.tv_sec = 1; + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)); + address.sun_family = AF_UNIX; + std::memcpy(address.sun_path, socket_path_.c_str(), socket_path_.size() + 1); + if (connect(fd, reinterpret_cast(&address), sizeof(address)) != 0) { + close(fd); + return false; + } + const uint32_t size = (uint32_t)payload.size(); + const unsigned char prefix[4] = { + (unsigned char)((size >> 24) & 0xff), + (unsigned char)((size >> 16) & 0xff), + (unsigned char)((size >> 8) & 0xff), + (unsigned char)(size & 0xff), + }; + auto send_all = [&](const void * raw, size_t length) { + const char * data = static_cast(raw); + size_t sent = 0; + while (sent < length) { + const ssize_t count = send(fd, data + sent, length - sent, MSG_NOSIGNAL); + if (count <= 0) return false; + sent += (size_t)count; + } + return true; + }; + const bool ok = send_all(prefix, sizeof(prefix)) && + send_all(payload.data(), payload.size()); + close(fd); + return ok; +#endif +} + +} // namespace dflash::common diff --git a/server/src/server/luce_odistill_bridge.h b/server/src/server/luce_odistill_bridge.h new file mode 100644 index 000000000..afe64e868 --- /dev/null +++ b/server/src/server/luce_odistill_bridge.h @@ -0,0 +1,94 @@ +// Experimental local bridge between the Lucebox data plane and the +// luce_odistill controller. Training and promotion policy stay outside the +// engine; this module only verifies one startup selection and emits consented +// trace events over a local Unix socket. + +#pragma once + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace dflash::common { + +struct LuceODistillSelection { + bool configured = false; + std::string selection_state; + std::string release_id; + std::string source_artifact_sha256; + std::string runtime_artifact_path; + std::string runtime_artifact_sha256; + std::string target_profile_sha256; + std::string drafter_profile_sha256; + std::string evaluation_report_sha256; +}; + +// Pure schema parser used by startup and model-free tests. It does not touch +// the runtime artifact; load_luce_odistill_selection additionally opens the +// GGUF and verifies its exact SHA-256 before returning. +bool parse_luce_odistill_selection(const nlohmann::json & value, + LuceODistillSelection & out, + std::string & error); +bool load_luce_odistill_selection(const std::string & manifest_path, + LuceODistillSelection & out, + std::string & error); + +nlohmann::json luce_odistill_runtime_drafter_json( + const LuceODistillSelection & selection); + +struct LuceODistillConsent { + bool requested = false; + bool granted = false; + std::string subject; + std::string conversation; + std::string profile = "default"; +}; + +// Header names are case-insensitive. Duplicate internal headers, an invalid +// truth value, an absent subject, or overlong identity values fail closed for +// capture while leaving the served request untouched. +LuceODistillConsent parse_luce_odistill_capture_headers( + const std::vector> & headers); + +nlohmann::json build_luce_odistill_native_event( + const std::string & served_model, + const nlohmann::json & request, + const nlohmann::json & response, + const LuceODistillConsent & consent, + const LuceODistillSelection & selection, + double latency_seconds); + +class LuceODistillTraceSink { +public: + explicit LuceODistillTraceSink(std::string socket_path); + ~LuceODistillTraceSink(); + + LuceODistillTraceSink(const LuceODistillTraceSink &) = delete; + LuceODistillTraceSink & operator=(const LuceODistillTraceSink &) = delete; + + // Best effort and bounded: a missing/stalled collector drops the event and + // never delays the inference worker or its client response. + bool enqueue(const nlohmann::json & event); + +private: + void worker_loop(); + bool send_payload(const std::string & payload) const; + + std::string socket_path_; + std::mutex mutex_; + std::condition_variable cv_; + std::deque queue_; + size_t queued_bytes_ = 0; + bool stopping_ = false; + std::thread worker_; +}; + +} // namespace dflash::common diff --git a/server/src/server/scheduler.cpp b/server/src/server/scheduler.cpp index e22a4cf3f..3a11a2fed 100644 --- a/server/src/server/scheduler.cpp +++ b/server/src/server/scheduler.cpp @@ -301,6 +301,16 @@ void HttpServer::scheduler_loop(SeqEngine & engine) { const double elapsed_s = std::chrono::duration( std::chrono::steady_clock::now() - s.started_at).count(); const int out_tokens = (int)s.gen_tokens.size(); + if (backend_ok && !s.failed && !s.client_disconnected) { + GenerateResult trace_result; + trace_result.succeed(); + trace_result.tokens = s.gen_tokens; + trace_result.prefill_s = s.prefill_s; + trace_result.decode_s = decode_s; + enqueue_luce_odistill_trace( + req, trace_result, gen_timings, *s.emitter, + s.completion_tokens, elapsed_s); + } std::fprintf(stderr, "[server] chat DONE %s ok=%s in=%zu out=%d %.1fs %.1f tok/s " "finish=%s slot=%d prefill=%.1fs decode=%.1fs(%.1ftok/s) parallel\n", diff --git a/server/src/server/server_main.cpp b/server/src/server/server_main.cpp index efae13bbe..313a9f59b 100644 --- a/server/src/server/server_main.cpp +++ b/server/src/server/server_main.cpp @@ -35,6 +35,7 @@ #include #include #include +#include #include #include #include @@ -75,6 +76,10 @@ static void print_usage(const char * prog) { "\n" "Options:\n" " --draft Draft model for speculative decode\n" + " --luce-odistill-selection \n" + " Experimental verified drafter selection; conflicts with --draft\n" + " --luce-odistill-capture-socket \n" + " Experimental local consented trace sink\n" " --port Listen port (default: 8080)\n" " --host Bind address (default: 0.0.0.0)\n" " --max-ctx Max context length (default: 131072)\n" @@ -253,6 +258,8 @@ int main(int argc, char ** argv) { bool ddtree_tau_set = false; bool specla_top_k_set = false; int specla_top_k = 4; + std::string luce_odistill_selection_path; + std::string luce_odistill_selected_draft_path; // Track which thinking-budget tunables the operator set via CLI. // Those values win over the model card (spec §3.1: "Explicit CLI @@ -280,6 +287,12 @@ int main(int argc, char ** argv) { for (int i = 2; i < argc; i++) { if (std::strcmp(argv[i], "--draft") == 0 && i + 1 < argc) { bargs.draft_path = argv[++i]; + } else if (std::strcmp(argv[i], "--luce-odistill-selection") == 0 && + i + 1 < argc) { + luce_odistill_selection_path = argv[++i]; + } else if (std::strcmp(argv[i], "--luce-odistill-capture-socket") == 0 && + i + 1 < argc) { + sconfig.luce_odistill_capture_socket = argv[++i]; } else if (std::strcmp(argv[i], "--port") == 0 && i + 1 < argc) { sconfig.port = std::atoi(argv[++i]); } else if (std::strcmp(argv[i], "--host") == 0 && i + 1 < argc) { @@ -713,6 +726,39 @@ int main(int argc, char ** argv) { set_environment_variable("DFLASH_SPLIT_FAST_ROLLBACK", "1", true); } + if (!luce_odistill_selection_path.empty()) { + if (bargs.draft_path) { + std::fprintf(stderr, + "[server] --luce-odistill-selection conflicts with --draft\n"); + return 2; + } + std::string selection_error; + if (!load_luce_odistill_selection( + luce_odistill_selection_path, + sconfig.luce_odistill_selection, selection_error)) { + std::fprintf(stderr, + "[server] invalid luce_odistill selection: %s\n", + selection_error.c_str()); + return 2; + } + luce_odistill_selected_draft_path = + sconfig.luce_odistill_selection.runtime_artifact_path; + bargs.draft_path = luce_odistill_selected_draft_path.c_str(); + } + if (!sconfig.luce_odistill_capture_socket.empty() && + !sconfig.luce_odistill_selection.configured) { + std::fprintf(stderr, + "[server] --luce-odistill-capture-socket requires " + "--luce-odistill-selection for exact runtime provenance\n"); + return 2; + } + if (!sconfig.luce_odistill_capture_socket.empty() && + !std::filesystem::path(sconfig.luce_odistill_capture_socket).is_absolute()) { + std::fprintf(stderr, + "[server] --luce-odistill-capture-socket must be an absolute path\n"); + return 2; + } + // Resolve documented environment defaults before factory preparation so // compatibility warnings describe the effective backend configuration. // An explicit --draft-swa value continues to take precedence. @@ -1151,6 +1197,20 @@ int main(int argc, char ** argv) { std::fprintf(stderr, "[server] │ port = %d\n", sconfig.port); std::fprintf(stderr, "[server] │ model = %s\n", bargs.model_path); std::fprintf(stderr, "[server] │ draft = %s\n", bargs.draft_path ? bargs.draft_path : "(none)"); + std::fprintf(stderr, "[server] │ luce_odistill = %s\n", + sconfig.luce_odistill_selection.configured + ? "experimental" : "off"); + if (sconfig.luce_odistill_selection.configured) { + const auto & selection = sconfig.luce_odistill_selection; + std::fprintf(stderr, "[server] │ odistill_state = %s\n", + selection.selection_state.c_str()); + std::fprintf(stderr, "[server] │ odistill_release= %s\n", + selection.release_id.empty() + ? "(baseline)" : selection.release_id.c_str()); + std::fprintf(stderr, "[server] │ odistill_capture= %s\n", + sconfig.luce_odistill_capture_socket.empty() + ? "off" : "local Unix socket"); + } std::fprintf(stderr, "[server] │ model_name = %s\n", sconfig.model_name.c_str()); std::fprintf(stderr, "[server] │ max_ctx = %d\n", sconfig.max_ctx); // max_tokens default for requests that omit the field. The request diff --git a/server/test/test_server_unit.cpp b/server/test/test_server_unit.cpp index 01dad8dcc..62f1b41bb 100644 --- a/server/test/test_server_unit.cpp +++ b/server/test/test_server_unit.cpp @@ -6907,3 +6907,92 @@ TEST_CASE(ServerUnitFixture, test_emitter_suppresses_malformed_multiline_tool_bu TEST_ASSERT(captured.find("[server] tool_call parse failed; suppressing buffered tool text") != std::string::npos); TEST_ASSERT(captured.find("text='\\n \\n malformed prose body with\\nnew lines and \\t tabs\\n'") != std::string::npos); } + +TEST_CASE(ServerUnitFixture, test_luce_odistill_selection_schema_is_exact) { + json selection = { + {"schema_version", 1}, + {"kind", "luce_odistill_lucebox_selection"}, + {"experimental", true}, + {"selection_state", "baseline"}, + {"release_id", nullptr}, + {"source_artifact_sha256", std::string(64, 'a')}, + {"runtime_artifact", { + {"format", "gguf"}, + {"path", "/models/stock.gguf"}, + {"sha256", std::string(64, 'b')}, + }}, + {"target_profile_sha256", std::string(64, 'c')}, + {"drafter_profile_sha256", std::string(64, 'd')}, + {"evaluation_report_sha256", nullptr}, + }; + LuceODistillSelection parsed; + std::string error; + TEST_ASSERT(parse_luce_odistill_selection(selection, parsed, error)); + TEST_ASSERT(parsed.configured); + TEST_ASSERT(parsed.selection_state == "baseline"); + TEST_ASSERT(parsed.runtime_artifact_path == "/models/stock.gguf"); + + selection["unexpected"] = true; + TEST_ASSERT(!parse_luce_odistill_selection(selection, parsed, error)); + selection.erase("unexpected"); + selection["runtime_artifact"]["path"] = "relative.gguf"; + TEST_ASSERT(!parse_luce_odistill_selection(selection, parsed, error)); +} + +TEST_CASE(ServerUnitFixture, test_luce_odistill_capture_headers_fail_closed) { + auto consent = parse_luce_odistill_capture_headers({ + {"X-Luce-ODistill-Consent", "true"}, + {"X-Luce-ODistill-Subject", "local-user-7"}, + {"X-Luce-ODistill-Conversation", "conversation-4"}, + {"X-Luce-ODistill-Profile", "coding"}, + }); + TEST_ASSERT(consent.requested); + TEST_ASSERT(consent.granted); + TEST_ASSERT(consent.subject == "local-user-7"); + TEST_ASSERT(consent.profile == "coding"); + + consent = parse_luce_odistill_capture_headers({ + {"X-Luce-ODistill-Consent", "true"}, + {"x-luce-odistill-consent", "true"}, + {"X-Luce-ODistill-Subject", "local-user-7"}, + }); + TEST_ASSERT(consent.requested); + TEST_ASSERT(!consent.granted); + + consent = parse_luce_odistill_capture_headers({ + {"X-Luce-ODistill-Consent", "true"}, + }); + TEST_ASSERT(consent.requested); + TEST_ASSERT(!consent.granted); +} + +TEST_CASE(ServerUnitFixture, test_luce_odistill_event_binds_runtime_identity) { + LuceODistillSelection selection; + selection.configured = true; + selection.selection_state = "promoted"; + selection.release_id = "release-20260826-test"; + selection.source_artifact_sha256 = std::string(64, 'a'); + selection.runtime_artifact_sha256 = std::string(64, 'b'); + selection.evaluation_report_sha256 = std::string(64, 'e'); + selection.target_profile_sha256 = std::string(64, 'c'); + selection.drafter_profile_sha256 = std::string(64, 'd'); + LuceODistillConsent consent; + consent.requested = true; + consent.granted = true; + consent.subject = "local-user-7"; + consent.profile = "default"; + + const json event = build_luce_odistill_native_event( + "Qwen3.8-27B", + {{"model", "Qwen3.8-27B"}, {"messages", json::array()}}, + {{"model", "Qwen3.8-27B"}, {"choices", json::array()}}, + consent, selection, 0.25); + TEST_ASSERT(event["kind"] == "luce_odistill_native_capture"); + TEST_ASSERT(event["consent"]["granted"] == true); + TEST_ASSERT(event["runtime_drafter"]["release_id"] == + "release-20260826-test"); + TEST_ASSERT(event["runtime_drafter"]["runtime_artifact_sha256"] == + std::string(64, 'b')); + TEST_ASSERT(event["runtime_drafter"]["evaluation_report_sha256"] == + std::string(64, 'e')); +}