Fabio
c36510c21e
Some checks failed
Linux Build / Linux cpu (push) Has been cancelled
Linux Build / Linux cpu-clang (push) Has been cancelled
Linux Build / Linux vulkan (push) Has been cancelled
mac-build / macOS CPU (push) Has been cancelled
Nix Build / Nix (cpu) (push) Has been cancelled
Nix Build / Nix (vulkan) (push) Has been cancelled
windows-build / Windows CPU (push) Has been cancelled
1231 lines
72 KiB
C++
1231 lines
72 KiB
C++
#include "engine/framework/debug/profiler.h"
|
|
#include "engine/framework/debug/trace.h"
|
|
// Run the actual server handlers with tiny controlled loaders. Barriers force
|
|
// lifecycle overlaps without GPU timing, sleeps in production, or large models.
|
|
#include "parallel_runtime.h"
|
|
#include "engine/framework/audio/wav_writer.h"
|
|
#include "engine/framework/runtime/errors.h"
|
|
#include <chrono>
|
|
#include <condition_variable>
|
|
#include <fstream>
|
|
#include <future>
|
|
#include <iostream>
|
|
#include <map>
|
|
#include <sstream>
|
|
#include <tuple>
|
|
|
|
using namespace std::chrono_literals;
|
|
namespace rt = engine::runtime;
|
|
namespace srv = minitts::server;
|
|
using engine::io::json::parse;
|
|
|
|
namespace {
|
|
void require(bool value, const std::string & message) {
|
|
if (!value) { throw std::runtime_error(message); }
|
|
}
|
|
void concurrent_json_parsing() {
|
|
std::promise<void> start;
|
|
auto ready = start.get_future().share();
|
|
std::vector<std::future<void>> workers;
|
|
for (int worker = 0; worker < 8; ++worker) {
|
|
workers.push_back(std::async(std::launch::async, [ready, worker] {
|
|
ready.wait();
|
|
for (int wave = 0; wave < 200; ++wave) {
|
|
const std::string label = std::to_string(worker) + ":" + std::to_string(wave);
|
|
const auto value = parse("{\"text\":\"" + label + "\",\"requests\":[null,true,1.5,{\"text\":\"hello\"}]}");
|
|
require(value.require("text").as_string() == label &&
|
|
value.require("requests").as_array().size() == 4,
|
|
"concurrent JSON parse mixed request contents");
|
|
require(parse(engine::io::json::stringify(value)).require("text").as_string() == label,
|
|
"concurrent JSON round-trip changed request contents");
|
|
bool rejected = false;
|
|
try { (void)parse("{\"text\":"); }
|
|
catch (const std::runtime_error & error) {
|
|
rejected = std::string(error.what()).find("failed to parse json at byte ") == 0;
|
|
}
|
|
require(rejected, "concurrent invalid JSON lost its own error position");
|
|
}
|
|
}));
|
|
}
|
|
start.set_value();
|
|
for (auto & worker : workers) { worker.get(); }
|
|
}
|
|
template<class F> bool observe(F f) {
|
|
const auto deadline = std::chrono::steady_clock::now() + 2s;
|
|
while (!f()) {
|
|
if (std::chrono::steady_clock::now() >= deadline) { return false; }
|
|
std::this_thread::yield();
|
|
}
|
|
return true;
|
|
}
|
|
struct Gate {
|
|
std::mutex mutex;
|
|
std::condition_variable cv;
|
|
bool enabled = false, entered = false, released = false;
|
|
void arm() { std::lock_guard<std::mutex> lock(mutex); enabled = true; entered = released = false; }
|
|
void wait() {
|
|
std::unique_lock<std::mutex> lock(mutex);
|
|
if (!enabled) { return; }
|
|
entered = true;
|
|
cv.notify_all();
|
|
// A failing test must still drain its futures and release server state.
|
|
cv.wait_for(lock, 5s, [&] { return released; });
|
|
}
|
|
bool seen() { std::lock_guard<std::mutex> lock(mutex); return entered; }
|
|
void release() { std::lock_guard<std::mutex> lock(mutex); released = true; cv.notify_all(); }
|
|
};
|
|
struct Control {
|
|
Gate load, run, destroy;
|
|
std::atomic<int> loads{0}, destructions{0}, runs{0}, finishes{0};
|
|
std::atomic<int> sessions{0}, clones{0}, final_text{0};
|
|
std::atomic<int> last_device{-1};
|
|
std::atomic<bool> fail_load{false}, null_session{false}, fail_prepare{false}, fail_run{false}, fail_clone{false}, reject_run{false};
|
|
std::atomic<bool> premature_model_destruction{false}, live_input{false}, no_text{false}; std::optional<rt::Transcript> live_text;
|
|
};
|
|
class Session final : public rt::IOfflineVoiceTaskSession,
|
|
public rt::IBatchedOfflineVoiceTaskSession,
|
|
public rt::IStreamingVoiceTaskSession,
|
|
public rt::IParallelVoiceTaskSessionFactory {
|
|
public:
|
|
Session(std::shared_ptr<Control> c, rt::TaskSpec task, std::string revision, size_t capacity = 4)
|
|
: control_(std::move(c)), task_(task), revision_(std::move(revision)), capacity_(capacity) { ++control_->sessions; }
|
|
~Session() override { --control_->sessions; }
|
|
std::string family() const override { return "lifecycle_fixture"; }
|
|
rt::VoiceTaskKind task_kind() const override { return task_.task; }
|
|
rt::RunMode run_mode() const override { return task_.mode; }
|
|
void prepare(const rt::SessionPreparationRequest &) override {
|
|
if (control_->fail_prepare) { throw std::runtime_error("controlled preparation failure"); }
|
|
}
|
|
size_t parallel_session_capacity() const noexcept override { return capacity_; }
|
|
std::unique_ptr<rt::IVoiceTaskSession> create_parallel_session() const override {
|
|
if (++control_->clones == 2 && control_->fail_clone) {
|
|
throw std::runtime_error("controlled partial pool failure");
|
|
}
|
|
return std::make_unique<Session>(control_, task_, revision_, capacity_);
|
|
}
|
|
rt::TaskResult run(const rt::TaskRequest & request) override {
|
|
control_->run.wait();
|
|
++control_->runs;
|
|
if (control_->fail_run) { throw std::runtime_error("controlled inference failure"); }
|
|
if (control_->reject_run) { throw rt::InvalidRequestError("controlled request rejection"); }
|
|
if (task_.task == rt::VoiceTaskKind::Asr && !request.audio_input) {
|
|
throw std::runtime_error("replacement ASR requires audio input");
|
|
}
|
|
rt::TaskResult result;
|
|
if (!control_->no_text) { result.text_output = rt::Transcript{revision_, "en"}; }
|
|
result.audio_output = rt::AudioBuffer{16000, 1, {0.1f, -0.1f}}; result.output_artifacts = request.input_artifacts;
|
|
return result;
|
|
}
|
|
std::vector<rt::TaskResult> run_batch(const std::vector<rt::TaskRequest> & requests) override {
|
|
std::vector<rt::TaskResult> results;
|
|
for (const auto & request : requests) { results.push_back(run(request)); }
|
|
return results;
|
|
}
|
|
rt::StreamingPolicy streaming_policy() const override {
|
|
rt::StreamingPolicy policy;
|
|
policy.input = control_->live_input ? rt::StreamingInputKind::AudioChunks : rt::StreamingInputKind::None;
|
|
policy.output = rt::StreamingOutputKind::PullEvents; policy.preferred_audio_chunk_samples = 160;
|
|
return policy;
|
|
}
|
|
void start_stream(const rt::TaskRequest & request) override { result_ = run(request); emitted_ = false; if (control_->live_input) { control_->live_text = request.text_input; } }
|
|
std::optional<rt::StreamEvent> next_stream_event() override {
|
|
if (emitted_) { return {}; }
|
|
emitted_ = true;
|
|
rt::StreamEvent event;
|
|
event.audio_output = result_.audio_output;
|
|
event.output_artifacts = result_.output_artifacts;
|
|
event.partial_text = result_.text_output; if (control_->final_text == 2) { event.partial_text->text += "!"; }
|
|
event.is_final = true;
|
|
return event;
|
|
}
|
|
void reset() override { emitted_ = false; }
|
|
rt::StreamEvent process_audio_chunk(const rt::AudioChunk &) override { return {}; }
|
|
rt::TaskResult finalize() override { ++control_->finishes; auto r = result_; if (control_->final_text == 1) { r.text_output.reset(); } return r; }
|
|
private:
|
|
std::shared_ptr<Control> control_;
|
|
rt::TaskSpec task_;
|
|
std::string revision_;
|
|
size_t capacity_;
|
|
rt::TaskResult result_;
|
|
bool emitted_ = false;
|
|
};
|
|
class Model final : public rt::ILoadedVoiceModel {
|
|
public:
|
|
Model(std::shared_ptr<Control> c, std::string revision) : control_(std::move(c)), revision_(std::move(revision)) {
|
|
metadata_.family = "lifecycle_fixture";
|
|
}
|
|
~Model() override {
|
|
control_->destroy.wait();
|
|
if (control_->sessions != 0) { control_->premature_model_destruction = true; }
|
|
++control_->destructions;
|
|
}
|
|
const rt::ModelMetadata & metadata() const noexcept override { return metadata_; }
|
|
const rt::CapabilitySet & capabilities() const noexcept override { return capabilities_; }
|
|
std::unique_ptr<rt::IVoiceTaskSession> create_task_session(const rt::TaskSpec & task, const rt::SessionOptions & options) const override {
|
|
if (control_->null_session) { return nullptr; }
|
|
control_->last_device = options.backend.device;
|
|
const auto cap = options.options.find("fixture_capacity");
|
|
return std::make_unique<Session>(control_, task, revision_,
|
|
cap == options.options.end() ? 4 : std::stoul(cap->second));
|
|
}
|
|
private:
|
|
std::shared_ptr<Control> control_;
|
|
std::string revision_;
|
|
rt::ModelMetadata metadata_;
|
|
rt::CapabilitySet capabilities_;
|
|
};
|
|
class Loader final : public rt::IVoiceModelLoader {
|
|
public:
|
|
std::map<std::string, std::shared_ptr<Control>> controls;
|
|
std::string family() const override { return "lifecycle_fixture"; }
|
|
bool can_load(const rt::ModelLoadRequest &) const override { return true; }
|
|
rt::ModelInspection inspect(const rt::ModelLoadRequest &) const override { return {}; }
|
|
std::unique_ptr<rt::ILoadedVoiceModel> load(const rt::ModelLoadRequest & request) const override {
|
|
const auto c = controls.at(request.model_path.filename().string());
|
|
++c->loads;
|
|
c->load.wait();
|
|
if (c->fail_load) { throw std::runtime_error("controlled load failure"); }
|
|
return std::make_unique<Model>(c, request.model_path.filename().string());
|
|
}
|
|
};
|
|
std::string quote(const std::string & text) { return engine::io::json::stringify_string(text); }
|
|
srv::HttpResponse post(srv::ParallelServerState & state, const std::string & path, std::string body) {
|
|
srv::HttpRequest request;
|
|
request.method = "POST"; request.path = path; request.body = std::move(body);
|
|
request.headers["content-type"] = "application/json";
|
|
return state.handle(request);
|
|
}
|
|
srv::HttpResponse run(srv::ParallelServerState & state, const std::string & id) {
|
|
return post(state, "/v1/tasks/run", "{\"model\":" + quote(id) + ",\"text\":\"old request\"}");
|
|
}
|
|
void success(const srv::HttpResponse & response) {
|
|
require(response.status == 200, "HTTP " + std::to_string(response.status) + ": " + response.body);
|
|
}
|
|
class Writer final : public srv::HttpStreamWriter {
|
|
public:
|
|
std::string output;
|
|
bool disconnected = false;
|
|
void write(std::string_view s) override {
|
|
if (disconnected) { throw std::runtime_error("client disconnected"); }
|
|
output.append(s);
|
|
}
|
|
};
|
|
}
|
|
|
|
namespace minitts::server {
|
|
class ServerRuntimeTestAccess {
|
|
public:
|
|
static void registry(ParallelServerState & state, const std::shared_ptr<Loader> & loader) {
|
|
state.registry_factory_ = [loader] {
|
|
rt::ModelRegistry registry; registry.register_loader(loader); return registry;
|
|
};
|
|
state.config_.ui_management = true;
|
|
}
|
|
static void add(ParallelServerState & state, ServerModelConfig config) {
|
|
auto model = state.make_model(std::move(config));
|
|
state.model_index_.emplace(model->registered_id, state.models_.size());
|
|
state.models_.push_back(std::move(model));
|
|
}
|
|
static auto & model(ParallelServerState & state, const std::string & id) {
|
|
return *state.models_.at(state.model_index_.at(id));
|
|
}
|
|
static ModelSlots::State slots(ParallelServerState & state, const std::string & id) { return model(state, id).busy.state(); }
|
|
static bool resident(ParallelServerState & state, const std::string & id) { return model(state, id).resident(); }
|
|
static bool loaded(ParallelServerState & state, const std::string & id) { return model(state, id).loaded.load(); }
|
|
static auto block_metadata(ParallelServerState & state, const std::string & id) {
|
|
return std::unique_lock<std::shared_mutex>(model(state, id).metadata_mutex);
|
|
}
|
|
static auto block_bulk_publication(ParallelServerState & state) {
|
|
return std::unique_lock<std::mutex>(state.bulk_admission_mutex_);
|
|
}
|
|
};
|
|
}
|
|
using Access = srv::ServerRuntimeTestAccess;
|
|
|
|
namespace {
|
|
struct Fixture {
|
|
std::filesystem::path root;
|
|
std::shared_ptr<Loader> loader = std::make_shared<Loader>();
|
|
std::unique_ptr<srv::ParallelServerState> state;
|
|
int count;
|
|
Fixture(int slots, int limit = 0, int device = 0) : count(slots) {
|
|
root = std::filesystem::temp_directory_path() /
|
|
("audiocpp-lifecycle-" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()));
|
|
std::filesystem::create_directories(root);
|
|
srv::ServerConfig config; config.backend = engine::core::BackendType::Cpu; config.device = device;
|
|
config.ui_enabled = false; config.max_loaded_models = limit; config.busy_timeout_ms = 0;
|
|
state = std::make_unique<srv::ParallelServerState>(config, root);
|
|
Access::registry(*state, loader);
|
|
}
|
|
~Fixture() {
|
|
for (const auto & [name, c] : loader->controls) {
|
|
c->load.release(); c->run.release(); c->destroy.release();
|
|
}
|
|
state.reset();
|
|
std::error_code ec; std::filesystem::remove_all(root, ec);
|
|
}
|
|
std::shared_ptr<Control> asset(const std::string & name) {
|
|
std::ofstream(root / name) << "controlled fixture";
|
|
auto c = std::make_shared<Control>(); loader->controls.emplace(name, c); return c;
|
|
}
|
|
std::shared_ptr<Control> add(const std::string & id, const std::string & mode = "offline") {
|
|
auto c = asset(id);
|
|
srv::ServerModelConfig model;
|
|
model.id = id; model.path = root / id; model.family = loader->family();
|
|
model.slots = count; model.lazy = true; model.mode = mode;
|
|
Access::add(*state, model); return c;
|
|
}
|
|
auto queued(const std::string & id, int n, bool management = false) {
|
|
return observe([&] { const auto s = Access::slots(*state, id);
|
|
return (management ? s.waiting_management : s.waiting_requests) == n; });
|
|
}
|
|
srv::HttpResponse replace(const std::string & id, const std::string & asset, const std::string & task = "tts") {
|
|
return post(*state, "/v1/models/load", "{\"id\":" + quote(id) + ",\"path\":" +
|
|
quote((root / asset).generic_string()) + ",\"family\":\"lifecycle_fixture\",\"task\":" +
|
|
quote(task) + ",\"slots\":" + std::to_string(count) + "}");
|
|
}
|
|
};
|
|
|
|
void per_model_device_selection() {
|
|
Fixture f(1, 0, 3);
|
|
auto inherited = f.add("inherited");
|
|
auto overridden = f.add("overridden");
|
|
Access::model(*f.state, "overridden").config.device = 7;
|
|
|
|
success(run(*f.state, "inherited"));
|
|
success(run(*f.state, "overridden"));
|
|
require(inherited->last_device == 3, "model without override did not use the server device");
|
|
require(overridden->last_device == 7, "model override did not reach SessionOptions");
|
|
|
|
srv::HttpRequest list; list.method = "GET"; list.path = "/v1/models";
|
|
const auto listed = f.state->handle(list);
|
|
success(listed);
|
|
const auto data = parse(listed.body).require("data").as_array();
|
|
require(data.size() == 2 && data[0].require("device").as_number() == 3 &&
|
|
data[1].require("device").as_number() == 7,
|
|
"model listing did not expose effective devices: " + listed.body);
|
|
|
|
f.asset("replacement");
|
|
success(f.replace("overridden", "replacement"));
|
|
require(overridden->last_device == 7,
|
|
"reloading a configured model without device discarded its assignment");
|
|
|
|
const auto invalid = post(*f.state, "/v1/models/load",
|
|
"{\"id\":\"bad\",\"path\":" + quote((f.root / "replacement").generic_string()) +
|
|
",\"family\":\"lifecycle_fixture\",\"device\":-1,\"slots\":1}");
|
|
require(invalid.status == 400 && invalid.body.find("device for model bad") != std::string::npos,
|
|
"dynamic load accepted a negative model device: " + invalid.body);
|
|
}
|
|
|
|
void older_work_before_management(int count, bool reconfigure) {
|
|
Fixture f(count); auto c = f.add("shared"); f.asset("replacement");
|
|
success(run(*f.state, "shared"));
|
|
c->run.arm();
|
|
std::vector<std::future<srv::HttpResponse>> active;
|
|
for (int i = 0; i < count; ++i) { active.push_back(std::async(std::launch::async, [&] { return run(*f.state, "shared"); })); }
|
|
require(observe([&] { return Access::slots(*f.state, "shared").active == count; }), "active work not established");
|
|
auto old = std::async(std::launch::async, [&] { return run(*f.state, "shared"); });
|
|
const bool old_queued = f.queued("shared", 1);
|
|
auto manager = std::async(std::launch::async, [&] {
|
|
return reconfigure ? f.replace("shared", "replacement", "asr") :
|
|
post(*f.state, "/v1/models/unload", "{\"id\":\"shared\"}");
|
|
});
|
|
const bool manager_queued = f.queued("shared", 1, true);
|
|
c->run.release();
|
|
for (auto & a : active) { success(a.get()); }
|
|
const auto result = old.get(); const auto managed = manager.get();
|
|
require(old_queued && manager_queued, "ordered overlaps not established");
|
|
success(result); success(managed);
|
|
require(result.body.find("shared") != std::string::npos, "older request used replacement state");
|
|
require(c->loads == 1, "queued request reloaded the old model");
|
|
require(Access::resident(*f.state, "shared") == reconfigure, "incorrect final residency");
|
|
const auto s = Access::slots(*f.state, "shared");
|
|
require(s.active == 0 && s.waiting_requests == 0 && s.waiting_management == 0, "ordered drain leaked leases");
|
|
}
|
|
|
|
void bulk_releases_idle_first(int count, bool hold_metadata) {
|
|
Fixture f(count); auto slow = f.add("slow"); auto idle = f.add("idle");
|
|
success(run(*f.state, "idle"));
|
|
slow->load.arm();
|
|
auto loading = std::async(std::launch::async, [&] { return run(*f.state, "slow"); });
|
|
require(observe([&] { return slow->load.seen(); }), "slow initial load did not start");
|
|
std::optional<std::unique_lock<std::shared_mutex>> metadata;
|
|
if (hold_metadata) { metadata.emplace(Access::block_metadata(*f.state, "slow")); }
|
|
auto bulk = std::async(std::launch::async, [&] {
|
|
return post(*f.state, "/v1/tasks/unload_all_models", "{}");
|
|
});
|
|
const bool idle_freed = observe([&] { return idle->destructions == 1; });
|
|
const bool older_loading = loading.wait_for(0ms) == std::future_status::timeout;
|
|
// A full status response may wait for this model's metadata, but must not
|
|
// hold the global registration lock while doing so.
|
|
std::future<srv::HttpResponse> listing;
|
|
std::promise<void> listing_started;
|
|
bool listing_blocked = true;
|
|
if (hold_metadata) {
|
|
auto started = listing_started.get_future();
|
|
listing = std::async(std::launch::async, [&] {
|
|
srv::HttpRequest request; request.method = "GET"; request.path = "/v1/models";
|
|
listing_started.set_value();
|
|
return f.state->handle(request);
|
|
});
|
|
started.wait();
|
|
listing_blocked = listing.wait_for(20ms) == std::future_status::timeout;
|
|
}
|
|
srv::HttpRequest voices; voices.method = "GET"; voices.path = "/v1/audio/voices"; voices.query = "model=idle";
|
|
auto independent = std::async(std::launch::async, [&] { return f.state->handle(voices); });
|
|
const bool unrelated_progress = independent.wait_for(200ms) == std::future_status::ready;
|
|
metadata.reset(); slow->load.release();
|
|
if (listing.valid()) { success(listing.get()); }
|
|
success(independent.get()); success(loading.get()); success(bulk.get());
|
|
require(idle_freed && older_loading, "bulk retained idle model behind a blocked initial load");
|
|
require(listing_blocked && unrelated_progress,
|
|
"bulk/status held the global registry while waiting for model metadata");
|
|
require(!Access::resident(*f.state, "idle") && !Access::resident(*f.state, "slow"), "bulk failed to drain residents");
|
|
}
|
|
|
|
void retiring_resident(int count, int limit) {
|
|
Fixture f(count, limit); auto retiring = f.add("retiring"); auto keeper = f.add("keeper"); auto incoming = f.add("incoming");
|
|
success(run(*f.state, "retiring"));
|
|
if (limit == 2) { success(run(*f.state, "keeper")); }
|
|
retiring->destroy.arm();
|
|
auto unload = std::async(std::launch::async, [&] { return post(*f.state, "/v1/models/unload", "{\"id\":\"retiring\"}"); });
|
|
require(observe([&] { return retiring->destroy.seen(); }), "retirement did not start");
|
|
const bool released = !Access::resident(*f.state, "retiring") && !Access::loaded(*f.state, "retiring");
|
|
auto target = std::async(std::launch::async, [&] { return run(*f.state, "incoming"); });
|
|
const bool concurrent = target.wait_for(500ms) == std::future_status::ready;
|
|
retiring->destroy.release();
|
|
success(target.get()); success(unload.get());
|
|
require(released && concurrent, "retiring model blocked incoming load");
|
|
if (limit == 2) {
|
|
require(Access::resident(*f.state, "keeper") && keeper->loads == 1 && keeper->destructions == 0,
|
|
"retiring model caused unnecessary eviction");
|
|
}
|
|
require(incoming->loads == 1, "incoming model retried instead of loading once");
|
|
}
|
|
|
|
void overlapping_managers(int count) {
|
|
Fixture f(count); auto c = f.add("shared"); f.asset("first"); f.asset("last");
|
|
c->run.arm();
|
|
auto active = std::async(std::launch::async, [&] { return run(*f.state, "shared"); });
|
|
require(observe([&] { return c->run.seen(); }), "manager test did not start inference");
|
|
auto first = std::async(std::launch::async, [&] { return f.replace("shared", "first"); });
|
|
const bool first_queued = f.queued("shared", 1, true);
|
|
auto last = std::async(std::launch::async, [&] { return f.replace("shared", "last"); });
|
|
const bool last_queued = f.queued("shared", 2, true);
|
|
c->run.release();
|
|
success(active.get()); success(first.get()); success(last.get());
|
|
require(first_queued && last_queued, "management order was not established");
|
|
const auto final = run(*f.state, "shared"); success(final);
|
|
require(final.body.find("last") != std::string::npos, "later manager lost to earlier one");
|
|
}
|
|
|
|
void deferred_stream_ownership(int count, bool disconnect) {
|
|
Fixture f(count); auto c = f.add("stream", "streaming"); f.asset("replacement");
|
|
auto response = post(*f.state, "/v1/audio/speech", "{\"model\":\"stream\",\"input\":\"hello\",\"stream\":true,\"response_format\":\"pcm\"}");
|
|
success(response); require(bool(response.stream_body), "missing streaming callback");
|
|
auto manager = std::async(std::launch::async, [&] { return f.replace("stream", "replacement"); });
|
|
const bool queued = f.queued("stream", 1, true);
|
|
const bool protected_state = manager.wait_for(0ms) == std::future_status::timeout;
|
|
Writer writer; writer.disconnected = disconnect;
|
|
bool threw = false;
|
|
try { response.stream_body(writer); } catch (const std::runtime_error &) { threw = true; }
|
|
require(threw == disconnect, "unexpected stream callback result");
|
|
response.stream_body = {}; // transport drops the callback at exchange end
|
|
success(manager.get());
|
|
require(queued && protected_state, "management replaced state before the deferred stream ran");
|
|
require(c->runs == 1 && c->loads == 1, "stream used a different configuration");
|
|
if (!disconnect) { require(writer.output.find("speech.audio.done") != std::string::npos, "stream returned replacement output"); }
|
|
const auto s = Access::slots(*f.state, "stream");
|
|
require(s.active == 0 && s.waiting_management == 0, "stream completion/disconnect leaked ownership");
|
|
}
|
|
|
|
void deferred_batch_ownership(int count, bool generic = false) {
|
|
Fixture f(count); auto c = f.add("batch"); f.asset("replacement");
|
|
const auto wav_path = f.root / "input.wav";
|
|
engine::audio::write_pcm16_wav(wav_path, 16000, 1, {0.1f, -0.1f});
|
|
std::ifstream wav(wav_path, std::ios::binary);
|
|
const std::string audio{std::istreambuf_iterator<char>(wav), std::istreambuf_iterator<char>()};
|
|
srv::HttpRequest request;
|
|
request.method = "POST"; request.path = "/v1/batches/transcriptions";
|
|
request.headers["content-type"] = "multipart/form-data; boundary=test_boundary";
|
|
request.body = "--test_boundary\r\nContent-Disposition: form-data; name=\"model\"\r\n\r\nbatch\r\n";
|
|
for (int i = 0; i < 2; ++i) {
|
|
request.body += "--test_boundary\r\nContent-Disposition: form-data; name=\"file\"; filename=\"input.wav\"\r\n"
|
|
"Content-Type: audio/wav\r\n\r\n" + audio + "\r\n";
|
|
}
|
|
request.body += "--test_boundary--\r\n";
|
|
if (generic) {
|
|
request.path = "/v1/tasks/batch";
|
|
request.headers["content-type"] = "application/json";
|
|
request.body = "{\"model\":\"batch\",\"requests\":[{\"text\":\"first\"},{\"text\":\"second\"}]}";
|
|
}
|
|
auto response = f.state->handle(request); success(response);
|
|
require(bool(response.stream_body), "missing native batch callback");
|
|
auto manager = std::async(std::launch::async, [&] { return f.replace("batch", "replacement"); });
|
|
const bool queued = f.queued("batch", 1, true);
|
|
const bool protected_state = manager.wait_for(0ms) == std::future_status::timeout;
|
|
Writer writer; response.stream_body(writer); response.stream_body = {};
|
|
success(manager.get());
|
|
require(queued && protected_state && c->loads == 1 && c->runs == 2,
|
|
"batch validation/execution crossed a configuration transition");
|
|
require(writer.output.find(generic ? "task.batch.done" : "batch.transcription.done") != std::string::npos &&
|
|
writer.output.find("\"text\":\"batch\"") != std::string::npos,
|
|
"native batch returned replacement output");
|
|
require(Access::slots(*f.state, "batch").active == 0, "native batch leaked a lease");
|
|
}
|
|
void generic_batches_hold_independent_slots(int count) {
|
|
Fixture f(count); auto c = f.add("batch"); f.asset("replacement");
|
|
const std::string body = "{\"model\":\"batch\",\"requests\":[{\"text\":\"first\"},{\"text\":\"second\"}]}";
|
|
auto first = post(*f.state, "/v1/tasks/batch", body);
|
|
auto second = post(*f.state, "/v1/tasks/batch", body);
|
|
success(first); success(second);
|
|
require(Access::slots(*f.state, "batch").active == 2, "generic batches did not lease independent slots");
|
|
auto manager = std::async(std::launch::async, [&] { return f.replace("batch", "replacement"); });
|
|
const bool queued = f.queued("batch", 1, true);
|
|
Writer one, two;
|
|
first.stream_body(one); first.stream_body = {};
|
|
const bool retained = Access::slots(*f.state, "batch").active == 1 &&
|
|
manager.wait_for(0ms) == std::future_status::timeout;
|
|
second.stream_body(two); second.stream_body = {};
|
|
success(manager.get());
|
|
require(queued && retained && c->runs == 4, "generic batch management did not wait for both callbacks");
|
|
require(one.output.find("task.batch.done") != std::string::npos &&
|
|
two.output.find("task.batch.done") != std::string::npos &&
|
|
one.output.find("\"text\":\"batch\"") != std::string::npos &&
|
|
two.output.find("\"text\":\"batch\"") != std::string::npos,
|
|
"generic batches used replacement state");
|
|
require(Access::slots(*f.state, "batch").active == 0, "generic batch callbacks stranded ownership");
|
|
}
|
|
|
|
void generic_batch_single_slot_and_invalid_input() {
|
|
Fixture f(1); f.add("batch");
|
|
auto response = post(*f.state, "/v1/tasks/batch",
|
|
"{\"model\":\"batch\",\"requests\":[{\"text\":\"hello\"}]}");
|
|
success(response); Writer writer; response.stream_body(writer); response.stream_body = {};
|
|
require(writer.output.find("task.batch.done") != std::string::npos, "single-slot parallel generic batch failed");
|
|
require(post(*f.state, "/v1/tasks/batch", "{\"model\":\"batch\",\"requests\":[]}").status == 400,
|
|
"empty generic batch was accepted");
|
|
require(Access::slots(*f.state, "batch").active == 0, "single-slot parallel generic batch leaked ownership");
|
|
}
|
|
|
|
void rejected_preparation_releases_lease(int count) {
|
|
Fixture f(count); f.add("shared");
|
|
bool rejected = false;
|
|
try {
|
|
const auto invalid = post(*f.state, "/v1/tasks/run",
|
|
"{\"model\":\"shared\",\"audio\":" + quote((f.root / "missing.wav").generic_string()) + "}");
|
|
rejected = invalid.status >= 400;
|
|
} catch (const std::runtime_error &) {
|
|
// The transport converts general handler exceptions into HTTP errors.
|
|
rejected = true;
|
|
}
|
|
require(rejected, "missing input was accepted");
|
|
require(Access::slots(*f.state, "shared").active == 0, "failed preparation leaked a bound lease");
|
|
success(run(*f.state, "shared"));
|
|
}
|
|
|
|
template<class F> void fails(F action) {
|
|
bool failed = false;
|
|
try { failed = action().status >= 400; } catch (const std::exception &) { failed = true; }
|
|
require(failed, "controlled operation did not fail");
|
|
}
|
|
|
|
void invalid_reconfiguration_preserves_live_state() {
|
|
Fixture f(2); auto c = f.add("shared"); f.asset("replacement");
|
|
success(run(*f.state, "shared"));
|
|
fails([&] { return f.replace("shared", "replacement", "invalid_task"); });
|
|
require(Access::resident(*f.state, "shared") && c->destructions == 0,
|
|
"invalid reconfiguration destroyed live state");
|
|
const auto result = run(*f.state, "shared"); success(result);
|
|
require(result.body.find("shared") != std::string::npos && c->loads == 1,
|
|
"invalid reconfiguration changed the working configuration");
|
|
}
|
|
|
|
void registration_is_visible_while_loading() {
|
|
Fixture f(2, 1); auto fresh = f.asset("fresh"); auto other = f.add("other");
|
|
fresh->load.arm();
|
|
auto registration = std::async(std::launch::async, [&] { return f.replace("fresh", "fresh"); });
|
|
require(observe([&] { return fresh->load.seen(); }), "new registration did not start loading");
|
|
srv::HttpRequest health; health.method = "GET"; health.path = "/health";
|
|
const auto status = f.state->handle(health); success(status);
|
|
const bool published = parse(status.body).require("models").as_number() == 2;
|
|
auto duplicate = std::async(std::launch::async, [&] { return f.replace("fresh", "fresh"); });
|
|
const bool queued = published && f.queued("fresh", 1, true);
|
|
fresh->load.release();
|
|
success(registration.get());
|
|
bool duplicate_ok = false;
|
|
try { duplicate_ok = duplicate.get().status == 200; } catch (const std::exception &) {}
|
|
require(published && queued && duplicate_ok && fresh->loads == 1,
|
|
"concurrent registration created an invisible/duplicate loaded model");
|
|
success(run(*f.state, "other"));
|
|
require(!Access::resident(*f.state, "fresh") && Access::resident(*f.state, "other") && other->loads == 1,
|
|
"new registration escaped resident-model accounting");
|
|
}
|
|
|
|
void live_policy_does_not_hold_registry_lock() {
|
|
Fixture f(1); f.add("blocked"); f.add("independent");
|
|
auto metadata = Access::block_metadata(*f.state, "blocked");
|
|
std::promise<void> start; auto started = start.get_future();
|
|
auto policy = std::async(std::launch::async, [&] {
|
|
srv::HttpRequest request; request.query = "model=blocked";
|
|
start.set_value(); return f.state->live_ingest_limits(request);
|
|
});
|
|
started.wait();
|
|
const bool waited_for_metadata = policy.wait_for(20ms) == std::future_status::timeout;
|
|
auto independent = std::async(std::launch::async, [&] { return run(*f.state, "independent"); });
|
|
const bool progressed = independent.wait_for(500ms) == std::future_status::ready;
|
|
metadata.unlock(); policy.get(); success(independent.get());
|
|
require(waited_for_metadata && progressed, "live policy raced metadata or blocked the unrelated model registry");
|
|
}
|
|
|
|
void guarded_loading_does_not_block_warm_model(int count) {
|
|
Fixture f(count, 2); auto warm = f.add("warm"); auto cold = f.add("cold");
|
|
success(run(*f.state, "warm")); cold->load.arm();
|
|
auto loading = std::async(std::launch::async, [&] { return run(*f.state, "cold"); });
|
|
require(observe([&] { return cold->load.seen(); }), "guarded cold load did not start");
|
|
auto independent = std::async(std::launch::async, [&] { return run(*f.state, "warm"); });
|
|
const bool progressed = independent.wait_for(500ms) == std::future_status::ready;
|
|
cold->load.release(); success(loading.get()); success(independent.get());
|
|
require(progressed && warm->loads == 1, "guarded cold load blocked/reloaded another warm model");
|
|
}
|
|
|
|
void loading_and_inference_failures_release_ownership(int count) {
|
|
Fixture f(count); auto c = f.add("shared");
|
|
c->fail_load = true;
|
|
fails([&] { return run(*f.state, "shared"); });
|
|
require(!Access::resident(*f.state, "shared") && Access::slots(*f.state, "shared").active == 0,
|
|
"load failure published residency or leaked a lease");
|
|
c->fail_load = false; c->null_session = true;
|
|
fails([&] { return run(*f.state, "shared"); });
|
|
require(c->sessions == 0 && c->destructions == 1 && !Access::resident(*f.state, "shared"),
|
|
"null session did not unwind model ownership");
|
|
c->null_session = false;
|
|
if (count == 4) {
|
|
c->fail_clone = true;
|
|
fails([&] { return run(*f.state, "shared"); });
|
|
require(c->sessions == 0 && c->destructions == 2 && !c->premature_model_destruction,
|
|
"partial pool did not destroy sessions before weights");
|
|
require(!Access::resident(*f.state, "shared") && Access::slots(*f.state, "shared").active == 0,
|
|
"partial pool construction published residency or retained a slot");
|
|
c->fail_clone = false;
|
|
}
|
|
success(run(*f.state, "shared"));
|
|
for (auto * fault : {&c->fail_prepare, &c->fail_run}) {
|
|
*fault = true;
|
|
fails([&] { return run(*f.state, "shared"); });
|
|
require(Access::slots(*f.state, "shared").active == 0, "prepare/run failure retained ownership");
|
|
*fault = false; success(run(*f.state, "shared"));
|
|
}
|
|
// A failed replacement load leaves one valid, unloaded entry that can retry.
|
|
auto replacement = f.asset("replacement"); replacement->fail_load = true;
|
|
fails([&] { return f.replace("shared", "replacement"); });
|
|
if (count > 1) {
|
|
// Multi-slot capability probing occurs before live state changes.
|
|
require(Access::resident(*f.state, "shared") && c->sessions == count,
|
|
"failed capability probe destroyed the working registration");
|
|
replacement->fail_load = false;
|
|
success(f.replace("shared", "replacement"));
|
|
} else {
|
|
require(!Access::resident(*f.state, "shared") && c->sessions == 0,
|
|
"replacement failure left stale sessions/weights");
|
|
replacement->fail_load = false;
|
|
}
|
|
require(!c->premature_model_destruction, "replacement destroyed weights before sessions");
|
|
const auto result = run(*f.state, "shared"); success(result);
|
|
require(result.body.find("replacement") != std::string::npos, "replacement failed to retry consistently");
|
|
}
|
|
|
|
// A request the session turns away as sent is the caller's to fix: a 400 on
|
|
// the generic routes in either run mode, with the lease released. Any other
|
|
// failure still leaves the handler for the transport's 500.
|
|
void session_request_errors(int count) {
|
|
Fixture f(count); auto offline = f.add("offline"); auto stream = f.add("stream", "streaming");
|
|
offline->reject_run = stream->reject_run = true;
|
|
for (const auto & [path, model] : std::vector<std::pair<std::string, std::string>>{
|
|
{"/v1/tasks/run", "offline"}, {"/v1/tasks/run", "stream"}, {"/v1/tasks/stream", "stream"}}) {
|
|
const auto rejected = post(*f.state, path, "{\"model\":" + quote(model) + ",\"text\":\"hi\"}");
|
|
require(rejected.status == 400 && rejected.body.find("controlled request rejection") != std::string::npos &&
|
|
rejected.body.find("invalid_request_error") != std::string::npos,
|
|
path + " on " + model + " answered a rejected request with HTTP " + std::to_string(rejected.status) + ": " + rejected.body);
|
|
require(Access::slots(*f.state, model).active == 0, path + " on " + model + " kept its lease after a rejected request");
|
|
}
|
|
offline->reject_run = false; offline->fail_run = true;
|
|
bool server_error = false; std::string outcome;
|
|
try {
|
|
const auto failed = run(*f.state, "offline");
|
|
server_error = failed.status == 500; outcome = "HTTP " + std::to_string(failed.status) + ": " + failed.body;
|
|
} catch (const std::runtime_error & error) {
|
|
outcome = error.what(); server_error = outcome == "controlled inference failure";
|
|
}
|
|
require(server_error, "a failure that is not a rejected request was answered as " + outcome);
|
|
offline->fail_run = false; success(run(*f.state, "offline"));
|
|
}
|
|
|
|
void abandoned_deferred_responses_release_ownership(int count) {
|
|
Fixture f(count); f.add("stream", "streaming"); f.add("batch");
|
|
for (const auto & [id, path, body] : std::vector<std::tuple<std::string, std::string, std::string>>{
|
|
{"stream", "/v1/audio/speech", "{\"model\":\"stream\",\"input\":\"hello\",\"stream\":true,\"response_format\":\"pcm\"}"},
|
|
{"batch", "/v1/tasks/batch", "{\"model\":\"batch\",\"requests\":[{\"text\":\"hello\"}]}"}}) {
|
|
auto response = post(*f.state, path, body); success(response);
|
|
require(bool(response.stream_body) && Access::slots(*f.state, id).active == 1,
|
|
"deferred response did not own its slot");
|
|
response = {}; // Dropped without invoking the callback.
|
|
require(Access::slots(*f.state, id).active == 0, "unstarted response leaked its lease");
|
|
success(post(*f.state, "/v1/models/unload", "{\"id\":" + quote(id) + "}"));
|
|
}
|
|
}
|
|
|
|
void overlapping_bulk_operations(int count) {
|
|
Fixture f(count); auto a = f.add("a"); auto b = f.add("b"); f.add("unrelated");
|
|
a->run.arm(); b->run.arm();
|
|
auto run_a = std::async(std::launch::async, [&] { return run(*f.state, "a"); });
|
|
auto run_b = std::async(std::launch::async, [&] { return run(*f.state, "b"); });
|
|
require(observe([&] { return a->run.seen() && b->run.seen(); }), "bulk active requests not established");
|
|
auto first = std::async(std::launch::async, [&] {
|
|
return post(*f.state, "/v1/tasks/unload_models", "{\"model_ids\":[\"a\",\"b\",\"a\"]}");
|
|
});
|
|
const bool first_queued = f.queued("a", 1, true) && f.queued("b", 1, true);
|
|
auto second = std::async(std::launch::async, [&] {
|
|
return post(*f.state, "/v1/tasks/unload_models", "{\"model_ids\":[\"b\",\"a\"]}");
|
|
});
|
|
const bool second_queued = f.queued("a", 2, true) && f.queued("b", 2, true);
|
|
auto independent = std::async(std::launch::async, [&] { return run(*f.state, "unrelated"); });
|
|
const bool progressed = independent.wait_for(500ms) == std::future_status::ready;
|
|
a->run.release(); b->run.release();
|
|
success(run_a.get()); success(run_b.get()); success(independent.get());
|
|
const auto result = first.get(); success(result); success(second.get());
|
|
require(first_queued && second_queued && progressed, "bulk barriers duplicated or blocked an unrelated model");
|
|
require(parse(result.body).require("unloaded").as_array().size() == 2, "bulk response repeated a model id");
|
|
for (const auto & id : {"a", "b"}) {
|
|
const auto s = Access::slots(*f.state, id);
|
|
require(s.active == 0 && s.waiting_management == 0 && s.waiting_requests == 0,
|
|
"overlapping bulk operations leaked a barrier");
|
|
success(run(*f.state, id));
|
|
}
|
|
}
|
|
void bulk_publication_excludes_only_other_bulk_calls(int count) {
|
|
Fixture f(count); f.add("a"); f.add("b"); f.add("unrelated");
|
|
auto publication = Access::block_bulk_publication(*f.state);
|
|
std::promise<void> one, two;
|
|
auto started_one = one.get_future(); auto started_two = two.get_future();
|
|
auto first = std::async(std::launch::async, [&] {
|
|
one.set_value(); return post(*f.state, "/v1/tasks/unload_models", "{\"model_ids\":[\"a\",\"b\"]}");
|
|
});
|
|
auto second = std::async(std::launch::async, [&] {
|
|
two.set_value(); return post(*f.state, "/v1/tasks/unload_models", "{\"model_ids\":[\"b\",\"a\"]}");
|
|
});
|
|
started_one.wait(); started_two.wait();
|
|
const bool excluded = first.wait_for(20ms) == std::future_status::timeout &&
|
|
second.wait_for(20ms) == std::future_status::timeout &&
|
|
Access::slots(*f.state, "a").waiting_management == 0 &&
|
|
Access::slots(*f.state, "b").waiting_management == 0;
|
|
auto independent = std::async(std::launch::async, [&] { return run(*f.state, "unrelated"); });
|
|
const bool progressed = independent.wait_for(500ms) == std::future_status::ready;
|
|
publication.unlock(); success(first.get()); success(second.get()); success(independent.get());
|
|
require(excluded && progressed, "bulk publication interleaved barriers or blocked normal inference");
|
|
}
|
|
}
|
|
|
|
void unsupported_capacity_preserves_registration() {
|
|
for (int count : {1, 2, 4}) {
|
|
Fixture f(count); auto old = f.add("shared"); auto replacement = f.asset("replacement");
|
|
success(run(*f.state, "shared"));
|
|
const auto before = Access::model(*f.state, "shared").sessions.get();
|
|
const auto reject = [&](const std::string & asset, int slots, const std::string & extra = "") {
|
|
srv::HttpResponse response;
|
|
try {
|
|
response = post(*f.state, "/v1/models/load",
|
|
"{\"id\":\"shared\",\"family\":\"lifecycle_fixture\",\"path\":" +
|
|
quote((f.root / asset).generic_string()) + ",\"slots\":" + std::to_string(slots) + extra + "}");
|
|
} catch (const std::exception & error) {
|
|
response.status = 500; response.body = error.what();
|
|
}
|
|
const std::string model_id = "shared";
|
|
const auto & model = Access::model(*f.state, model_id);
|
|
require(model.sessions.get() == before && model.config.slots == count &&
|
|
model.config.path == f.root / "shared" && Access::resident(*f.state, "shared") &&
|
|
old->destructions == 0 && old->loads == 1,
|
|
"unsupported capacity destroyed or changed a working registration");
|
|
require(response.status == 400 && response.body.find("capacity=") != std::string::npos,
|
|
"unsupported capacity was not rejected as HTTP 400");
|
|
success(run(*f.state, "shared"));
|
|
const auto state = Access::slots(*f.state, "shared");
|
|
require(state.slots == count && state.active == 0 && state.waiting_management == 0,
|
|
"unsupported capacity stranded admission ownership");
|
|
};
|
|
// Valid schema but unsupported capacity: the maintainer's slots=5 case.
|
|
reject("shared", 5);
|
|
// A different checkpoint or session options can advertise a different cap.
|
|
reject("replacement", 5);
|
|
reject("shared", 2, ",\"session_options\":{\"fixture_capacity\":\"1\"}");
|
|
require(replacement->sessions == 0 && !replacement->premature_model_destruction,
|
|
"capability probe retained state or destroyed weights before its session");
|
|
f.count = count == 1 ? 2 : 1;
|
|
success(f.replace("shared", "shared"));
|
|
require(Access::slots(*f.state, "shared").slots == f.count,
|
|
"valid resize after rejection could not recover");
|
|
}
|
|
}
|
|
|
|
void dynamic_capacity_changes_use_parallel_lifecycle() {
|
|
Fixture f(1); f.add("shared");
|
|
success(run(*f.state, "shared"));
|
|
for (int count : {2, 1, 4, 1}) {
|
|
f.count = count;
|
|
success(f.replace("shared", "shared", "tts"));
|
|
const auto state = Access::slots(*f.state, "shared");
|
|
require(state.slots == count, "parallel runtime could not reconfigure through capacity one");
|
|
success(run(*f.state, "shared"));
|
|
}
|
|
for (const std::string value : {"null", "true", "\"2\"", "1.5", "17"}) {
|
|
bool rejected = false;
|
|
try {
|
|
const auto invalid = post(*f.state, "/v1/models/load",
|
|
"{\"id\":\"shared\",\"family\":\"lifecycle_fixture\",\"path\":" + quote((f.root / "shared").generic_string()) +
|
|
",\"slots\":" + value + "}");
|
|
rejected = invalid.status == 400;
|
|
} catch (const std::runtime_error & error) {
|
|
// A direct validation exception must also identify the bad capacity.
|
|
rejected = std::string(error.what()).find("model slots must be an integer") != std::string::npos;
|
|
}
|
|
require(rejected, "parallel registration accepted invalid capacity");
|
|
require(Access::resident(*f.state, "shared"), "invalid capacity unloaded an existing model");
|
|
}
|
|
const auto ordered = post(*f.state, "/v1/models/load",
|
|
"{\"id\":\"shared\",\"family\":\"lifecycle_fixture\",\"path\":" +
|
|
quote((f.root / "shared").generic_string()) +
|
|
",\"slots\":null,\"load_options\":[]}");
|
|
require(ordered.status == 400 &&
|
|
ordered.body.find("model slots must be an integer") != std::string::npos,
|
|
"parallel registration parsed options before validating capacity");
|
|
}
|
|
|
|
// An SSE stream's event types in order, its joined text deltas and its final text.
|
|
std::tuple<std::string, std::string, std::string> sse_events(const std::string & sse) {
|
|
std::string types, deltas, done;
|
|
for (size_t at = sse.find("data: {"); at != std::string::npos; at = sse.find("data: {", at + 1)) {
|
|
const auto event = parse(sse.substr(at + 6, sse.find("\n\n", at) - at - 6));
|
|
const auto type = event.require("type").as_string();
|
|
if (type == "speech.text.delta") { deltas += event.require("delta").as_string(); }
|
|
if (type == "speech.text.done") { done = event.require("text").as_string(); }
|
|
types += (types.empty() ? "" : " ") + type;
|
|
}
|
|
return {types, deltas, done};
|
|
}
|
|
|
|
// With live_input the fixture reads the body as audio chunks, then writes its
|
|
// revision ("live") as text in the one event that carries its audio.
|
|
void live_speech_return_text(int count) {
|
|
Fixture f(count); auto c = f.add("live", "streaming"); c->live_input = true;
|
|
std::string error;
|
|
const auto live = [&](const std::string & query) {
|
|
std::istringstream pcm(std::string(640, '\0'));
|
|
srv::HttpRequest request; request.method = "POST"; request.path = "/v1/audio/speech/live";
|
|
request.query = "model=live&input=hello" + query; request.body_stream = &pcm;
|
|
auto response = f.state->handle(request);
|
|
Writer writer; error.clear();
|
|
try { if (response.stream_body) { response.stream_body(writer); } }
|
|
catch (const std::runtime_error & e) { error = e.what(); }
|
|
if (response.stream_body) { response.body = writer.output; response.stream_body = {}; }
|
|
require(Access::slots(*f.state, "live").active == 0, "live speech '" + query + "' left its lease held");
|
|
return response;
|
|
};
|
|
const auto text = live("&return_text=true"); success(text);
|
|
const auto [types, deltas, done] = sse_events(text.body);
|
|
require(deltas == "live" && done == deltas, "return_text did not stream the model's text: " + types);
|
|
require(types == "speech.text.delta speech.audio.delta speech.text.done speech.audio.done" && error.empty(),
|
|
"return_text events out of order: " + types);
|
|
for (const std::string query : {"", "&return_text=false", "&return_text=0"}) {
|
|
const auto audio = live(query); success(audio);
|
|
require(std::get<0>(sse_events(audio.body)) == "speech.audio.delta speech.audio.done" && error.empty(),
|
|
"live speech '" + query + "' streamed text");
|
|
}
|
|
for (const auto & [query, message] : std::vector<std::pair<std::string, std::string>>{
|
|
{"&return_text=yes", "live speech return_text must be true or false"},
|
|
{"&return_text=true&stream_format=audio", "live speech return_text needs stream_format=sse"}}) {
|
|
const auto rejected = live(query);
|
|
require(rejected.status == 400 && rejected.body.find(message) != std::string::npos,
|
|
"live speech '" + query + "' was not rejected: HTTP " + std::to_string(rejected.status));
|
|
}
|
|
// speech.text.done is the model's final text, or the joined deltas when it gives none.
|
|
for (const auto & [mode, streamed] : std::vector<std::pair<int, std::string>>{{1, "live"}, {2, "live!"}}) {
|
|
c->final_text = mode;
|
|
const auto [order, joined, done_text] = sse_events(live("&return_text=true").body);
|
|
require(joined == streamed && done_text == "live" && order == types && error.empty(),
|
|
"speech.text.done '" + done_text + "' after deltas '" + joined + "': " + error);
|
|
}
|
|
c->final_text = 0; c->no_text = true;
|
|
const auto silent = live("&return_text=1");
|
|
require(silent.status == 200 && std::get<0>(sse_events(silent.body)) == "speech.audio.delta" &&
|
|
error == "live speech model wrote no text; return_text needs a model that writes text",
|
|
"return_text without model text did not fail after the audio: " + error);
|
|
}
|
|
|
|
// On the live route `input` is the text a tts model speaks, so it stays
|
|
// required there. An s2s or vc model answers or converts the live audio and
|
|
// takes `input` as an optional prompt: without it the model gets no text_input
|
|
// and uses its own default, as /v1/tasks/run does without `text`.
|
|
void live_speech_optional_input(int count) {
|
|
Fixture f(count);
|
|
std::map<std::string, std::shared_ptr<Control>> controls;
|
|
for (const std::string task : {"s2s", "vc", "tts"}) {
|
|
auto c = f.asset(task); c->live_input = true; controls[task] = c;
|
|
srv::ServerModelConfig model;
|
|
model.id = task; model.path = f.root / task; model.family = f.loader->family();
|
|
model.task = task; model.mode = "streaming"; model.slots = count; model.lazy = true;
|
|
Access::add(*f.state, model);
|
|
}
|
|
std::string error;
|
|
const auto live = [&](const std::string & query) {
|
|
std::istringstream pcm(std::string(640, '\0'));
|
|
srv::HttpRequest request; request.method = "POST"; request.path = "/v1/audio/speech/live";
|
|
request.query = query; request.body_stream = &pcm;
|
|
auto response = f.state->handle(request);
|
|
Writer writer; error.clear();
|
|
try { if (response.stream_body) { response.stream_body(writer); } }
|
|
catch (const std::runtime_error & e) { error = e.what(); }
|
|
if (response.stream_body) { response.body = writer.output; response.stream_body = {}; }
|
|
return response;
|
|
};
|
|
for (const std::string task : {"s2s", "vc"}) {
|
|
const auto & c = controls.at(task);
|
|
for (const std::string input : {"", "&input="}) {
|
|
c->live_text = rt::Transcript{"stale", ""};
|
|
const auto response = live("model=" + task + input); success(response);
|
|
require(std::get<0>(sse_events(response.body)) == "speech.audio.delta speech.audio.done" && error.empty(),
|
|
task + " live speech without input did not stream: " + error);
|
|
require(!c->live_text.has_value(), task + " live speech without input passed a text_input");
|
|
}
|
|
success(live("model=" + task + "&input=Be%20brief."));
|
|
require(c->live_text.has_value() && c->live_text->text == "Be brief.", task + " live speech lost its input");
|
|
success(live("model=" + task + "&language=ja"));
|
|
require(c->live_text.has_value() && c->live_text->text.empty() && c->live_text->language == "ja",
|
|
task + " live speech without input lost its language");
|
|
}
|
|
for (const std::string input : {"", "&input="}) {
|
|
const auto rejected = live("model=tts" + input);
|
|
require(rejected.status == 400 &&
|
|
rejected.body.find("live speech requires an 'input' query parameter") != std::string::npos,
|
|
"tts live speech without input was not rejected: HTTP " + std::to_string(rejected.status));
|
|
}
|
|
require(controls.at("tts")->runs == 0, "tts live speech without input reached the model");
|
|
for (const std::string task : {"s2s", "vc", "tts"}) {
|
|
require(Access::slots(*f.state, task).active == 0, task + " live speech left its lease held");
|
|
}
|
|
}
|
|
|
|
// The generic routes read a request `artifacts` array into input_artifacts. The
|
|
// fixture returns its input artifacts as its result's, so a response shows what
|
|
// the session was given, in order, in the JSON shape the request used.
|
|
void generic_request_artifacts(int count) {
|
|
Fixture f(count); auto offline = f.add("offline"); auto stream = f.add("stream", "streaming");
|
|
const auto file = f.root / "reply.bin";
|
|
std::ofstream(file, std::ios::binary) << std::string("\x01\x00\xff", 3);
|
|
std::vector<std::vector<std::string>> expected{
|
|
{"turn.question", "custom", "aGVsbG8="}, {"turn.reply", "acoustic_tokens", "AQD/"},
|
|
{"turn.question", "custom", "d29ybGQ="}};
|
|
std::string artifacts = "[{\"id\":\"turn.question\",\"kind\":\"custom\",\"payload\":\"aGVsbG8=\","
|
|
"\"meta\":{\"format\":\"wav\",\"steps\":3}},"
|
|
"{\"id\":\"turn.reply\",\"kind\":\"acoustic_tokens\",\"path\":" + quote(file.generic_string()) + "},"
|
|
"{\"id\":\"turn.question\",\"kind\":\"custom\",\"payload\":\"data:audio/wav;base64,d29ybGQ=\"}";
|
|
// Every kind name a response writes reads back as the same kind.
|
|
for (const std::string kind : {"speaker_embedding", "style_embedding", "prompt_embedding", "acoustic_tokens",
|
|
"midi", "transcript_alignment", "diarization_state", "vad_state", "custom"}) {
|
|
artifacts += ",{\"id\":\"kind\",\"kind\":" + quote(kind) + ",\"payload\":\"\"}";
|
|
expected.push_back({"kind", kind, ""});
|
|
}
|
|
artifacts += "]";
|
|
const auto check = [&](const engine::io::json::Value & result, const std::string & route) {
|
|
const auto & out = result.require("artifacts").as_array();
|
|
require(out.size() == expected.size(), route + " returned " + std::to_string(out.size()) + " artifacts");
|
|
for (size_t i = 0; i < out.size(); ++i) {
|
|
require(out[i].require("id").as_string() == expected[i][0] &&
|
|
out[i].require("kind").as_string() == expected[i][1] &&
|
|
out[i].require("payload").as_string() == expected[i][2],
|
|
route + " changed artifact " + std::to_string(i));
|
|
}
|
|
const auto & meta = out[0].require("meta");
|
|
require(meta.as_object().size() == 2 && meta.require("format").as_string() == "wav" &&
|
|
meta.require("steps").as_string() == "3" && out[1].require("meta").as_object().empty(),
|
|
route + " changed artifact meta");
|
|
};
|
|
const auto body = [&](const std::string & model, bool wrapped) {
|
|
const std::string request = "\"text\":\"hi\",\"artifacts\":" + artifacts;
|
|
return "{\"model\":" + quote(model) + (wrapped ? ",\"request\":{" + request + "}}" : "," + request + "}");
|
|
};
|
|
for (const bool wrapped : {false, true}) {
|
|
const auto ran = post(*f.state, "/v1/tasks/run", body("offline", wrapped)); success(ran);
|
|
check(parse(ran.body), "offline /v1/tasks/run");
|
|
const auto streamed_run = post(*f.state, "/v1/tasks/run", body("stream", wrapped)); success(streamed_run);
|
|
check(parse(streamed_run.body), "streaming /v1/tasks/run");
|
|
const auto streamed = post(*f.state, "/v1/tasks/stream", body("stream", wrapped)); success(streamed);
|
|
check(parse(streamed.body).require("result"), "/v1/tasks/stream");
|
|
}
|
|
auto batch = post(*f.state, "/v1/tasks/batch", "{\"model\":\"offline\",\"requests\":[{\"text\":\"hi\",\"artifacts\":" +
|
|
artifacts + "},{\"text\":\"plain\"}]}");
|
|
success(batch); Writer writer; batch.stream_body(writer); batch.stream_body = {};
|
|
int results = 0;
|
|
for (size_t at = writer.output.find("data: {"); at != std::string::npos; at = writer.output.find("data: {", at + 1)) {
|
|
const auto event = parse(writer.output.substr(at + 6, writer.output.find("\n\n", at) - at - 6));
|
|
if (event.require("type").as_string() != "task.batch.result") { continue; }
|
|
++results;
|
|
const auto & result = event.require("result");
|
|
if (event.require("index").as_i64() == 0) { check(result, "/v1/tasks/batch"); }
|
|
else { require(result.find("artifacts") == nullptr, "batch request without artifacts got some"); }
|
|
}
|
|
require(results == 2, "batch with artifacts returned " + std::to_string(results) + " results");
|
|
require(run(*f.state, "offline").body.find("\"artifacts\"") == std::string::npos,
|
|
"request without artifacts returned some");
|
|
const int offline_runs = offline->runs, stream_runs = stream->runs;
|
|
for (const auto & [bad, message] : std::vector<std::pair<std::string, std::string>>{
|
|
{"{\"id\":\"x\"}", "artifacts must be an array of artifact objects"},
|
|
{"[{\"id\":\"x\",\"kind\":\"tokens\",\"payload\":\"\"}]", "artifacts[0] (x): unknown kind 'tokens'"},
|
|
{"[{\"id\":\"x\",\"kind\":\"custom\",\"payload\":\"%%\"}]", "artifacts[0] (x): payload is not valid base64"},
|
|
{"[{\"id\":\"x\",\"kind\":\"custom\",\"path\":" + quote((f.root / "missing.bin").generic_string()) + "}]",
|
|
"artifacts[0] (x): path is not a regular file"}}) {
|
|
for (const auto & [path, request] : std::vector<std::pair<std::string, std::string>>{
|
|
{"/v1/tasks/run", "{\"model\":\"offline\",\"artifacts\":" + bad + "}"},
|
|
{"/v1/tasks/run", "{\"model\":\"stream\",\"request\":{\"artifacts\":" + bad + "}}"},
|
|
{"/v1/tasks/stream", "{\"model\":\"stream\",\"request\":{\"artifacts\":" + bad + "}}"},
|
|
{"/v1/tasks/batch", "{\"model\":\"offline\",\"requests\":[{\"text\":\"hi\"},{\"artifacts\":" + bad + "}]}"}}) {
|
|
const auto rejected = post(*f.state, path, request);
|
|
require(rejected.status == 400 && rejected.body.find(message) != std::string::npos &&
|
|
rejected.body.find("invalid_request_error") != std::string::npos,
|
|
path + " did not reject artifacts " + bad + ": HTTP " + std::to_string(rejected.status) + " " + rejected.body);
|
|
}
|
|
}
|
|
require(offline->runs == offline_runs && stream->runs == stream_runs, "rejected artifacts reached the model");
|
|
for (const std::string id : {"offline", "stream"}) {
|
|
require(Access::slots(*f.state, id).active == 0, id + " left its lease held after artifact requests");
|
|
}
|
|
}
|
|
|
|
// With a top-level "stream_format":"sse", /v1/tasks/stream writes each event
|
|
// as the model produces it, as the object the JSON response puts in "events",
|
|
// then the JSON response's "result" and [DONE]. Without the field, or with
|
|
// null, the response is the JSON body. The fixture counts finalize calls, so a
|
|
// writer that records the count at each write shows when an event left.
|
|
void stream_artifacts_are_output() {
|
|
rt::StreamEvent event;
|
|
require(!srv::stream_event_has_output(event), "empty event counted as output");
|
|
require(srv::stream_event_json(event,false)=="{\"is_final\":false}", "empty event shape changed");
|
|
event.output_artifacts.push_back(rt::make_voice_artifact(rt::ArtifactKind::Custom,"frame",{std::byte{0},std::byte{255}},{{"format","rgb24"}}));
|
|
require(!srv::stream_event_has_output(event), "artifact-only event changed default output timing");
|
|
require(parse(srv::stream_event_json(event,false)).find("artifacts") == nullptr, "artifacts leaked without opt-in");
|
|
require(srv::stream_event_has_output(event, true), "opt-in artifact-only event not counted as output");
|
|
const auto encoded=parse(srv::stream_event_json(event,false,true));
|
|
const auto & artifacts=encoded.require("artifacts").as_array();
|
|
require(artifacts.size()==1 && artifacts[0].require("id").as_string()=="frame" &&
|
|
artifacts[0].require("payload").as_string()=="AP8=" &&
|
|
artifacts[0].require("meta").require("format").as_string()=="rgb24", "stream artifact was lost or changed");
|
|
rt::TaskResult result;result.output_artifacts=event.output_artifacts;
|
|
const auto final=parse(srv::task_result_json(result,0));
|
|
require(engine::io::json::stringify(final.require("artifacts"))==engine::io::json::stringify(encoded.require("artifacts")),
|
|
"stream and final artifact formats differ");
|
|
}
|
|
|
|
void generic_task_stream_sse(int count) {
|
|
Fixture f(count); auto c = f.add("stream", "streaming");
|
|
const auto active = [&] { return Access::slots(*f.state, "stream").active; };
|
|
const auto body = [](const std::string & fields, bool wrapped) {
|
|
const std::string request = "\"text\":\"hi\",\"artifacts\":[{\"id\":\"turn.reply\",\"kind\":\"acoustic_tokens\","
|
|
"\"payload\":\"AQD/\",\"meta\":{\"steps\":\"3\"}}]";
|
|
return "{\"model\":\"stream\"" + fields + (wrapped ? ",\"request\":{" + request + "}}" : "," + request + "}");
|
|
};
|
|
const std::string sse = ",\"stream_format\":\"sse\"";
|
|
const auto stream = [&](const std::string & body_text) { return post(*f.state, "/v1/tasks/stream", body_text); };
|
|
// The JSON body's event and result as text: a parse and stringify round
|
|
// trip reorders object keys, so comparisons use the bytes as written.
|
|
const auto json_parts = [](const srv::HttpResponse & response) {
|
|
const std::string events = "{\"events\":[", result = "],\"result\":";
|
|
const auto end = response.body.find(result);
|
|
require(response.status == 200 && response.content_type == "application/json" && !response.stream_body &&
|
|
response.body.rfind(events, 0) == 0 && end != std::string::npos,
|
|
"task stream did not answer its JSON body: HTTP " + std::to_string(response.status) + " " + response.body);
|
|
const auto from = end + result.size();
|
|
return std::make_pair(response.body.substr(events.size(), end - events.size()),
|
|
response.body.substr(from, response.body.size() - from - 1));
|
|
};
|
|
const auto without_ttft = [](std::string json) {
|
|
const std::string key = "\"ttft_ms\":";
|
|
const auto at = json.find(key);
|
|
require(at != std::string::npos, "task stream result has no ttft_ms: " + json);
|
|
json.erase(at + key.size(), json.find_first_of(",}", at + key.size()) - at - key.size());
|
|
return json;
|
|
};
|
|
struct Recorder final : srv::HttpStreamWriter {
|
|
explicit Recorder(std::shared_ptr<Control> counted) : control(std::move(counted)) {}
|
|
void write(std::string_view s) override { writes.emplace_back(control->finishes.load(), std::string(s)); }
|
|
std::shared_ptr<Control> control;
|
|
std::vector<std::pair<int, std::string>> writes;
|
|
};
|
|
for (const bool wrapped : {false, true}) {
|
|
const auto [event, result] = json_parts(stream(body("", wrapped)));
|
|
require(json_parts(stream(body(",\"stream_format\":null", wrapped))).first == event,
|
|
"a null stream_format changed the task stream's JSON events");
|
|
auto response = stream(body(sse, wrapped)); success(response);
|
|
const auto buffering = response.headers.find("X-Accel-Buffering");
|
|
require(response.content_type == "text/event-stream; charset=utf-8" && bool(response.stream_body) &&
|
|
response.body.empty() && buffering != response.headers.end() && buffering->second == "no",
|
|
"task stream with sse did not answer an event stream: " + response.content_type);
|
|
require(active() == 1, "task stream with sse did not hold its slot before the headers");
|
|
Recorder writer(c); const int before = c->finishes;
|
|
response.stream_body(writer);
|
|
require(active() == 1, "task stream callback released the slot its owner holds");
|
|
response.stream_body = {}; // transport drops the callback at exchange end
|
|
require(active() == 0, "task stream with sse leaked its slot");
|
|
require(writer.writes.size() == 3, "task stream with sse wrote " + std::to_string(writer.writes.size()) + " times");
|
|
require(writer.writes[0].second == "data: {\"type\":\"task.stream.event\",\"event\":" + event + "}\n\n",
|
|
"task stream event is not the JSON response's: " + writer.writes[0].second);
|
|
const std::string done_key = "data: {\"type\":\"task.stream.done\",\"result\":", done_end = "}\n\n";
|
|
const auto & done = writer.writes[1].second;
|
|
require(done.size() > done_key.size() + done_end.size() && done.rfind(done_key, 0) == 0 &&
|
|
done.compare(done.size() - done_end.size(), done_end.size(), done_end) == 0,
|
|
"task stream done line: " + done);
|
|
const auto streamed = done.substr(done_key.size(), done.size() - done_key.size() - done_end.size());
|
|
require(without_ttft(streamed) == without_ttft(result), "task stream result is not the JSON response's: " + streamed);
|
|
const auto parsed = parse(streamed);
|
|
const auto & artifacts = parsed.require("artifacts").as_array();
|
|
require(parsed.require("timing").require("ttft_ms").is_number() && artifacts.size() == 1 &&
|
|
artifacts[0].require("id").as_string() == "turn.reply" && artifacts[0].require("payload").as_string() == "AQD/",
|
|
"task stream done event lost the result's timing or artifacts: " + streamed);
|
|
require(writer.writes[2].second == "data: [DONE]\n\n", "task stream did not end with [DONE]");
|
|
require(writer.writes[0].first == before && writer.writes[1].first == before + 1,
|
|
"task stream wrote its event after the run was over");
|
|
}
|
|
const auto default_events = parse(stream(body("", true)).body).require("events").as_array();
|
|
require(default_events.front().find("artifacts") == nullptr, "default route leaked event artifacts");
|
|
const auto opted_events = parse(stream(body(",\"include_artifacts\":true", true)).body).require("events").as_array();
|
|
require(opted_events.front().require("artifacts").as_array().size() == 1,
|
|
"opt-in JSON route omitted event artifacts");
|
|
auto opted_sse = stream(body(sse + ",\"include_artifacts\":true", true));
|
|
Recorder opted_writer(c);
|
|
opted_sse.stream_body(opted_writer);
|
|
require(opted_writer.writes.front().second.find("\"artifacts\"") != std::string::npos,
|
|
"opt-in SSE route omitted event artifacts");
|
|
opted_sse.stream_body = {};
|
|
// Only the top-level field of this route opts in: not the Accept header,
|
|
// not a field inside `request`, not the field on /v1/tasks/run.
|
|
srv::HttpRequest accept; accept.method = "POST"; accept.path = "/v1/tasks/stream"; accept.body = body("", true);
|
|
accept.headers["content-type"] = "application/json"; accept.headers["accept"] = "text/event-stream";
|
|
(void)json_parts(f.state->handle(accept));
|
|
(void)json_parts(stream("{\"model\":\"stream\",\"request\":{\"text\":\"hi\",\"stream_format\":\"sse\"}}"));
|
|
const auto ran = post(*f.state, "/v1/tasks/run", body(sse, true)); success(ran);
|
|
require(ran.content_type == "application/json" && !ran.stream_body && parse(ran.body).find("artifacts") != nullptr,
|
|
"/v1/tasks/run read stream_format: " + ran.body);
|
|
// Errors found before the headers keep their status.
|
|
const int runs = c->runs;
|
|
for (const std::string value : {"\"audio\"", "\"SSE\"", "\"\"", "1", "true", "{}", "[]"}) {
|
|
const auto rejected = stream(body(",\"stream_format\":" + value, true));
|
|
require(rejected.status == 400 && !rejected.stream_body &&
|
|
rejected.body.find("task stream stream_format must be sse") != std::string::npos &&
|
|
rejected.body.find("invalid_request_error") != std::string::npos,
|
|
"task stream took stream_format " + value + ": HTTP " + std::to_string(rejected.status) + " " + rejected.body);
|
|
}
|
|
const auto malformed = [&](const std::string & fields) {
|
|
return stream("{\"model\":\"stream\"" + fields + ",\"request\":{\"artifacts\":{\"id\":\"x\"}}}");
|
|
};
|
|
const auto json_rejected = malformed(""), sse_rejected = malformed(sse);
|
|
require(json_rejected.status == 400 && sse_rejected.status == 400 && !sse_rejected.stream_body &&
|
|
sse_rejected.body == json_rejected.body,
|
|
"task stream with sse answered malformed artifacts differently: " + sse_rejected.body);
|
|
require(c->runs == runs && active() == 0, "a rejected task stream reached the model or kept its slot");
|
|
// The slot is bound before the headers, so a busy model is still a 503.
|
|
std::vector<srv::HttpResponse> held;
|
|
for (int i = 0; i < count; ++i) {
|
|
held.push_back(stream(body(sse, true)));
|
|
require(bool(held.back().stream_body), "task stream with sse did not stream");
|
|
}
|
|
require(active() == count, "unstarted task streams did not hold every slot");
|
|
const auto busy = stream(body(sse + ",\"busy_timeout_ms\":50", true));
|
|
require(busy.status == 503 && !busy.stream_body && busy.body.find("server_busy") != std::string::npos,
|
|
"task stream with sse on a busy model answered HTTP " + std::to_string(busy.status) + " " + busy.body);
|
|
held.clear();
|
|
require(active() == 0, "dropped task streams kept their slots");
|
|
// After the headers a failure leaves the callback for the transport's
|
|
// error event, and a write to a closed connection stops the run.
|
|
std::string error;
|
|
c->reject_run = true;
|
|
auto rejected = stream(body(sse, true)); success(rejected);
|
|
require(bool(rejected.stream_body), "task stream with sse did not stream");
|
|
Writer silent;
|
|
try { rejected.stream_body(silent); } catch (const rt::InvalidRequestError & e) { error = e.what(); }
|
|
rejected.stream_body = {};
|
|
require(error == "controlled request rejection" && silent.output.empty() && active() == 0,
|
|
"a task stream the model turned away ended with '" + error + "' after: " + silent.output);
|
|
c->reject_run = false;
|
|
auto dropped = stream(body(sse, true)); success(dropped);
|
|
require(bool(dropped.stream_body), "task stream with sse did not stream");
|
|
Writer gone; gone.disconnected = true; error.clear();
|
|
const int before = c->finishes;
|
|
try { dropped.stream_body(gone); } catch (const std::runtime_error & e) { error = e.what(); }
|
|
dropped.stream_body = {};
|
|
require(error == "client disconnected" && c->finishes == before && active() == 0,
|
|
"a task stream to a closed connection ended with '" + error + "' and kept running or kept its slot");
|
|
auto abandoned = stream(body(sse, true));
|
|
require(bool(abandoned.stream_body) && active() == 1, "unstarted task stream did not own its slot");
|
|
abandoned = {}; // Dropped without invoking the callback.
|
|
require(active() == 0, "unstarted task stream leaked its slot");
|
|
}
|
|
|
|
int main(int argc, char ** argv) {
|
|
try {
|
|
engine::io::json::enable_serialized_json_parsing();
|
|
if (argc >= 2 && std::string(argv[1]) == "stream-artifacts") {
|
|
engine::debug::configure_logging({true});
|
|
const auto started = std::chrono::steady_clock::now();
|
|
stream_artifacts_are_output();
|
|
for (int count : {1, 2}) generic_task_stream_sse(count);
|
|
engine::debug::timing_log_scalar("session.wall_ms", engine::debug::elapsed_ms(started));
|
|
std::cout << "stream artifact defaults and opt-in JSON/SSE passed\n";
|
|
return 0;
|
|
}
|
|
if (argc == 2) {
|
|
const std::string test = argv[1];
|
|
if (test == "audit-capacity") { unsupported_capacity_preserves_registration(); }
|
|
else if (test == "audit-reconfiguration") { invalid_reconfiguration_preserves_live_state(); }
|
|
else if (test == "audit-registration") { registration_is_visible_while_loading(); }
|
|
else if (test == "audit-policy") { live_policy_does_not_hold_registry_lock(); }
|
|
else { throw std::runtime_error("unknown audit case"); }
|
|
return 0;
|
|
}
|
|
invalid_reconfiguration_preserves_live_state();
|
|
registration_is_visible_while_loading();
|
|
live_policy_does_not_hold_registry_lock();
|
|
concurrent_json_parsing();
|
|
stream_artifacts_are_output();
|
|
per_model_device_selection();
|
|
unsupported_capacity_preserves_registration();
|
|
dynamic_capacity_changes_use_parallel_lifecycle();
|
|
generic_batch_single_slot_and_invalid_input();
|
|
for (int count : {1, 2, 4}) {
|
|
older_work_before_management(count, false);
|
|
older_work_before_management(count, true);
|
|
overlapping_managers(count);
|
|
bulk_releases_idle_first(count, false);
|
|
bulk_releases_idle_first(count, true);
|
|
retiring_resident(count, 1);
|
|
retiring_resident(count, 2);
|
|
deferred_stream_ownership(count, false);
|
|
deferred_stream_ownership(count, true);
|
|
deferred_batch_ownership(count);
|
|
deferred_batch_ownership(count, true);
|
|
if (count > 1) { generic_batches_hold_independent_slots(count); }
|
|
rejected_preparation_releases_lease(count);
|
|
loading_and_inference_failures_release_ownership(count);
|
|
session_request_errors(count);
|
|
abandoned_deferred_responses_release_ownership(count);
|
|
overlapping_bulk_operations(count);
|
|
bulk_publication_excludes_only_other_bulk_calls(count);
|
|
guarded_loading_does_not_block_warm_model(count);
|
|
live_speech_return_text(count);
|
|
live_speech_optional_input(count);
|
|
generic_request_artifacts(count);
|
|
generic_task_stream_sse(count);
|
|
std::cout << "PASS real handlers: slots=" << count
|
|
<< " queued unload/reconfiguration, manager order, bulk release, global lock,"
|
|
" resident limits 1/2, deferred stream/disconnect, live speech return_text and optional input,"
|
|
" request artifacts, task stream SSE\n";
|
|
}
|
|
} catch (const std::exception & e) { std::cerr << e.what() << '\n'; return 1; }
|
|
}
|