diff --git a/backend/cpp/audio-cpp/CMakeLists.txt b/backend/cpp/audio-cpp/CMakeLists.txt index 2bb15b366..455d22ddf 100644 --- a/backend/cpp/audio-cpp/CMakeLists.txt +++ b/backend/cpp/audio-cpp/CMakeLists.txt @@ -135,6 +135,7 @@ add_executable(${TARGET} stream_delta.cpp wav_header.cpp inference_lane.cpp + live_watchdog.cpp ) # loaded_model.cpp mirrors engine::runtime::VoiceTaskKind onto its own Task enum. diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index e63dd6052..beefb42ea 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -17,6 +17,7 @@ #include "capability_routing.h" #include "generation_request.h" #include "inference_lane.h" +#include "live_watchdog.h" #include "loaded_model.h" #include "model_options.h" #include "result_map.h" @@ -466,6 +467,18 @@ void refuse_cloning_without_a_clip(const audiocpp_backend::LoadedModel &model, "named preset)"); } +// A live stream whose peer stopped sending without closing, cancelled by the +// idle watchdog. Its own type rather than a flag, so the read loop can unwind +// without the driver going on to finalize a decode nobody is waiting for. +// +// NOT handled by to_status: it is thrown and caught inside one handler, and +// giving it a clause there keeps to_status a statement about exceptions that +// cross unit boundaries. +class LiveIdleTimeout : public std::runtime_error { +public: + using std::runtime_error::runtime_error; +}; + // Maps a thrown exception onto the gRPC status the client should see. GStatus to_status(const std::exception &err) { if (dynamic_cast(&err) != nullptr) { @@ -1559,7 +1572,7 @@ public: // cuts people off mid-sentence. False is not a degradation here, it is the // truth: this backend does not know. GStatus AudioTranscriptionLive( - ServerContext *, + ServerContext *context, grpc::ServerReaderWriter *stream) override { try { @@ -1587,6 +1600,24 @@ public: } const backend::TranscriptLiveConfig config = incoming.config(); + audiocpp_backend::RequestShape shape; + // No request signal chooses the task here: TranscriptLiveConfig has + // no prompt field, so has_prompt_text stays false and the only + // candidate is Asr. pinned_task is still copied, because without it + // the model's `task:` option is dead on this RPC. + shape.pinned_task = model->pinned_task(); + + // FIRST, before the lane and BEFORE the rate check below, like every + // other handler here. A family that cannot stream ASR is answering a + // question about itself, and that answer outranks any complaint + // about the request: pkg/grpc/grpcerrors/errors.go degrades to the + // file path on UNIMPLEMENTED and on nothing else, so a live-incapable + // model asked at a wrong rate must not answer INVALID_ARGUMENT and + // cost the caller its fallback. Unreachable from LocalAI today, + // which always sends 16000, and wrong on its own terms regardless. + model->check_can_serve(audiocpp_backend::Rpc::AudioTranscriptionLive, + shape); + // 16 kHz mono or nothing, and this is a REFUSAL rather than a // resample for the reason spelled out at kSpeechSampleRate: the // streaming families build their spans in their own 16 kHz feature @@ -1598,34 +1629,23 @@ public: // resampler cannot be applied to it incrementally, so the only // honest answers are this refusal or a wrong timestamp. // + // ZERO means 16000, which is what backend.proto documents. Every + // OTHER value is refused, including a negative one: -1 is malformed + // rather than absent, and quietly handing it the default would be + // this handler inventing a request the caller did not make. + // // Costs nothing today: core/backend/transcript_live.go hardcodes - // liveSampleRate = 16000, and backend.proto documents 0 as 16000 and - // says backends may reject other rates. - const int sample_rate = - config.sample_rate() > 0 ? config.sample_rate() : kSpeechSampleRate; - if (sample_rate != kSpeechSampleRate) { + // liveSampleRate = 16000. + if (config.sample_rate() != 0 && + config.sample_rate() != kSpeechSampleRate) { return GStatus(grpc::StatusCode::INVALID_ARGUMENT, "audio-cpp: live transcription accepts " + std::to_string(kSpeechSampleRate) + - " Hz mono PCM only, got " + - std::to_string(sample_rate) + + " Hz mono PCM only (or 0 to mean it), got " + + std::to_string(config.sample_rate()) + " Hz; resample on the client side"); } - - audiocpp_backend::RequestShape shape; - // No request signal chooses the task here: TranscriptLiveConfig has - // no prompt field, so has_prompt_text stays false and the only - // candidate is Asr. pinned_task is still copied, because without it - // the model's `task:` option is dead on this RPC. - shape.pinned_task = model->pinned_task(); - - // Before the lane, like every other handler: a family that cannot - // stream ASR is answering a question about itself and must not - // queue behind somebody else's run to be told no. This is also what - // produces the UNIMPLEMENTED the client reads as "degrade to the - // file path", so it has to happen before the ack below. - model->check_can_serve(audiocpp_backend::Rpc::AudioTranscriptionLive, - shape); + const int sample_rate = kSpeechSampleRate; // THE LANE IS HELD FOR THE WHOLE STREAM, which is longer than any // other handler holds it: as long as the user keeps talking. The @@ -1718,9 +1738,43 @@ public: } }; + // ARMED ONLY NOW, which is after the lane was taken and before the + // first audio frame is read, and DISARMED when the read side closes. + // Both ends matter: + // + // - not earlier, because acquire() above can legitimately block + // for as long as another live stream is running, and cancelling + // a caller for waiting its turn would be this backend punishing + // its own queue. The read of the CONFIG is likewise unwatched: + // it holds no lane, so a client that stalls there wedges + // nothing. + // - not later, because everything past the read loop is OUR + // compute. A decode that outruns the window is not a peer that + // went quiet, and cancelling there would throw away the + // transcript the client is waiting for. + // + // The status the client actually receives is CANCELLED, not the + // DEADLINE_EXCEEDED returned below: TryCancel is the only way to + // unblock a synchronous Read, and it decides the wire status itself. + // The point of this is not the status, it is that the lane comes + // back. + audiocpp_backend::IdleWatchdog idle( + std::chrono::milliseconds(model->live_idle_timeout_ms()), + [context, &model] { + // Logged, because from the client's side this looks like an + // unexplained cancellation and there is nowhere else to say + // why. + std::cerr << "audio-cpp: live stream idle for " + << model->live_idle_timeout_ms() + << " ms with the send side still open; cancelling " + "it to release the model lane\n"; + context->TryCancel(); + }); + std::int64_t consumed_frames = 0; const auto next_frames = [&](std::vector &out) { while (stream->Read(&incoming)) { + idle.touch(); if (incoming.has_config()) { // backend.proto says a second Config resets the decode // session. audio.cpp cannot do that truthfully: a reset @@ -1752,6 +1806,21 @@ public: consumed_frames += static_cast(pcm.size()); return true; } + // The read side is closed, one way or the other. Stop the timer + // BEFORE saying so, since what follows is finalize(). + idle.disarm(); + if (idle.fired()) { + // Read returned false because we cancelled the RPC, not + // because the client closed. Thrown rather than returned so + // the driver does not go on to finalize: there is nobody + // left to send a transcript to, and the whole point was to + // stop occupying the lane. + throw LiveIdleTimeout( + "audio-cpp: no audio frame arrived within " + + std::to_string(model->live_idle_timeout_ms()) + + " ms and the client did not close the stream; cancelled " + "to release the model"); + } return false; }; @@ -1762,6 +1831,19 @@ public: // also buffers the wire's frames up to the family's own preferred // window, which is one second for nemotron_asr and nothing like the // 512-sample frames a client's audio callback produces. + // + // "LIVE" HERE MEANS INCREMENTAL INPUT, NOT LOW LATENCY, and with the + // families in the pinned checkout it does not yet mean incremental + // OUTPUT either. nemotron_asr's process_audio_chunk only appends to + // its buffer (src/models/nemotron_asr/session.cpp:424-436); the whole + // decode, and therefore every delta, happens inside finalize(). So + // against nemotron_asr every delta arrives AFTER the client closes + // its send side, and the driver's policy-window buffering changes + // nothing measurable. It is not inert for vibevoice_asr or + // higgs_audio_stt, which decode per chunk. Nobody should benchmark + // time-to-first-delta against nemotron_asr and conclude this is + // broken: it is the family, and the fix is a family that streams its + // decode. const auto result = audiocpp_backend::run_streaming_live( session, task, next_frames, emit, lane); @@ -1781,6 +1863,11 @@ public: stream->Write(final_response); } return GStatus::OK; + } catch (const LiveIdleTimeout &err) { + // For the server's own record, and for a client that somehow reads + // it: TryCancel has already decided the wire status is CANCELLED. + // The lane was released by unwinding to here, which is the point. + return GStatus(grpc::StatusCode::DEADLINE_EXCEEDED, err.what()); } catch (const std::exception &err) { return to_status(err); } diff --git a/backend/cpp/audio-cpp/live_watchdog.cpp b/backend/cpp/audio-cpp/live_watchdog.cpp new file mode 100644 index 000000000..d0d7aa001 --- /dev/null +++ b/backend/cpp/audio-cpp/live_watchdog.cpp @@ -0,0 +1,71 @@ +#include "live_watchdog.h" + +#include + +namespace audiocpp_backend { + +IdleWatchdog::IdleWatchdog(std::chrono::milliseconds window, + std::function on_idle) + : window_(window), on_idle_(std::move(on_idle)), + last_(std::chrono::steady_clock::now()) { + if (window_.count() <= 0) { + // Disabled: no thread at all, rather than a thread with an infinite + // deadline. A thread that exists is a thread that has to be joined on + // every exit path, and there is nothing for this one to do. + return; + } + thread_ = std::thread([this] { run(); }); +} + +IdleWatchdog::~IdleWatchdog() { disarm(); } + +void IdleWatchdog::touch() { + std::lock_guard lock(mu_); + last_ = std::chrono::steady_clock::now(); + // Deliberately does NOT notify. The waiter recomputes its deadline from + // last_ every time it wakes, so a touch that lands mid-window is picked up + // when the old deadline expires, and a touch is the hot path: it runs once + // per frame on the wire. +} + +void IdleWatchdog::disarm() { + { + std::lock_guard lock(mu_); + stop_ = true; + } + cv_.notify_all(); + if (thread_.joinable()) { + thread_.join(); + } +} + +bool IdleWatchdog::fired() const { + std::lock_guard lock(mu_); + return fired_; +} + +void IdleWatchdog::run() { + std::unique_lock lock(mu_); + while (!stop_) { + const auto deadline = last_ + window_; + if (cv_.wait_until(lock, deadline, [this] { return stop_; })) { + return; // disarmed + } + // The deadline passed, but last_ may have moved while this thread was + // waiting, and a condition variable may also wake spuriously. Re-read + // it: without this check a touch that landed mid-window would still be + // followed by a cancellation, i.e. a live client cut off mid-sentence. + if (std::chrono::steady_clock::now() < last_ + window_) { + continue; + } + fired_ = true; + auto callback = on_idle_; + lock.unlock(); + if (callback) { + callback(); + } + return; // one shot + } +} + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/live_watchdog.h b/backend/cpp/audio-cpp/live_watchdog.h new file mode 100644 index 000000000..37e6149bc --- /dev/null +++ b/backend/cpp/audio-cpp/live_watchdog.h @@ -0,0 +1,84 @@ +#pragma once + +// A one-shot idle timer for a bidirectional stream. Standard library only, so +// it is tested without an audio.cpp checkout or a gRPC server. +// +// WHY IT EXISTS. AudioTranscriptionLive holds the model's inference lane for the +// whole stream, because the streaming session is stateful and a concurrent run +// would interleave two callers' audio. Every other RPC in this backend holds the +// lane across COMPUTE, or across a write to a slow reader, and both of those +// terminate on their own. A live stream instead blocks in a client-driven read, +// and a peer that goes silent WITHOUT closing the stream never terminates +// anything: the lane stays taken and every other RPC against that model queues +// behind a client that has stopped speaking. A websocket death does cancel the +// RPC and free it, but "the peer's TCP connection eventually dies" is not a +// bound anyone can state, so this supplies one. +// +// HOW IT ENDS THE STREAM, and the part that is not obvious: gRPC's synchronous +// ServerReaderWriter::Read has no timeout and cannot be given one. The only way +// to unblock it from another thread is ServerContext::TryCancel, which is what +// the callback is for. That means the client sees CANCELLED rather than whatever +// status the handler goes on to return: the returned status is for the server's +// own record. Releasing the lane is the point. +// +// ONE SHOT on purpose. Once the callback has run the stream is being torn down, +// so there is nothing left to watch, and a repeating timer would call TryCancel +// on a context the handler may already have returned from. + +#include +#include +#include +#include +#include + +namespace audiocpp_backend { + +class IdleWatchdog { +public: + // A window that is not positive DISABLES the watchdog entirely: no thread is + // started and fired() never becomes true. That is the operator's escape + // hatch for a client that legitimately holds a stream open through long + // pauses, and it is why the option carrying it documents 0 as "no limit" + // rather than as "expire immediately". + // + // `on_idle` runs on the watchdog's own thread with no lock held. It must be + // safe to call while the watched thread is blocked in a read, which is the + // only reason this class exists; ServerContext::TryCancel is documented as + // exactly that. + IdleWatchdog(std::chrono::milliseconds window, std::function on_idle); + + // Joins the thread, so the callback can safely capture anything that + // outlives this object's scope and nothing else has to be reasoned about. + ~IdleWatchdog(); + + IdleWatchdog(const IdleWatchdog &) = delete; + IdleWatchdog &operator=(const IdleWatchdog &) = delete; + + // Restarts the window. Call it whenever the peer proves it is still there. + void touch(); + + // Stops watching and joins. Idempotent, and REQUIRED before any long + // non-read work the window must not cover: the caller's own decode is not + // the peer going quiet, and cancelling in the middle of it would throw away + // a transcript the client is waiting for. + void disarm(); + + // True once the window elapsed and the callback ran. Stays true after + // disarm, so the caller can tell "the peer closed" from "we cancelled it". + bool fired() const; + +private: + void run(); + + const std::chrono::milliseconds window_; + std::function on_idle_; + + mutable std::mutex mu_; + std::condition_variable cv_; + std::chrono::steady_clock::time_point last_; + bool stop_ = false; + bool fired_ = false; + std::thread thread_; +}; + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/live_watchdog_test.cpp b/backend/cpp/audio-cpp/live_watchdog_test.cpp new file mode 100644 index 000000000..bb00db9dc --- /dev/null +++ b/backend/cpp/audio-cpp/live_watchdog_test.cpp @@ -0,0 +1,159 @@ +// Unit tests for the live stream idle watchdog. Standard library only; the +// harness compiles this as a single translation unit, so the implementation is +// included directly. +// +// These are TIMING tests, which is unavoidable: what is under test is a +// deadline. Every window here is short and every assertion waits several +// multiples of it, so a loaded machine slows the test down rather than +// flipping its answer. The one thing never asserted is how SOON something +// happens, only that it eventually does or never does. + +#include "live_watchdog.cpp" + +#include +#include +#include +#include + +using namespace std::chrono_literals; + +static int failures = 0; + +static void check(bool ok, const std::string &name) { + if (!ok) { + failures++; + fprintf(stderr, "FAIL: %s\n", name.c_str()); + } else { + fprintf(stderr, "ok: %s\n", name.c_str()); + } +} + +// A peer that goes quiet without closing. This is the whole point: the lane it +// holds has to come back. +static void test_it_fires_when_nothing_touches_it() { + std::atomic calls{0}; + audiocpp_backend::IdleWatchdog watchdog(100ms, [&calls] { ++calls; }); + std::this_thread::sleep_for(600ms); + check(watchdog.fired(), "a window that elapses untouched fires"); + check(calls.load() == 1, "the callback runs exactly once, not once per window"); +} + +// A peer that is still speaking must never be cut off. Touches land at a third +// of the window, for six windows' worth of wall clock. +static void test_touching_defers_it_indefinitely() { + std::atomic calls{0}; + audiocpp_backend::IdleWatchdog watchdog(300ms, [&calls] { ++calls; }); + for (int i = 0; i < 20; ++i) { + std::this_thread::sleep_for(100ms); + watchdog.touch(); + } + check(!watchdog.fired(), + "a stream touched inside every window is never cancelled"); + check(calls.load() == 0, "no callback runs while the peer is still there"); +} + +// Disarm is what the handler calls when the read side closes, before a decode +// that can take longer than the window. Firing after that would throw away the +// transcript the client is waiting for. +static void test_disarm_stops_it_before_the_window() { + std::atomic calls{0}; + audiocpp_backend::IdleWatchdog watchdog(200ms, [&calls] { ++calls; }); + std::this_thread::sleep_for(20ms); + watchdog.disarm(); + std::this_thread::sleep_for(600ms); + check(!watchdog.fired(), "a disarmed watchdog does not fire"); + check(calls.load() == 0, "a disarmed watchdog runs no callback"); +} + +static void test_disarm_is_idempotent() { + audiocpp_backend::IdleWatchdog watchdog(50ms, [] {}); + watchdog.disarm(); + watchdog.disarm(); + watchdog.disarm(); + check(true, "disarming three times joins once and does not abort"); +} + +// The operator's escape hatch, for a client that legitimately holds a stream +// open through long pauses. Not "expire immediately", which is what a naive +// reading of a zero timeout would give. +static void test_a_non_positive_window_disables_it() { + std::atomic calls{0}; + { + audiocpp_backend::IdleWatchdog watchdog(0ms, [&calls] { ++calls; }); + std::this_thread::sleep_for(300ms); + check(!watchdog.fired(), "a zero window never fires"); + } + { + audiocpp_backend::IdleWatchdog watchdog(-5ms, [&calls] { ++calls; }); + std::this_thread::sleep_for(300ms); + check(!watchdog.fired(), "a negative window never fires"); + } + check(calls.load() == 0, "a disabled watchdog runs no callback"); +} + +// fired() has to survive the disarm, because the handler reads it AFTER the +// read loop ends to tell "the peer closed" from "we cancelled the peer", and +// those two get different statuses. +static void test_fired_survives_a_later_disarm() { + audiocpp_backend::IdleWatchdog watchdog(80ms, [] {}); + std::this_thread::sleep_for(500ms); + watchdog.disarm(); + check(watchdog.fired(), "a watchdog that fired still says so after disarm"); +} + +// The destructor joins, so a callback capturing the handler's frame cannot run +// after that frame is gone. Without the join this is a use after free that only +// shows up under load. +// +// The window is LONGER than the scope on purpose. An earlier version of this +// test slept past the window inside the scope, so the callback had already run +// by the time the object was destroyed and a destructor that DETACHED the thread +// instead of joining it passed unnoticed. Mutation testing is what found that; +// the shape below kills it, because a detached thread wakes after the object is +// gone and calls a callback that must never run. +static void test_the_destructor_joins() { + std::atomic calls{0}; + std::atomic alive{true}; + { + audiocpp_backend::IdleWatchdog watchdog(200ms, [&calls, &alive] { + check(alive.load(), + "the callback never runs after the watched scope ended"); + ++calls; + }); + std::this_thread::sleep_for(20ms); + } + alive.store(false); + std::this_thread::sleep_for(600ms); + check(calls.load() == 0, + "destruction stops the timer rather than leaving it running against a " + "dead frame"); +} + +// The other half of that pair: a callback that DOES fire inside the scope runs +// exactly once, so the test above is not passing merely because nothing ever +// fires. +static void test_a_firing_watchdog_still_joins_cleanly() { + std::atomic calls{0}; + { + audiocpp_backend::IdleWatchdog watchdog(50ms, [&calls] { ++calls; }); + std::this_thread::sleep_for(400ms); + } + check(calls.load() == 1, "the callback ran once, inside the scope"); +} + +int main() { + test_it_fires_when_nothing_touches_it(); + test_touching_defers_it_indefinitely(); + test_disarm_stops_it_before_the_window(); + test_disarm_is_idempotent(); + test_a_non_positive_window_disables_it(); + test_fired_survives_a_later_disarm(); + test_the_destructor_joins(); + test_a_firing_watchdog_still_joins_cleanly(); + if (failures) { + fprintf(stderr, "%d check(s) failed\n", failures); + return 1; + } + fprintf(stderr, "all live_watchdog checks passed\n"); + return 0; +} diff --git a/backend/cpp/audio-cpp/loaded_model.cpp b/backend/cpp/audio-cpp/loaded_model.cpp index f38a1dfc9..ba4d28ab8 100644 --- a/backend/cpp/audio-cpp/loaded_model.cpp +++ b/backend/cpp/audio-cpp/loaded_model.cpp @@ -371,6 +371,7 @@ LoadedModel::LoadedModel(const std::string &resolved_path, pinned_task_ = options.task; wait_budget_ceiling_ms_ = options.busy_timeout_ms; + live_idle_timeout_ms_ = options.live_idle_timeout_ms; engine::runtime::ModelLoadRequest request; request.model_path = std::filesystem::path(resolved_path); diff --git a/backend/cpp/audio-cpp/loaded_model.h b/backend/cpp/audio-cpp/loaded_model.h index 9222ab3fc..9714b70d6 100644 --- a/backend/cpp/audio-cpp/loaded_model.h +++ b/backend/cpp/audio-cpp/loaded_model.h @@ -99,6 +99,12 @@ public: // from the load to the request that honours it. const std::string &pinned_task() const noexcept { return pinned_task_; } + // The `live_idle_timeout_ms` option: how long AudioTranscriptionLive waits + // for the next audio frame before cancelling the stream to give this + // model's lane back. 0 means no limit. See the option in model_options.h + // for why it exists and how the default was chosen. + int live_idle_timeout_ms() const noexcept { return live_idle_timeout_ms_; } + // Throws the same CapabilityError session_for would throw when this family // cannot serve the RPC, and RETURNS THE RESOLVED ROUTE otherwise. // @@ -231,6 +237,7 @@ private: std::string identity_; bool supports_timestamps_ = false; int wait_budget_ceiling_ms_ = 0; + int live_idle_timeout_ms_ = 0; }; // Prepares and runs an offline session. prepare() is called for every run diff --git a/backend/cpp/audio-cpp/model_options.cpp b/backend/cpp/audio-cpp/model_options.cpp index 0bece33b3..7a4b8c277 100644 --- a/backend/cpp/audio-cpp/model_options.cpp +++ b/backend/cpp/audio-cpp/model_options.cpp @@ -129,13 +129,20 @@ ParsedOptions parse_model_options(const std::vector &entries) { "non-negative integer, got '" + value + "'"; return parsed; } + } else if (key == "live_idle_timeout_ms") { + if (!parse_non_negative_int(value, + parsed.options.live_idle_timeout_ms)) { + parsed.error = "audio-cpp: option 'live_idle_timeout_ms' needs a " + "non-negative integer, got '" + value + "'"; + return parsed; + } } else { // Quotes the whole entry, not just the key: an entry like ":value" // has an empty key and would otherwise leave nothing to grep for. parsed.error = "audio-cpp: unknown option key '" + entry + "'. Known keys: family, task, backend, device, " "threads, model_spec_override, busy_timeout_ms, " - "load., session."; + "live_idle_timeout_ms, load., session."; return parsed; } } diff --git a/backend/cpp/audio-cpp/model_options.h b/backend/cpp/audio-cpp/model_options.h index ba0a10e37..9cb4464f8 100644 --- a/backend/cpp/audio-cpp/model_options.h +++ b/backend/cpp/audio-cpp/model_options.h @@ -30,6 +30,25 @@ struct ModelOptions { std::string model_spec_override; // 0 disables the run guard's fail-fast, restoring an unbounded wait. int busy_timeout_ms = 0; + // How long AudioTranscriptionLive waits for the next audio frame before it + // cancels the stream and gives the model's lane back. 0 means NO LIMIT. + // + // It exists because that RPC holds the lane for the whole stream, so a peer + // that stops sending WITHOUT closing blocks every other request against this + // model for as long as its socket stays up. No other RPC can do that: they + // hold the lane across compute, which ends on its own. + // + // 30 seconds, and the number is picked from what the only in-tree client + // does. core/http/endpoints/openai/realtime.go drives a 300 ms ticker and + // feeds every tick that produced new audio while a turn is open, so 30 s of + // silence is a hundred ticks that delivered nothing: the peer is gone, or + // its socket is wedged. It is also comfortably longer than any pause a + // speaker takes mid-utterance, which is the case that must never be cut off, + // and backend.proto allows one stream to span many utterances, so a client + // that pauses for longer than this between them should raise it rather than + // discover it. Lowering it below a few seconds risks cancelling a live + // speaker; 0 turns the limit off for a client that legitimately idles. + int live_idle_timeout_ms = 30000; // `load.:` entries, prefix stripped. std::map load_options; // `session.:` entries, prefix stripped. diff --git a/backend/cpp/audio-cpp/model_options_test.cpp b/backend/cpp/audio-cpp/model_options_test.cpp index dc107d6db..c3cb2f75a 100644 --- a/backend/cpp/audio-cpp/model_options_test.cpp +++ b/backend/cpp/audio-cpp/model_options_test.cpp @@ -45,6 +45,11 @@ static void test_defaults() { check(r.options.device == 0, "device defaults to 0"); check(r.options.threads == 0, "threads defaults to 0"); check(r.options.busy_timeout_ms == 0, "busy_timeout_ms defaults to 0"); + // NOT zero, unlike every other numeric option here. A live stream holds the + // model's lane while it waits on the client, so the default has to bound + // that wait; 0 is the explicit "no limit" the operator opts into. + check(r.options.live_idle_timeout_ms == 30000, + "live_idle_timeout_ms defaults to 30000"); check(r.options.load_options.empty(), "load_options defaults empty"); check(r.options.session_options.empty(), "session_options defaults empty"); } @@ -57,6 +62,7 @@ static void test_scalar_options() { "device:1", "threads:8", "busy_timeout_ms:30000", + "live_idle_timeout_ms:5000", }); check(r.error.empty(), "scalar options parse without error"); check(r.options.family == "qwen3_tts", "family parsed"); @@ -65,6 +71,10 @@ static void test_scalar_options() { check(r.options.device == 1, "device parsed"); check(r.options.threads == 8, "threads parsed"); check(r.options.busy_timeout_ms == 30000, "busy_timeout_ms parsed"); + check(r.options.live_idle_timeout_ms == 5000, "live_idle_timeout_ms parsed"); + check(parse_model_options({"live_idle_timeout_ms:0"}).options.live_idle_timeout_ms == 0, + "an explicit 0 turns the live idle limit off rather than reverting to " + "the default"); } // Values containing colons must survive: split on the FIRST colon only. @@ -106,6 +116,10 @@ static void test_errors() { "empty session key is rejected"); check(!parse_model_options({"busy_timeout_ms:abc"}).error.empty(), "non-numeric busy_timeout_ms is rejected"); + check(!parse_model_options({"live_idle_timeout_ms:abc"}).error.empty(), + "non-numeric live_idle_timeout_ms is rejected"); + check(!parse_model_options({"live_idle_timeout_ms:-1"}).error.empty(), + "negative live_idle_timeout_ms is rejected"); check(!parse_model_options({"device:-1"}).error.empty(), "negative device is rejected"); check(!parse_model_options({"threads:x"}).error.empty(),