api_server.cpp

// Ported from: vllm/entrypoints/openai/api_server.py @ e24d1b24 + the
// per-endpoint api_router modules. See api_server.h for scope + the cpp-httplib
// dependency deviation.
#include "vllm/entrypoints/openai/api_server.h"

#include <exception>
#include <fstream>
#include <iostream>
#include <iterator>
#include <limits>
#include <memory>
#include <mutex>
#include <thread>
#include <stdexcept>
#include <string>
#include <utility>

#include <httplib/httplib.h>
#include <nlohmann/json.hpp>

#include "vllm/entrypoints/openai/protocol.h"
#include "vllm/tokenizer/tokenizer.h"
#include "vllm/v1/engine/async_llm.h"
#include "vllm/v1/metrics/loggers.h"

namespace vllm::entrypoints::openai {

namespace {

size_t HttpWorkerCount(size_t max_concurrent_streams) {
  if (max_concurrent_streams == 0) {
    throw std::invalid_argument("max_concurrent_streams must be positive");
  }
  if (max_concurrent_streams >
      std::numeric_limits<size_t>::max() -
          ApiServer::kControlWorkerHeadroom) {
    throw std::invalid_argument("max_concurrent_streams is too large");
  }
  return max_concurrent_streams + ApiServer::kControlWorkerHeadroom;
}

size_t HttpWorkerCount(size_t max_concurrent_streams,
                       ApiServer::HttpWorkerPoolMode mode) {
  const size_t fixed_count = HttpWorkerCount(max_concurrent_streams);
  return mode == ApiServer::HttpWorkerPoolMode::kCapacityFixed ? fixed_count
                                                               : 0;
}

}  // namespace

// Opaque httplib::Server (pimpl — keeps httplib.h out of api_server.h).
struct ApiServer::Impl {
  Impl(size_t max_concurrent_streams, HttpWorkerPoolMode mode)
      : http_worker_count(HttpWorkerCount(max_concurrent_streams, mode)) {
    // cpp-httplib's default pool starts at hardware_concurrency()-1 and only
    // grows if idle_thread_count_ is exactly zero at enqueue. A burst can queue
    // accepted sockets while that counter is stale-positive; long-lived SSE
    // jobs then prevent the queued sockets from ever being read. A fixed floor
    // derived from the configured stream capacity removes that race and makes
    // resource use reproducible.
    if (http_worker_count != 0) {
      server.new_task_queue = [workers = http_worker_count]() {
        return new httplib::ThreadPool(workers);
      };
    }
    // Mirror vLLM's serving transport: vLLM serves through uvicorn over asyncio
    // (entrypoints/launcher.py:71,76), and asyncio disables Nagle on every
    // accepted TCP stream socket by default (CPython asyncio/base_events.py
    // _set_nodelay → setsockopt(IPPROTO_TCP, TCP_NODELAY, 1), invoked from
    // selector_events.py _SelectorSocketTransport). cpp-httplib defaults it off
    // (third_party/httplib/httplib.h:142) and applies it to the accepted socket
    // only when tcp_nodelay_ is set (httplib.h:12083). Per-token SSE frames are
    // tiny writes, so enabling TCP_NODELAY here puts each streamed frame on the
    // wire immediately instead of coalescing it under Nagle/delayed-ACK.
    server.set_tcp_nodelay(true);
  }

  httplib::Server server;
  size_t http_worker_count;
  // The legacy LLMEngine serving constructors remain for small synthetic
  // tests and embedding compatibility. Unlike AsyncLLM, that engine is driven
  // synchronously by its caller, so retain one shared lock for that seam only.
  // Production handlers use AsyncLLM and never take this request-level lock.
  std::mutex legacy_engine_mutex;
};

namespace {

// Build the OpenAI ErrorResponse JSON body for a failed request
// (serve/utils/error_response.py::create_error_response). `code` == the HTTP
// status code (upstream ErrorInfo.code carries it).
ApiServer::DispatchResult MakeError(int status, const std::string& type,
                                    const std::string& message) {
  ErrorResponse err;
  err.error.message = message;
  err.error.type = type;
  err.error.code = status;
  ApiServer::DispatchResult r;
  r.status = status;
  r.content_type = "application/json";
  r.body = nlohmann::json(err).dump();
  return r;
}

}  // namespace

ApiServer::ApiServer(OpenAIServingCompletion& completion,
                     OpenAIServingChat& chat, OpenAIServingModels& models,
                     std::string version, size_t max_concurrent_streams,
                     HttpWorkerPoolMode worker_pool_mode)
    : completion_(&completion),
      chat_(&chat),
      models_(models),
      version_(std::move(version)),
      impl_(std::make_unique<Impl>(max_concurrent_streams, worker_pool_mode)) {}

// Serving-less construction (transcription-only servers, ARCH-ONE-SURFACE
// ROW 1): no AsyncLLM exists, so the generate handlers stay null and their
// routes are not registered — vLLM's task-conditional route registration
// (api_server.py:255-265) expressed at construction.
ApiServer::ApiServer(OpenAIServingModels& models, std::string version,
                     size_t max_concurrent_streams,
                     HttpWorkerPoolMode worker_pool_mode)
    : models_(models),
      version_(std::move(version)),
      impl_(std::make_unique<Impl>(max_concurrent_streams, worker_pool_mode)) {}

ApiServer::~ApiServer() {
  // Drain the async /v1/videos workers before the job store they write into is
  // destroyed. Threads are joined, never detached, precisely so this ordering is
  // guaranteed rather than hoped for.
  std::vector<std::thread> workers;
  {
    std::lock_guard<std::mutex> lock(video_workers_mutex_);
    workers.swap(video_workers_);
  }
  for (auto& worker : workers) {
    if (worker.joinable()) worker.join();
  }
}

ApiServer::DispatchResult ApiServer::handle_completions(
    const std::string& request_body) {
  if (completion_ == nullptr) {
    // vLLM's api_router `if handler is None: raise NotImplementedError` mirror
    // for a serving-less (transcription-only) server; the socket layer never
    // registers the route in that mode, so this answers direct dispatch only.
    return MakeError(500, "InternalServerError",
                     "The model does not support Completions API "
                     "(transcription-only server)");
  }
  // completion/api_router.py:46 (create_completion): parse → check_model →
  // handler → JSON (non-stream) or text/event-stream (stream).
  nlohmann::json body;
  try {
    body = nlohmann::json::parse(request_body);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid JSON body: ") + e.what());
  }
  CompletionRequest request;
  try {
    from_json(body, request);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid request: ") + e.what());
  }
  if (!models_.check_model(request.model)) {
    return MakeError(404, "NotFoundError",
                     "The model `" + request.model.value_or("") +
                         "` does not exist.");
  }

  CompletionResult result;
  try {
    std::unique_lock<std::mutex> legacy_lock(impl_->legacy_engine_mutex,
                                             std::defer_lock);
    if (!completion_->uses_async_engine()) legacy_lock.lock();
    result = completion_->create_completion(request);
  } catch (const std::exception& e) {
    // DISCRIMINATOR: attribute a 500 to its endpoint + model + raw cause so a
    // benchmark driver that only sees the generic HTTP body can still recover
    // the true failure. std::cerr only (survives SIGKILL escalation).
    std::cerr << "api-server: 500 endpoint=/v1/completions model="
              << request.model.value_or("") << " what=" << e.what() << "\n";
    return MakeError(500, "InternalServerError", e.what());
  }

  DispatchResult out;
  if (result.streaming) {
    out.streaming = true;
    out.content_type = "text/event-stream";
    out.sse_chunks = std::move(result.sse_chunks);
    out.sse_stream = std::move(result.sse_stream);
  } else {
    out.status = 200;
    out.content_type = "application/json";
    out.body = nlohmann::json(*result.response).dump();
  }
  return out;
}

ApiServer::DispatchResult ApiServer::handle_chat_completions(
    const std::string& request_body) {
  if (chat_ == nullptr) {
    return MakeError(500, "InternalServerError",
                     "The model does not support Chat Completions API "
                     "(transcription-only server)");
  }
  // chat_completion/api_router.py:53 (create_chat_completion).
  nlohmann::json body;
  try {
    body = nlohmann::json::parse(request_body);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid JSON body: ") + e.what());
  }
  ChatCompletionRequest request;
  try {
    from_json(body, request);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid request: ") + e.what());
  }
  if (!models_.check_model(request.model)) {
    return MakeError(404, "NotFoundError",
                     "The model `" + request.model.value_or("") +
                         "` does not exist.");
  }

  ChatCompletionResult result;
  try {
    std::unique_lock<std::mutex> legacy_lock(impl_->legacy_engine_mutex,
                                             std::defer_lock);
    if (!chat_->uses_async_engine()) legacy_lock.lock();
    result = chat_->create_chat_completion(request);
  } catch (const std::exception& e) {
    std::cerr << "api-server: 500 endpoint=/v1/chat/completions model="
              << request.model.value_or("") << " what=" << e.what() << "\n";
    return MakeError(500, "InternalServerError", e.what());
  }

  DispatchResult out;
  if (result.streaming) {
    out.streaming = true;
    out.content_type = "text/event-stream";
    out.sse_chunks = std::move(result.sse_chunks);
    out.sse_stream = std::move(result.sse_stream);
  } else {
    out.status = 200;
    out.content_type = "application/json";
    out.body = nlohmann::json(*result.response).dump();
  }
  return out;
}

ApiServer::DispatchResult ApiServer::handle_models() const {
  // models/api_router.py:21 (show_available_models).
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = nlohmann::json(models_.show_available_models()).dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_health() const {
  // Upstream calls engine_client.check_health() before returning an empty 200.
  // This bounded server currently exposes process liveness only.
  DispatchResult out;
  out.status = 200;
  out.content_type = "text/plain";
  out.body.clear();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_version() const {
  // serve/instrumentator/basic.py:53 — {"version": <ver>}.
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = nlohmann::json{{"version", version_}}.dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_ping() const {
  // sagemaker/api_router.py:47-50 — GET/POST /ping is a liveness probe that
  // returns the same empty 200 as /health.
  return handle_health();
}

namespace {

ApiServer::DispatchResult VideoJsonOk(std::string body) {
  ApiServer::DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = std::move(body);
  return out;
}

}  // namespace

std::string ApiServer::video_model_warning(
    const ::vllm::openai::VideoRequest& request) const {
  // OpenAI clients send the SORA model name ("sora-2-pro"); this server generates
  // with whatever video model it was started with, whose name they cannot know.
  // Refusing would defeat the compatibility, and ignoring would hide a real
  // mismatch, so the request is honoured and the divergence is STATED on the job.
  if (request.model.empty() || models_.is_base_model(request.model)) return {};
  return "requested model '" + request.model +
         "' is not a served model ('" + models_.model_name() +
         "'); generated with the video model this server was started with";
}

ApiServer::DispatchResult ApiServer::handle_audio_transcriptions(
    const std::string& file_bytes, const std::string& response_format) const {
  // Mirror of vLLM speech_to_text/transcription: api_router.py:31
  // `create_transcriptions` reads the multipart upload
  // (read_upload_with_limit) and hands the bytes to
  // OpenAIServingTranscription.create_transcription (serving.py:50), which
  // answers TranscriptionResponse {"text": ...} for response_format json and
  // the raw text otherwise. The transcription itself runs through the ONE
  // library seam (ParakeetTranscriber) — the same code path vllm_transcribe
  // drives, so HTTP and FFI cannot drift.
  if (!transcriber_) {
    // The api_router `if handler is None: raise NotImplementedError` mirror;
    // the socket layer never registers the route without a transcriber.
    return MakeError(500, "InternalServerError",
                     "The model does not support Transcriptions API");
  }
  if (file_bytes.empty()) {
    return MakeError(400, "BadRequestError",
                     "Expected a non-empty `file` upload (16-bit PCM mono "
                     "RIFF/WAVE)");
  }
  const std::string fmt = response_format.empty() ? "json" : response_format;
  if (fmt != "json" && fmt != "text") {
    // verbose_json / srt / vtt are NAMED RESIDUALS of this fold (protocol.py
    // AudioResponseFormat lists them; nothing here produces segment timing).
    return MakeError(400, "BadRequestError",
                     "response_format '" + fmt +
                         "' is not supported (supported: json, text; "
                         "verbose_json/srt/vtt are named residuals)");
  }
  try {
    const ::vllm::multimodal::ParakeetTranscription result = transcriber_(
        reinterpret_cast<const uint8_t*>(file_bytes.data()), file_bytes.size());
    if (!result.has_text) {
      return MakeError(500, "InternalServerError",
                       "the checkpoint ships no tokenizer.json, so ids-only "
                       "transcription has no OpenAI response shape");
    }
    DispatchResult r;
    if (fmt == "text") {
      r.content_type = "text/plain; charset=utf-8";
      r.body = result.text;
    } else {
      r.body = nlohmann::json{{"text", result.text}}.dump();
    }
    return r;
  } catch (const std::exception& e) {
    // Undecodable audio (not RIFF/WAVE, not PCM16 mono, wrong sample rate) is
    // a caller error; the pipeline names the cause.
    return MakeError(400, "BadRequestError", e.what());
  }
}

ApiServer::DispatchResult ApiServer::handle_videos(
    const std::string& request_body) {
  // vLLM-Omni's ASYNC video endpoint: validate, enqueue, and return the job id
  // immediately. Generation is minutes-long (a 50-step denoise over the packed
  // video+audio sequence), so answering inline would hold an HTTP worker for the
  // whole run -- which is exactly why upstream splits async from /sync.
  if (!video_runner_) {
    return MakeError(500, "InternalServerError", "No video runner configured.");
  }
  ::vllm::openai::VideoRequest request;
  try {
    request = ::vllm::openai::ParseVideoRequest(request_body);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError", e.what());
  }

  const std::string id =
      video_jobs_.Create(request.model, video_model_warning(request));
  std::thread worker([this, id, request]() {
    try {
      video_jobs_.MarkRunning(id);
      video_jobs_.MarkSucceeded(id, video_runner_(request));
    } catch (const std::exception& e) {
      // A runner throw is a FAILED job, not a crashed server: the thread must
      // never let an exception escape (std::terminate) and must always leave the
      // job in a terminal state, or a poller would wait on "running" forever.
      try {
        video_jobs_.MarkFailed(id, e.what());
      } catch (const std::exception&) {
      }
    } catch (...) {
      try {
        video_jobs_.MarkFailed(id, "unknown error");
      } catch (const std::exception&) {
      }
    }
  });
  {
    std::lock_guard<std::mutex> lock(video_workers_mutex_);
    video_workers_.push_back(std::move(worker));
  }

  ::vllm::openai::VideoJob job;
  video_jobs_.Get(id, &job);
  return VideoJsonOk(::vllm::openai::VideoJobStatusJson(job));
}

ApiServer::DispatchResult ApiServer::handle_videos_sync(
    const std::string& request_body) {
  // The SYNCHRONOUS twin: run to completion on the calling worker and answer with
  // the terminal job record. Same runner, same failure mapping -- only the
  // waiting differs.
  if (!video_runner_) {
    return MakeError(500, "InternalServerError", "No video runner configured.");
  }
  ::vllm::openai::VideoRequest request;
  try {
    request = ::vllm::openai::ParseVideoRequest(request_body);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError", e.what());
  }

  const std::string id =
      video_jobs_.Create(request.model, video_model_warning(request));
  try {
    video_jobs_.MarkRunning(id);
    video_jobs_.MarkSucceeded(id, video_runner_(request));
  } catch (const std::exception& e) {
    video_jobs_.MarkFailed(id, e.what());
    return MakeError(500, "InternalServerError", e.what());
  }
  ::vllm::openai::VideoJob job;
  video_jobs_.Get(id, &job);
  return VideoJsonOk(::vllm::openai::VideoJobStatusJson(job));
}

ApiServer::DispatchResult ApiServer::handle_video_status(
    const std::string& job_id) const {
  ::vllm::openai::VideoJob job;
  if (!video_jobs_.Get(job_id, &job)) {
    return MakeError(404, "NotFoundError", "Unknown video job: " + job_id);
  }
  return VideoJsonOk(::vllm::openai::VideoJobStatusJson(job));
}

ApiServer::DispatchResult ApiServer::handle_video_content(
    const std::string& job_id) const {
  // OpenAI's GET /v1/videos/{video_id}/content. Without it a caller can start and
  // poll a job but never FETCH the result over HTTP, which makes the endpoint
  // unusable to anyone without a filesystem view of the server.
  ::vllm::openai::VideoJob job;
  if (!video_jobs_.Get(job_id, &job)) {
    return MakeError(404, "NotFoundError", "Unknown video job: " + job_id);
  }
  if (job.status == ::vllm::openai::VideoJobStatus::kFailed) {
    return MakeError(500, "InternalServerError",
                     "Video job " + job_id + " failed: " + job.error);
  }
  if (job.status != ::vllm::openai::VideoJobStatus::kSucceeded) {
    // A pending job must NEVER answer with bytes: a partially muxed file would
    // reach the client as a valid-looking, truncated MP4.
    return MakeError(409, "ConflictError",
                     std::string("Video job ") + job_id + " is not finished (status: " +
                         ::vllm::openai::VideoJobStatusName(job.status) +
                         "); poll GET /v1/videos/" + job_id + " until it succeeds");
  }

  std::ifstream file(job.output_path, std::ios::binary);
  if (!file) {
    return MakeError(500, "InternalServerError",
                     "Video job " + job_id + " succeeded but its output is not "
                     "readable: " + job.output_path);
  }
  std::string bytes((std::istreambuf_iterator<char>(file)),
                    std::istreambuf_iterator<char>());
  if (!file.eof() && file.fail()) {
    return MakeError(500, "InternalServerError",
                     "Video job " + job_id + " output could not be read in full: " +
                         job.output_path);
  }

  DispatchResult out;
  out.status = 200;
  out.content_type = "video/mp4";
  out.body = std::move(bytes);
  return out;
}

ApiServer::DispatchResult ApiServer::handle_metrics() const {
  // serve/instrumentator/metrics.py:82 — the prometheus text exposition served
  // by make_asgi_app(registry). The PrometheusResponse content type is
  // "text/plain; version=0.0.4; charset=utf-8".
  DispatchResult out;
  out.status = 200;
  out.content_type = v1::metrics::kContentTypeLatest;
  out.body = (metrics_ != nullptr) ? metrics_->Expose() : std::string();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_tokenize(
    const std::string& request_body) const {
  // serve/tokenize/api_router.py:46 (tokenize) over the TokenizeRequest union
  // (serve/tokenize/protocol.py:156): TokenizeCompletionRequest{prompt} OR
  // TokenizeChatRequest{messages, ...}. vLLM discriminates on the body shape
  // (pydantic Union) and, for the chat form, renders the model chat template
  // then tokenizes (serve/tokenize/serving.py:70-124). The response is
  // {count, max_model_len, tokens, token_strs} for both forms (:119).
  nlohmann::json body;
  try {
    body = nlohmann::json::parse(request_body);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid JSON body: ") + e.what());
  }
  if (tokenizer_ == nullptr) {
    return MakeError(500, "InternalServerError", "No tokenizer configured.");
  }

  const bool return_token_strs = body.value("return_token_strs", false);
  std::string prompt;
  // add_special_tokens default differs by form: True for the completion form
  // (protocol.py:28), False for the chat form (protocol.py:78 — the chat
  // template already emits the model's special tokens).
  bool add_special_tokens;

  if (body.contains("messages")) {
    // ── TokenizeChatRequest (serve/tokenize/protocol.py:50). Render the
    // messages through the SAME chat-template seam create_chat_completion
    // tokenizes through (chat_.prompt_fn()), then tokenize — never reinvent
    // template rendering here (serve/tokenize/serving.py:84 preprocess_chat).
    if (!body.at("messages").is_array()) {
      return MakeError(400, "BadRequestError",
                       "`messages` must be an array.");
    }
    // check_generation_prompt (protocol.py:120-128): the two are exclusive.
    const bool add_generation_prompt =
        body.value("add_generation_prompt", true);
    const bool continue_final_message =
        body.value("continue_final_message", false);
    if (add_generation_prompt && continue_final_message) {
      return MakeError(400, "BadRequestError",
                       "Cannot set both `continue_final_message` and "
                       "`add_generation_prompt` to True.");
    }
    add_special_tokens = body.value("add_special_tokens", false);

    std::vector<ChatMessage> messages;
    std::vector<ChatCompletionToolsParam> tools;
    try {
      messages = body.at("messages").get<std::vector<ChatMessage>>();
      if (auto it = body.find("tools");
          it != body.end() && it->is_array()) {
        tools = it->get<std::vector<ChatCompletionToolsParam>>();
      }
    } catch (const std::exception& e) {
      return MakeError(400, "BadRequestError",
                       std::string("Invalid request: ") + e.what());
    }

    // build_chat_params folds add_generation_prompt / continue_final_message
    // into the template kwargs (protocol.py:130-146). Our ChatPromptFn seam
    // renders through the `add_generation_prompt` gate; continue_final_message
    // (open-ended final turn) suppresses the generation header, so it maps to
    // add_generation_prompt=false here.
    const bool render_generation_prompt =
        add_generation_prompt && !continue_final_message;
    try {
      if (chat_ == nullptr) {
        return MakeError(500, "InternalServerError",
                         "tokenize: the chat form needs the chat template of a "
                         "text-generation server (transcription-only server)");
      }
      prompt = chat_->prompt_fn()(messages, render_generation_prompt, tools);
    } catch (const std::exception& e) {
      return MakeError(400, "BadRequestError",
                       std::string("Chat template render failed: ") + e.what());
    }
  } else {
    // ── TokenizeCompletionRequest (serve/tokenize/protocol.py:24).
    if (!body.contains("prompt") || !body.at("prompt").is_string()) {
      return MakeError(400, "BadRequestError",
                       "`prompt` (string) is required.");
    }
    prompt = body.at("prompt").get<std::string>();
    add_special_tokens = body.value("add_special_tokens", true);
  }

  std::vector<int32_t> ids = add_special_tokens
                                 ? tokenizer_->EncodeWithSpecialTokens(prompt)
                                 : tokenizer_->Encode(prompt);

  nlohmann::json resp;
  resp["count"] = ids.size();
  resp["max_model_len"] = max_model_len_;
  resp["tokens"] = ids;
  if (return_token_strs) {
    std::vector<std::string> token_strs;
    token_strs.reserve(ids.size());
    for (int32_t id : ids) token_strs.push_back(tokenizer_->TokenText(id));
    resp["token_strs"] = token_strs;
  } else {
    resp["token_strs"] = nullptr;
  }
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = resp.dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_detokenize(
    const std::string& request_body) const {
  // serve/tokenize/api_router.py:73 (detokenize) over DetokenizeRequest
  // (serve/tokenize/protocol.py:166) → DetokenizeResponse{prompt}.
  nlohmann::json body;
  try {
    body = nlohmann::json::parse(request_body);
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid JSON body: ") + e.what());
  }
  if (!body.contains("tokens") || !body.at("tokens").is_array()) {
    return MakeError(400, "BadRequestError",
                     "`tokens` (array of token ids) is required.");
  }
  if (tokenizer_ == nullptr) {
    return MakeError(500, "InternalServerError", "No tokenizer configured.");
  }
  std::vector<int32_t> ids;
  try {
    for (const auto& t : body.at("tokens")) {
      ids.push_back(t.get<int32_t>());
    }
  } catch (const std::exception& e) {
    return MakeError(400, "BadRequestError",
                     std::string("Invalid token id: ") + e.what());
  }
  nlohmann::json resp;
  resp["prompt"] = tokenizer_->Decode(ids);
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = resp.dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_reset_prefix_cache(
    bool reset_running_requests, bool reset_external) const {
  // serve/dev/cache/api_router.py:20 → {"success": bool}.
  bool success = false;
  if (reset_prefix_cache_) {
    try {
      success = reset_prefix_cache_(reset_running_requests, reset_external);
    } catch (const std::exception& e) {
      return MakeError(500, "InternalServerError", e.what());
    }
  }
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = nlohmann::json{{"success", success}}.dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_server_info() const {
  // serve/dev/server_info/api_router.py:43 — the three-key server_info shape.
  // vllm_config is rendered as a string; vllm_env/system_env are objects. This
  // bounded server exposes version + served model rather than the full config.
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  nlohmann::json info;
  info["vllm_config"] = std::string("served_model_name=") + models_.model_name();
  info["vllm_env"] = nlohmann::json::object();
  info["system_env"] =
      nlohmann::json{{"vllm_cpp_version", version_}};
  out.body = info.dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_tokenizer_info() const {
  // serve/tokenize/api_router.py:95-108 (attach_router → get_tokenizer_info,
  // registered only when app.state.args.enable_tokenizer_info_endpoint) →
  // serve/tokenize/serving.py:154-160 get_tokenizer_info →
  // TokenizerInfo(tokenizer, chat_template).to_dict() (:164-195) →
  // TokenizerInfoResponse (serve/tokenize/protocol.py:185, ConfigDict
  // extra="allow", required `tokenizer_class`).
  //
  // vLLM emits the HF `tokenizer_config.json` init_kwargs verbatim (minus the
  // vocab_file/merges_file paths) plus `tokenizer_class` and optional
  // `chat_template`. We surface EXACTLY the fields our byte-level /
  // SentencePiece BPE tokenizer can genuinely back. Fields vLLM emits that our
  // tokenizer does not carry are OMITTED (never fabricated) and NAMED in
  // specs/utility-endpoints.md: the raw `chat_template` string (our chat
  // template lives in the ChatPromptFn render seam, not the tokenizer), the HF
  // init_kwargs (clean_up_tokenization_spaces / add_bos_token /
  // model_input_names / padding-truncation defaults — not parsed), and the
  // added-token `normalized` / `single_word` flags (not stored on SpecialToken).
  if (tokenizer_ == nullptr) {
    return MakeError(500, "InternalServerError", "No tokenizer configured.");
  }
  nlohmann::json info;
  // `tokenizer_class` (REQUIRED by TokenizerInfoResponse): the genuine BPE
  // family (the HF `tokenizers` library's own class names for these two
  // byte-level / SentencePiece BPE tokenizers).
  info["tokenizer_class"] = tokenizer_->IsSentencePiece()
                                ? "SentencePieceBPETokenizer"
                                : "ByteLevelBPETokenizer";
  if (max_model_len_ > 0) info["model_max_length"] = max_model_len_;
  info["vocab_size"] = tokenizer_->VocabSize();
  if (tokenizer_->BosId() >= 0) info["bos_token_id"] = tokenizer_->BosId();
  if (tokenizer_->EosId() >= 0) info["eos_token_id"] = tokenizer_->EosId();
  // `added_tokens_decoder` mirrors the HF tokenizer_config.json map id →
  // {content, special, lstrip, rstrip}. `normalized`/`single_word` are not
  // tracked by our SpecialToken, so they are omitted (named above).
  nlohmann::json added = nlohmann::json::object();
  for (const vllm::tok::SpecialToken& t : tokenizer_->AddedTokens()) {
    added[std::to_string(t.id)] = {{"content", t.text},
                                   {"special", t.special},
                                   {"lstrip", t.lstrip},
                                   {"rstrip", t.rstrip}};
  }
  info["added_tokens_decoder"] = added;
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body = info.dump();
  return out;
}

ApiServer::DispatchResult ApiServer::handle_abort_requests(
    const std::string& request_body) const {
  // serve/dev/rlhf/api_router.py:94-138 (abort_requests): parse the body,
  // extract `request_ids`; a non-empty list aborts exactly those (external)
  // ids via engine.abort(request_ids); an empty/missing list aborts ALL
  // in-flight requests. Response {"status":"aborted","aborted":<count>}. A
  // malformed body → 400 {"detail":"Invalid JSON format"}; an abort failure →
  // 500 {"error": "..."} (both mirror the upstream router shapes exactly).
  nlohmann::json body;
  try {
    body = nlohmann::json::parse(request_body);
  } catch (const std::exception&) {
    DispatchResult err;
    err.status = 400;
    err.content_type = "application/json";
    err.body = nlohmann::json{{"detail", "Invalid JSON format"}}.dump();
    return err;
  }
  std::vector<std::string> request_ids;
  if (auto it = body.find("request_ids");
      it != body.end() && it->is_array()) {
    try {
      request_ids = it->get<std::vector<std::string>>();
    } catch (const std::exception& e) {
      DispatchResult err;
      err.status = 400;
      err.content_type = "application/json";
      err.body =
          nlohmann::json{{"detail",
                          std::string("Invalid request_ids: ") + e.what()}}
              .dump();
      return err;
    }
  }
  int aborted = 0;
  if (abort_requests_) {
    try {
      aborted = abort_requests_(request_ids);
    } catch (const std::exception& e) {
      DispatchResult err;
      err.status = 500;
      err.content_type = "application/json";
      err.body =
          nlohmann::json{
              {"error", std::string("Failed to abort requests: ") + e.what()}}
              .dump();
      return err;
    }
  }
  DispatchResult out;
  out.status = 200;
  out.content_type = "application/json";
  out.body =
      nlohmann::json{{"status", "aborted"}, {"aborted", aborted}}.dump();
  return out;
}

void ApiServer::register_routes() {
  httplib::Server& server = impl_->server;

  // Write a DispatchResult onto an httplib::Response — either the full JSON body
  // or a chunked text/event-stream (SSE), matching upstream's JSONResponse vs
  // StreamingResponse(media_type="text/event-stream").
  auto write = [](const DispatchResult& result, httplib::Response& res) {
    res.status = result.status;
    if (result.streaming) {
      if (result.sse_stream != nullptr) {
        // W2 live StreamingResponse: one provider invocation pulls one
        // per-request collector output. A slow/disconnected client occupies
        // only its httplib worker; AsyncLLM keeps batching other requests.
        std::shared_ptr<SseStream> stream = result.sse_stream;
        res.set_chunked_content_provider(
            result.content_type,
            [stream](size_t /*offset*/, httplib::DataSink& sink) -> bool {
              try {
                std::string chunk;
                if (!stream->next(chunk)) {
                  sink.done();
                  return true;
                }
                if (!sink.write(chunk.data(), chunk.size())) {
                  stream->abort();
                  return false;
                }
                return true;
              } catch (...) {
                // MID-FLIGHT WITNESS: a provider exception here means the live
                // stream died after headers were already sent (the client sees
                // a truncated body, not a 500). Rethrow-to-inspect so the raw
                // cause reaches stderr before the abort.
                try {
                  std::rethrow_exception(std::current_exception());
                } catch (const std::exception& e) {
                  std::cerr << "sse: stream aborted mid-flight: " << e.what()
                            << "\n";
                } catch (...) {
                  std::cerr << "sse: stream aborted mid-flight: unknown error\n";
                }
                stream->abort();
                return false;
              }
            },
            [stream](bool success) {
              if (!success) stream->abort();
            });
        return;
      }

      // Legacy synchronous compatibility/test seam: write the precomputed
      // chunks exactly as before.
      auto chunks = std::make_shared<std::vector<std::string>>(result.sse_chunks);
      res.set_chunked_content_provider(
          result.content_type,
          [chunks](size_t /*offset*/, httplib::DataSink& sink) -> bool {
            for (const std::string& chunk : *chunks) {
              if (!sink.write(chunk.data(), chunk.size())) return false;
            }
            sink.done();
            return true;
          });
    } else {
      res.set_content(result.body, result.content_type);
    }
  };

  // TASK-CONDITIONAL (mirrors vLLM registering the generate routes only when
  // "generate" is in supported_tasks, api_server.py:255-265): a serving-less
  // (transcription-only) server has no completion/chat handlers, so the two
  // generate routes are NOT registered and answer 404.
  if (completion_ != nullptr) {
    server.Post("/v1/completions",
                [this, write](const httplib::Request& req, httplib::Response& res) {
                  write(handle_completions(req.body), res);
                });
  }
  if (chat_ != nullptr) {
    server.Post("/v1/chat/completions",
                [this, write](const httplib::Request& req, httplib::Response& res) {
                  write(handle_chat_completions(req.body), res);
                });
  }
  server.Get("/v1/models",
             [this, write](const httplib::Request&, httplib::Response& res) {
               write(handle_models(), res);
             });
  server.Get("/health",
             [this, write](const httplib::Request&, httplib::Response& res) {
               write(handle_health(), res);
             });
  server.Get("/version",
             [this, write](const httplib::Request&, httplib::Response& res) {
               write(handle_version(), res);
             });

  // ── C8 additive routes: each registered only when its backing is attached,
  // so a default-constructed server is byte-identical to before. /ping and
  // /server_info are read-only liveness/introspection and always present. ─────
  server.Get("/ping",
             [this, write](const httplib::Request&, httplib::Response& res) {
               write(handle_ping(), res);
             });
  server.Post("/ping",
              [this, write](const httplib::Request&, httplib::Response& res) {
                write(handle_ping(), res);
              });
  server.Get("/server_info",
             [this, write](const httplib::Request&, httplib::Response& res) {
               write(handle_server_info(), res);
             });

  if (transcriber_) {
    // Parakeet ASR (ARCH-ONE-SURFACE ROW 1). Registered ONLY when a
    // transcriber is attached, so a text server answers 404 exactly as before.
    // The multipart shape mirrors vLLM's create_transcriptions
    // (speech_to_text/transcription/api_router.py:31): the audio arrives as
    // the `file` upload, `response_format` as an ordinary form field.
    server.Post("/v1/audio/transcriptions",
                [this, write](const httplib::Request& req,
                              httplib::Response& res) {
                  if (!req.form.has_file("file")) {
                    write(MakeError(400, "BadRequestError",
                                    "multipart/form-data with a `file` upload "
                                    "is required"),
                          res);
                    return;
                  }
                  write(handle_audio_transcriptions(
                            req.form.get_file("file").content,
                            req.form.get_field("response_format")),
                        res);
                });
  }

  if (video_runner_) {
    // MiniMax-H3. Registered ONLY when a runner is attached, so a server built
    // without video support answers 404 exactly as before.
    server.Post("/v1/videos",
                [this, write](const httplib::Request& req,
                              httplib::Response& res) {
                  write(handle_videos(req.body), res);
                });
    server.Post("/v1/videos/sync",
                [this, write](const httplib::Request& req,
                              httplib::Response& res) {
                  write(handle_videos_sync(req.body), res);
                });
    // Registered BEFORE the bare-id pattern so the intent is readable in one
    // place; the two cannot collide in any case, since `[^/]+` stops at the '/'
    // and httplib full-matches the path.
    server.Get(R"(/v1/videos/([^/]+)/content)",
               [this, write](const httplib::Request& req,
                             httplib::Response& res) {
                 write(handle_video_content(req.matches[1]), res);
               });
    server.Get(R"(/v1/videos/([^/]+))",
               [this, write](const httplib::Request& req,
                             httplib::Response& res) {
                 write(handle_video_status(req.matches[1]), res);
               });
  }
  if (metrics_ != nullptr) {
    server.Get("/metrics",
               [this, write](const httplib::Request&, httplib::Response& res) {
                 write(handle_metrics(), res);
               });
  }
  if (tokenizer_ != nullptr) {
    server.Post(
        "/tokenize",
        [this, write](const httplib::Request& req, httplib::Response& res) {
          write(handle_tokenize(req.body), res);
        });
    server.Post(
        "/detokenize",
        [this, write](const httplib::Request& req, httplib::Response& res) {
          write(handle_detokenize(req.body), res);
        });
  }
  if (reset_prefix_cache_) {
    server.Post(
        "/reset_prefix_cache",
        [this, write](const httplib::Request& req, httplib::Response& res) {
          // Query params default false (cache/api_router.py:22-26).
          const bool reset_running =
              req.has_param("reset_running_requests") &&
              req.get_param_value("reset_running_requests") == "true";
          const bool reset_external =
              req.has_param("reset_external") &&
              req.get_param_value("reset_external") == "true";
          write(handle_reset_prefix_cache(reset_running, reset_external), res);
        });
  }
  // GET /tokenizer_info registered only when a tokenizer is attached AND the
  // info endpoint is enabled — mirrors vLLM gating it behind
  // enable_tokenizer_info_endpoint (serve/tokenize/api_router.py:95).
  if (tokenizer_ != nullptr && tokenizer_info_enabled_) {
    server.Get(
        "/tokenizer_info",
        [this, write](const httplib::Request&, httplib::Response& res) {
          write(handle_tokenizer_info(), res);
        });
  }
  // POST /abort_requests registered only when the abort callback is attached
  // (serve/dev/rlhf/api_router.py:94).
  if (abort_requests_) {
    server.Post(
        "/abort_requests",
        [this, write](const httplib::Request& req, httplib::Response& res) {
          write(handle_abort_requests(req.body), res);
        });
  }
}

bool ApiServer::listen(const std::string& host, int port) {
  register_routes();
  return impl_->server.listen(host, port);
}

int ApiServer::bind_to_any_port(const std::string& host) {
  register_routes();
  return impl_->server.bind_to_any_port(host);
}

bool ApiServer::serve() { return impl_->server.listen_after_bind(); }

void ApiServer::stop() { impl_->server.stop(); }

bool ApiServer::is_running() const { return impl_->server.is_running(); }

size_t ApiServer::http_worker_count() const {
  return impl_->http_worker_count;
}

void ConfigureUtilityEndpoints(ApiServer& server,
                               const vllm::tok::Tokenizer& tokenizer,
                               int64_t max_model_len, vllm::v1::AsyncLLM& engine,
                               const UtilityEndpointOptions& options) {
  // /tokenize + /detokenize — ON by default whenever a tokenizer exists (vLLM
  // always calls attach_tokenize_router, serve/tokenize/api_router.py:36,62).
  server.set_tokenizer(&tokenizer, max_model_len);

  // /tokenizer_info — OFF unless --enable-tokenizer-info-endpoint, mirroring
  // vLLM's enable_tokenizer_info_endpoint gate (serve/tokenize/api_router.py:95;
  // cli_args.py:140 default False). When off, the route is never registered → 404.
  if (options.enable_tokenizer_info_endpoint) {
    server.set_tokenizer_info_enabled(true);
  }

  // /abort_requests — DEV-mode gated. vLLM only registers the dev/rlhf router
  // under `if envs.VLLM_SERVER_DEV_MODE` (api_server.py:238; envs.py:157 default
  // 0), so it stays 404 in a default production server. Under
  // --enable-server-dev-mode we wire the abort callback to the live engine:
  // explicit request_ids are aborted via AsyncLLM::abort (async_llm.h:115), and
  // the returned "aborted" count is the drop in unfinished requests (abort()
  // synchronously removes the request states under the output-processor lock, so
  // the delta is exact). The empty-request_ids "abort ALL" contract is a NAMED
  // RESIDUAL: AsyncLLM exposes no active-request-id enumeration, so empty ids
  // abort nothing and report 0 rather than fabricate an all-abort we cannot reach.
  if (options.enable_server_dev_mode) {
    server.set_abort_requests(
        [&engine](const std::vector<std::string>& request_ids) -> int {
          if (request_ids.empty()) {
            return 0;  // abort-ALL residual: no active-request-id accessor.
          }
          const int before = engine.get_num_unfinished_requests();
          engine.abort(request_ids);
          const int after = engine.get_num_unfinished_requests();
          return before > after ? before - after : 0;
        });
  }

  // /metrics and /reset_prefix_cache are deliberately NOT wired here — the
  // production AsyncLLM frontend exposes no live PrometheusStatLogger and no
  // thread-safe prefix-cache reset RPC (see the header block comment +
  // specs/{utility,admin}-endpoints.md). They stay 404, byte-identical to before.
}

}  // namespace vllm::entrypoints::openai