mirror/LocalAI
1
0
Fork 0
mirror of https://github.com/mudler/LocalAI.git synced 2026-07-30 18:09:05 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix(audio-cpp): refuse a task pin the RPC cannot serve, and stop empty frames holding the lane

Browse Source 
The model's `task:` option is copied into the request shape by all nine
handlers, which is correct, but resolve_route then replaced the RPC's candidate
list with the pin WHOLESALE and never asked whether the pin was something that
RPC routes to. One pin therefore bled across all nine surfaces, and because the
family still supported the pinned task the result was a wrong 200 rather than an
error. Reproduced live: nemotron with task:asr made Vad return 200 with zero
segments after a full ASR decode, so 14 seconds of speech was reported as
silence, and Diarize did the same; silero_vad with task:vad made
AudioTranscription return 200 with empty text and four segments whose spans were
VAD segments, which combined with response_format in {text,srt,vtt,lrc} building
the body solely from Segments[].Text yields a well formed SRT of four timed
EMPTY cues. It also contradicted the documented contract, that a family which
cannot serve a request is refused rather than rerouted.

A pin is now checked against the RPC's admissible task set before it is adopted,
and the refusal names both the pin and the RPC. The set is derived from
task_candidates with every shape flag set rather than restated, so a task added
to an RPC's candidates cannot become inadmissible by omission. Every legitimate
pin survives, and the test asserts all fifteen of them alongside the eight
crossings that must not.

The live watchdog was defeated by empty frames. idle.touch() ran on ANY message,
before the has_audio and pcm.empty() filters, so a peer writing unset-oneof or
zero-length frames faster than the window held the lane indefinitely while
feeding the decoder nothing. There is one lane per model and one model per
process, so that is a single client denying the whole backend, which is what the
watchdog exists to prevent, and the thrown text already said "no audio frame
arrived". The touch moved below the filters, which are now a named predicate so
the distinction is testable rather than a call order nobody can see.

Three comments corrected against measurement rather than reasoning:

- CMakeLists claimed zero google::protobuf:: definitions remain in the
  executable. nm -C --defined-only reports 2515, and that is expected: they are
  generated code, sentencepiece::ModelProto's own _InternalParse among them. The
  claim that holds, and the one the ABI fix is actually about, is that no
  vendored protobuf RUNTIME is linked and ParseContext::ParseMessage is
  UNDEFINED in the executable, resolving to libprotobuf.so.
- refuse_cloning_without_a_clip's "cannot misfire" paragraph had its reasoning
  backwards. Routing picks VoiceCloning as the FALLBACK when there is no clip,
  which is the case being caught; chatterbox, which ships in the gallery,
  advertises clon and no tts at all, so every voice-less request lands there.
- audio_units read "2.1 min at 96 kHz" for index 11289602, which is 1.96 min.
  2.1 min is 96 kHz's OWN first failure at 12288002. Both were remeasured and
  the note is now a per-rate table.

Assisted-by: Claude:claude-opus-5 [Claude Code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
Ettore Di Giacinto
2026-07-27 07:51:05 +00:00
committed by localai-org-maint-bot
parent e9d03c7c48
commit a5144fd958
8 changed files with 292 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
backend/cpp/audio-cpp/CMakeLists.txt
View File

@@ -105,9 +105,23 @@ target_link_libraries(hw_grpc_proto PUBLIC protobuf::libprotobuf gRPC::grpc++)
#
# "package" makes sentencepiece use the protobuf found above, which is the one
# the generated code was built against. It must be set before add_subdirectory,
# since that is when sentencepiece's own cache entry is created. After it, zero
# google::protobuf:: definitions remain in the executable, and citrinet_asr,
# which parses a SentencePiece ModelProto at load, still tokenizes correctly.
# since that is when sentencepiece's own cache entry is created.
#
# WHAT THAT BUYS IS ONE PROTOBUF RUNTIME, not an executable free of protobuf
# symbols, and the difference matters to whoever checks this next. Measured with
# nm -C --defined-only on the linked grpc-server, 2515 google::protobuf::
# symbols are still DEFINED in it, and that is what should be there: they are
# generated code, sentencepiece::ModelProto's own _InternalParse and
# CheckTypeAndMergeFrom among them, which name protobuf types in their
# signatures and are compiled into every user of a .proto. Expecting zero would
# send a reader looking for a regression that is not one.
#
# The claim that decides whether the ABI mismatch above is gone is the RUNTIME
# one, and it holds: google::protobuf::internal::ParseContext::ParseMessage is
# UNDEFINED in the executable, so every generated _InternalParse resolves it to
# libprotobuf.so at load instead of to a vendored 3.14 copy. No vendored
# protobuf-lite archive is pulled in at all, and citrinet_asr, which parses a
# SentencePiece ModelProto at load, tokenizes correctly as a result.
set(SPM_PROTOBUF_PROVIDER "package" CACHE STRING
"Make sentencepiece use the found protobuf, not its vendored 3.14 copy" FORCE)

18
backend/cpp/audio-cpp/audio_units.cpp
View File

@@ -62,11 +62,19 @@ std::int64_t seconds_to_samples(double seconds, int sample_rate) {
//
// That round trip is exact only below roughly 2^23 samples. Past that the
// float samples_to_seconds returns can no longer resolve adjacent indices
// and the trip fails whatever the rounding: measured first failures run
// from 11289602 samples (4.3 min at 44.1 kHz, 2.1 min at 96 kHz) to
// 16384001 (17 min at 16 kHz). That is a property of the float seconds API
// itself, not of the rounding here, and it is why nothing should use these
// to carry a sample-accurate position in a long recording.
// and the trip fails whatever the rounding. Both the first failing INDEX
// and the duration it stands for depend on the rate, so they are listed per
// rate rather than folded into one range; measured:
//
// 16 kHz 16384001 samples 17.1 min
// 44.1 kHz 11289602 samples 4.3 min
// 48 kHz 12288002 samples 4.3 min
// 96 kHz 12288002 samples 2.1 min
//
// The shortest recording this bites is therefore a couple of minutes of
// 96 kHz audio. It is a property of the float seconds API itself, not of
// the rounding here, and it is why nothing should use these to carry a
// sample-accurate position in a long recording.
return static_cast<std::int64_t>(std::llround(scaled));
}

59
backend/cpp/audio-cpp/capability_routing.cpp
View File

@@ -166,6 +166,35 @@ const char *const kVoiceEmbedReason =
"itself exists upstream, and TitaNet and ECAPA-TDNN exist as internal "
"conditioning encoders, but neither is registered as a loadable family";
// The tasks an RPC is ever willing to route to, independent of request shape.
//
// DERIVED from task_candidates rather than restated, so a task added to an
// RPC's candidate list cannot become inadmissible as a pin by omission. Setting
// every shape flag yields each RPC's widest list: the per-flag branches only
// reorder the same three tasks for Tts, and only ADD Alignment for
// transcription, so the union is what comes back.
std::vector<Task> admissible_tasks(Rpc rpc) {
RequestShape widest;
widest.has_voice_reference = true;
widest.has_instructions = true;
widest.has_prompt_text = true;
return task_candidates(rpc, widest);
}
std::string join_task_names(const std::vector<Task> &tasks) {
std::string out;
for (const Task task : tasks) {
if (!out.empty()) {
out += ", ";
}
out += task_name(task);
}
if (out.empty()) {
out = "nothing";
}
return out;
}
std::string join_attempts(const std::vector<Task> &tasks,
const std::vector<Mode> &modes) {
std::string out;
@@ -324,8 +353,34 @@ Route resolve_route(Rpc rpc, const RequestShape &shape,
"align, vad, diar, sep, vdes, spk";
return route;
}
// A pinned task is honoured exactly. Silently rerouting would make the
// option meaningless and hide a misconfigured model.
// A pin is honoured exactly, but ONLY on an RPC that could have routed
// to it anyway. It used to replace the candidate list wholesale for
// every RPC, and because the model's `task:` option is copied into the
// shape by all nine handlers, one pin bled across all nine surfaces and
// produced wrong 200s rather than errors: nemotron with task:asr made
// Vad return 200 with zero segments after a full ASR decode, so 14
// seconds of speech was reported as silence, and silero_vad with
// task:vad made AudioTranscription return 200 with empty text and four
// segments whose spans were VAD segments, which the srt/vtt/lrc writers
// then rendered as a well formed subtitle file of four timed EMPTY
// cues. Refusing is what the docs already promise: "if the family
// cannot serve it, the request is refused rather than rerouted".
//
// Every legitimate pin survives, because a pin only ever names the task
// its own RPC already routes to: svc is in AudioTransform's candidates,
// tts/clon/vdes in TTS's, asr in transcription's, vad and diar in
// theirs.
const std::vector<Task> admissible = admissible_tasks(rpc);
if (std::find(admissible.begin(), admissible.end(), pinned) ==
admissible.end()) {
route.error = std::string("audio-cpp: this model pins task '") +
task_name(pinned) + "', which the " + rpc_name(rpc) +
" RPC never routes to (it routes to " +
join_task_names(admissible) +
"). Remove the task option to reach this RPC, or call "
"the RPC the pinned task serves";
return route;
}
tasks = {pinned};
} else {
tasks = task_candidates(rpc, shape);

98
backend/cpp/audio-cpp/capability_routing_test.cpp
View File

@@ -240,6 +240,103 @@ static void test_pinned_task_overrides_routing() {
check(!u.ok, "a pinned but unsupported task is refused");
}
// A pin lives on the MODEL, and every one of the nine handlers copies it into
// the shape, so a pin set for one RPC arrives at all of them. It used to
// replace the candidate list wholesale, which turned the other eight into wrong
// 200s rather than errors: nemotron pinned to asr made Vad answer with zero
// segments after a full ASR decode, and silero_vad pinned to vad made
// AudioTranscription answer with empty text and four segments whose spans were
// VAD segments, which the srt/vtt/lrc writers rendered as timed EMPTY cues.
static void test_pin_must_be_admissible_for_the_rpc() {
Capabilities nemotron_asr{"nemotron_asr",
{{Task::Asr, {Mode::Offline, Mode::Streaming}},
{Task::Vad, {Mode::Offline}}}};
// The pin is legitimate on the RPC it was meant for.
RequestShape asr_pin;
asr_pin.pinned_task = "asr";
const auto transcription =
resolve_route(Rpc::AudioTranscription, asr_pin, nemotron_asr);
check(transcription.ok && transcription.task == Task::Asr,
"an admissible pin is still honoured exactly");
// ...and refused on one that never routes to it, EVEN THOUGH the family
// advertises the pinned task. That is the whole point: family support is
// not the question, RPC admissibility is.
const auto vad = resolve_route(Rpc::Vad, asr_pin, nemotron_asr);
check(!vad.ok, "an inadmissible pin is refused rather than served");
check(vad.error.find("asr") != std::string::npos,
"the refusal names the pinned task");
check(vad.error.find(rpc_name(Rpc::Vad)) != std::string::npos,
"the refusal names the RPC that cannot serve it");
Capabilities silero{"silero_vad", {{Task::Vad, {Mode::Offline}}}};
RequestShape vad_pin;
vad_pin.pinned_task = "vad";
const auto vad_ok = resolve_route(Rpc::Vad, vad_pin, silero);
check(vad_ok.ok && vad_ok.task == Task::Vad, "vad is admissible on Vad");
const auto transcribe_vad =
resolve_route(Rpc::AudioTranscription, vad_pin, silero);
check(!transcribe_vad.ok,
"a vad pin cannot make a transcription request return empty cues");
// Every pin a shipped configuration could sensibly set stays reachable on
// the RPC that serves it. This is the list the fix was checked against.
struct AdmissibleCase {
Rpc rpc;
const char *task;
};
const AdmissibleCase kAdmissible[] = {
{Rpc::AudioTransform, "svc"}, {Rpc::AudioTransform, "sep"},
{Rpc::AudioTransform, "vc"}, {Rpc::AudioTransform, "s2s"},
{Rpc::Tts, "tts"}, {Rpc::Tts, "clon"},
{Rpc::Tts, "vdes"}, {Rpc::TtsStream, "tts"},
{Rpc::AudioTranscription, "asr"},
{Rpc::AudioTranscription, "align"},
{Rpc::AudioTranscriptionStream, "asr"},
{Rpc::AudioTranscriptionLive, "asr"},
{Rpc::Vad, "vad"}, {Rpc::Diarize, "diar"},
{Rpc::SoundGeneration, "gen"},
};
for (const auto &entry : kAdmissible) {
Task task = Task::Tts;
check(parse_task_name(entry.task, task),
std::string("known task name: ") + entry.task);
// A family that advertises the pinned task offline and nothing else, so
// the ONLY thing that can refuse the route is the admissibility check.
Capabilities only{"probe",
{{task, {Mode::Offline, Mode::Streaming}}}};
RequestShape pin;
pin.pinned_task = entry.task;
const auto route = resolve_route(entry.rpc, pin, only);
check(route.ok && route.task == task,
std::string("pin '") + entry.task + "' stays admissible on " +
rpc_name(entry.rpc));
}
// And the pins that must NOT cross over, one per RPC pair that was
// observed producing a wrong 200.
const AdmissibleCase kInadmissible[] = {
{Rpc::Vad, "asr"}, {Rpc::Diarize, "asr"},
{Rpc::AudioTranscription, "vad"}, {Rpc::AudioTranscription, "diar"},
{Rpc::Tts, "asr"}, {Rpc::Vad, "tts"},
{Rpc::SoundGeneration, "tts"}, {Rpc::AudioTransform, "asr"},
};
for (const auto &entry : kInadmissible) {
Task task = Task::Tts;
check(parse_task_name(entry.task, task),
std::string("known task name: ") + entry.task);
Capabilities only{"probe",
{{task, {Mode::Offline, Mode::Streaming}}}};
RequestShape pin;
pin.pinned_task = entry.task;
const auto route = resolve_route(entry.rpc, pin, only);
check(!route.ok,
std::string("pin '") + entry.task + "' is refused on " +
rpc_name(entry.rpc));
}
}
static void test_vad_and_diarize() {
Capabilities silero{"silero_vad", {{Task::Vad, {Mode::Offline, Mode::Streaming}}}};
const auto v = resolve_route(Rpc::Vad, RequestShape{}, silero);
@@ -484,6 +581,7 @@ int main() {
test_prompt_does_not_hijack_asr();
test_audio_transform_prefers_separation();
test_pinned_task_overrides_routing();
test_pin_must_be_admissible_for_the_rpc();
test_vad_and_diarize();
test_sound_generation();
test_names_round_trip();

33
backend/cpp/audio-cpp/grpc-server.cpp
View File

@@ -496,10 +496,17 @@ float audio_duration_seconds(const engine::runtime::AudioBuffer &audio) {
// chatterbox advertises clon and no tts at all, so before this every preset-only
// or voice-less request to it got that engine-internal message.
//
// It cannot misfire on the legitimate case: has_voice_reference is what made
// routing choose VoiceCloning in the first place, so reaching here with a clip
// present is impossible unless the model pinned task:clon, and a clon pin with
// no clip is exactly the misconfiguration worth naming.
// The case it catches is the FALLBACK one, which is worth stating the right way
// round. A clip is what makes routing put VoiceCloning first, but VoiceCloning
// is also the last candidate a clip-less TTS request falls back to
// (task_candidates returns {Tts, VoiceCloning, VoiceDesign} when no clip and no
// instructions are supplied), so a family that advertises clon and no tts at
// all - chatterbox, which is what ships in the gallery - routes EVERY voice-less
// request here. A task:clon pin arrives here the same way.
//
// It cannot misfire on the legitimate case for the opposite reason: a request
// that did supply a clip has voice_is_file set and returns on the line above
// before anything is thrown.
//
// Deliberately NOT generalised to "every task whose family declares
// supports_speaker_reference". That flag lives on the engine's CapabilitySet, is
@@ -1827,7 +1834,6 @@ public:
std::int64_t consumed_frames = 0;
const auto next_frames = [&](std::vector<float> &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
@@ -1844,15 +1850,20 @@ public:
"' cannot reset a live session mid-stream; close "
"this stream and open a new one");
}
if (!incoming.has_audio()) {
// Neither arm of the oneof is set. Nothing to feed, and
// not worth failing a live session over.
continue;
}
// Touched only for a frame the decoder can actually consume,
// and touched AFTER the filters rather than before them.
// Neither arm of the oneof set, or an empty pcm field, is
// nothing to feed and not worth failing a live session over
// - but it is also not the peer proving it is still there,
// and resetting the window for it let one client hold the
// model's only lane indefinitely by writing empty frames
// faster than the window. See live_frame_carries_audio.
const auto &pcm = incoming.audio().pcm();
if (pcm.empty()) {
if (!audiocpp_backend::live_frame_carries_audio(
incoming.has_audio(), pcm.empty())) {
continue;
}
idle.touch();
out.assign(pcm.begin(), pcm.end());
// Mono, so floats and frames are the same count. Only used
// for the duration reported in final_result.

4
backend/cpp/audio-cpp/live_watchdog.cpp
View File

@@ -68,4 +68,8 @@ void IdleWatchdog::run() {
}
}
bool live_frame_carries_audio(bool has_audio, bool pcm_empty) {
return has_audio && !pcm_empty;
}
} // namespace audiocpp_backend

15
backend/cpp/audio-cpp/live_watchdog.h
View File

@@ -81,4 +81,19 @@ private:
std::thread thread_;
};
// Whether one message read off a live stream is a frame the decoder can
// actually consume, which is the ONLY thing that counts as the peer proving it
// is still there.
//
// Split out of the read loop so the distinction is testable, and because
// getting it wrong is silent. The loop used to touch the watchdog on ANY
// message, before it filtered on has_audio and on an empty pcm field, so a peer
// writing unset-oneof or zero-length frames faster than the window held the
// lane forever: no audio was ever fed, no work was ever done, and the timer
// that exists to break exactly that grip was reset by the frames doing it.
// There is one lane per model and one model per process, so that is a single
// client denying the whole backend. The thrown message already said "no audio
// frame arrived"; this is the code agreeing with it.
bool live_frame_carries_audio(bool has_audio, bool pcm_empty);
} // namespace audiocpp_backend

66
backend/cpp/audio-cpp/live_watchdog_test.cpp
View File

@@ -141,6 +141,70 @@ static void test_a_firing_watchdog_still_joins_cleanly() {
check(calls.load() == 1, "the callback ran once, inside the scope");
}
// The predicate the live read loop filters on.
static void test_only_a_frame_with_audio_counts() {
using audiocpp_backend::live_frame_carries_audio;
check(live_frame_carries_audio(true, false),
"a frame with a non-empty pcm field carries audio");
check(!live_frame_carries_audio(true, true),
"an empty pcm field does not");
check(!live_frame_carries_audio(false, false),
"an unset audio oneof does not, whatever the pcm field looks like");
check(!live_frame_carries_audio(false, true),
"and neither does an unset oneof with an empty pcm field");
}
// The defect this closes, expressed as behaviour rather than as a call order:
// a peer writing frames the decoder cannot consume, faster than the window,
// used to hold the model's only inference lane forever, because the read loop
// touched the watchdog before it filtered them out. One lane per model and one
// model per process, so that is a single client denying the whole backend,
// which is exactly what the watchdog exists to prevent.
static void test_empty_frames_do_not_hold_the_lane() {
std::atomic<int> cancels{0};
std::atomic<bool> stop{false};
audiocpp_backend::IdleWatchdog watchdog(80ms, [&cancels] { ++cancels; });
// The read loop with the real filter in it: frames arrive continuously,
// none of them carries audio, and only a frame that does may touch.
std::thread peer([&] {
while (!stop.load()) {
if (audiocpp_backend::live_frame_carries_audio(false, true)) {
watchdog.touch();
}
std::this_thread::sleep_for(5ms);
}
});
std::this_thread::sleep_for(600ms);
stop.store(true);
peer.join();
watchdog.disarm();
check(cancels.load() == 1,
"a flood of frames with no audio in them still releases the lane");
// The mirror image, so this cannot pass merely because the watchdog always
// fires: a peer that keeps sending audio is left alone, exactly as before.
std::atomic<int> live_cancels{0};
std::atomic<bool> live_stop{false};
audiocpp_backend::IdleWatchdog live(80ms, [&live_cancels] { ++live_cancels; });
std::thread speaker([&] {
while (!live_stop.load()) {
if (audiocpp_backend::live_frame_carries_audio(true, false)) {
live.touch();
}
std::this_thread::sleep_for(5ms);
}
});
std::this_thread::sleep_for(600ms);
live_stop.store(true);
speaker.join();
live.disarm();
check(live_cancels.load() == 0,
"a peer that keeps sending audio is never cancelled");
}
int main() {
test_it_fires_when_nothing_touches_it();
test_touching_defers_it_indefinitely();
@@ -150,6 +214,8 @@ int main() {
test_fired_survives_a_later_disarm();
test_the_destructor_joins();
test_a_firing_watchdog_still_joins_cleanly();
test_only_a_frame_with_audio_counts();
test_empty_frames_do_not_hold_the_lane();
if (failures) {
fprintf(stderr, "%d check(s) failed\n", failures);
return 1;