diff --git a/backend/cpp/audio-cpp/audio_io.cpp b/backend/cpp/audio-cpp/audio_io.cpp index 275523b0b..65dfba5fb 100644 --- a/backend/cpp/audio-cpp/audio_io.cpp +++ b/backend/cpp/audio-cpp/audio_io.cpp @@ -15,7 +15,17 @@ engine::runtime::AudioBuffer read_audio_file(const std::string &path) { throw ConfigError("audio-cpp: no input audio path was supplied"); } std::error_code ec; - if (!std::filesystem::exists(std::filesystem::path(path), ec)) { + const bool present = std::filesystem::exists(std::filesystem::path(path), ec); + if (ec) { + // exists() returning false with ec set does NOT mean the file is + // absent, it means the question could not be answered: most often a + // parent directory is not searchable. Reporting that as "does not + // exist" sends the operator after the file when the fault is the + // permissions on the directory above it. + throw ConfigError("audio-cpp: cannot stat input audio " + path + ": " + + ec.message()); + } + if (!present) { throw ConfigError("audio-cpp: input audio does not exist: " + path); } engine::runtime::AudioBuffer buffer; diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index 715d53a34..c582b823c 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -107,6 +107,52 @@ int parse_device_index(const std::string &value) { return static_cast(parsed); } +// Mirrors pkg/grpc/server.go's checkModelIdentity, backend/cpp/llama-cpp, +// backend/cpp/ds4, backend/cpp/privacy-filter and +// backend/python/common/model_identity.py. +// +// Why every backend has to carry this: in distributed mode a worker can recycle +// a stopped backend's gRPC port for a different model's backend, and the +// controller's liveness-only health probe cannot tell a stale cached route from +// a live one. Only the backend knows which model it actually loaded, so only +// the backend can catch it (#10952). Without this, an audio-cpp process reached +// through a stale route answers with a DIFFERENT model's VAD or diarization +// result and a 200, which is a wrong answer nobody can see. +// +// Either side empty means "skip". The request side is empty for a controller +// that predates the field; the loaded side is empty when such a controller +// performed the load. Neither can judge the other, and a false rejection is +// worse than the miss it prevents. +// +// Templated over the request type because every guarded message exposes +// modelidentity(), and one body keeps the rule identical across RPCs rather +// than letting it drift per handler. +// +// The loaded identity is read off the LoadedModel rather than a separate +// global, which is where this differs from llama-cpp. A handler holding the +// model through snapshot() is then necessarily judging against the identity +// THAT model was loaded with, and a concurrent reload cannot swap one without +// the other. +template +GStatus check_model_identity(const audiocpp_backend::LoadedModel &model, + const Request *request) { + if (request == nullptr || request->modelidentity().empty()) { + return GStatus::OK; + } + const std::string &loaded = model.identity(); + if (loaded.empty() || loaded == request->modelidentity()) { + return GStatus::OK; + } + // NOT_FOUND plus this exact sentinel is the cross-language wire contract + // the router matches on (grpcerrors.ModelMismatchSentinel). The code alone + // is not enough, since NOT_FOUND is returned for unrelated reasons + // elsewhere, so the substring "model identity mismatch" must survive + // verbatim through any edit to this message. + return GStatus(grpc::StatusCode::NOT_FOUND, + "audio-cpp: model identity mismatch: loaded \"" + loaded + + "\", requested \"" + request->modelidentity() + "\""); +} + // Maps a thrown exception onto the gRPC status the client should see. GStatus to_status(const std::exception &err) { if (dynamic_cast(&err) != nullptr) { @@ -164,8 +210,12 @@ public: const std::string path = audiocpp_backend::resolve_model_path( request->modelpath(), request->modelfile(), request->model()); + // ModelOptions.Model is the UNTRANSLATED controller-side name, and + // is what every ModelIdentity field on a request is compared + // against. Passed at construction so the model and the identity it + // was loaded with are published together. auto loaded = std::make_shared( - path, parsed.options); + path, parsed.options, request->model()); // Read everything the reply and the log line need while this scope // still owns the model. After the move, `loaded` is null and the @@ -223,6 +273,15 @@ public: return GStatus::OK; } + // Verifying this by hand: silero_vad is a SPEECH detector, so a synthetic + // stimulus does not exercise it. A 220 Hz sine, broadband noise and a + // harmonic buzz all return zero segments, correctly. Use real speech, which + // upstream bundles and which therefore needs no download: + // audio.cpp/assets/resources/sample_16k.wav (16 kHz mono, 14.07 s) + // Taking 1 s from offset 8000 samples (0.5 s in), padded with 1 s of + // silence either side, yields exactly one segment at start=0.9940 + // end=2.0460. A different excerpt of the same file shifts the start, + // because it changes where speech actually begins inside the window. GStatus VAD(ServerContext *, const backend::VADRequest *request, backend::VADResponse *response) override { try { @@ -234,6 +293,13 @@ public: return GStatus(grpc::StatusCode::FAILED_PRECONDITION, "audio-cpp: no model is loaded; call LoadModel first"); } + // Before any work: a stale route must be refused, not served. + // Cannot precede snapshot(), because the identity being compared + // against belongs to the model this call is about to use. + const GStatus identity = check_model_identity(*model, request); + if (!identity.ok()) { + return identity; + } audiocpp_backend::RequestShape shape; // Without this the `task:` model option is dead: routing is @@ -252,8 +318,9 @@ public: // because LaneEntry is immovable and C++17 elides the return. audiocpp_backend::LaneEntry lane = model->acquire(0); const auto session = - model->session_for(audiocpp_backend::Rpc::Vad, shape); - const auto result = audiocpp_backend::run_offline(session, task); + model->session_for(audiocpp_backend::Rpc::Vad, shape, lane); + const auto result = + audiocpp_backend::run_offline(session, task, lane); // VADSegment.start/end are float SECONDS, not sample indices. for (const auto &segment : result.speech_segments) { @@ -277,6 +344,10 @@ public: return GStatus(grpc::StatusCode::FAILED_PRECONDITION, "audio-cpp: no model is loaded; call LoadModel first"); } + const GStatus identity = check_model_identity(*model, request); + if (!identity.ok()) { + return identity; + } audiocpp_backend::RequestShape shape; shape.pinned_task = model->pinned_task(); @@ -290,7 +361,7 @@ public: // capability refusal already pays because session_for needs the lane. audiocpp_backend::LaneEntry lane = model->acquire(0); const auto session = - model->session_for(audiocpp_backend::Rpc::Diarize, shape); + model->session_for(audiocpp_backend::Rpc::Diarize, shape, lane); // DiarizeRequest.dst is the INPUT path, despite the field name: the // HTTP layer materialises the upload to a temp file and passes the @@ -307,15 +378,41 @@ public: // Moved, not copied: a long recording runs to tens of megabytes and // this is its only owner. task.audio_input = std::move(audio); - // Forwarded verbatim, and deliberately not validated against what - // the family can do. audio.cpp's sortformer diarizer takes its - // speaker count from the model package (the bundled variant is - // 4-speaker) and reads none of these keys, so today they have no - // effect. That is the documented LocalAI contract, not a silent - // failure: core/backend/diarization.go says outright that backends - // ignore the hints they do not act on. Forwarding them still means a - // family that does read them gets them, with no change here. The - // knobs that DO reach sortformer (speaker_threshold, + // READ THIS BEFORE WIRING THE FIRST DIARIZATION FAMILY. This + // forwarding is currently DEAD for sortformer, the only diarizer + // audio.cpp ships, and from the caller's side that is a silent + // wrong answer, not a soft hint: + // + // - backend.proto documents num_speakers as "exact speaker count + // if known (>0 forces)". A caller who asks for 2 and gets the + // model's 4 labels back with a 200 and no diagnostic has been + // given a wrong answer, so core/backend/diarization.go's + // "backends ignore what they don't act on" does not cover it. + // - sortformer takes its speaker count from the model package + // (assets.model_config.modules.num_speakers; the bundled + // variant is 4-speaker), and its postprocess config is parsed + // from runtime::SessionOptions, i.e. load time, not from + // TaskRequest.options at all. + // - worse, its unknown-option check only inspects keys prefixed + // "sortformer_diar.", so a bare "num_speakers" here is not even + // a recognised key: it is dropped without a trace. + // + // The keys are still forwarded, because a family that does read + // them then works with no change here. But the family that lands + // MUST either honour num_speakers or refuse it with + // INVALID_ARGUMENT. No refusal path is added now because there is + // no diarization family wired yet to test one against, and an + // untested refusal is its own hazard. + // + // Also dropped today, and worth revisiting at the same time: + // clustering_threshold, min_duration_on and min_duration_off. + // The two min_duration_* fields are NOT family-specific: they are + // generic post-filters over the returned turns (drop a turn shorter + // than min_duration_on, merge two turns of the same speaker + // separated by less than min_duration_off), so the handler could + // honour them in about six lines whatever the family does. + // + // The knobs that DO reach sortformer (speaker_threshold, // speaker_min_frames, speaker_pad_frames, session_len_sec) are // session options and arrive through the model YAML's // `session.:` entries at load time, not per request. @@ -332,7 +429,8 @@ public: std::to_string(request->max_speakers()); } - const auto result = audiocpp_backend::run_offline(session, task); + const auto result = + audiocpp_backend::run_offline(session, task, lane); // Emitted verbatim, in the order the model produced them. NOT // sorted, merged or de-overlapped: sortformer binarizes each speaker @@ -347,6 +445,15 @@ public: auto *out = response->add_segments(); out->set_id(id++); // Seconds, like VADSegment. Only TranscriptSegment/Word are ns. + // + // Converted against the INPUT rate, which is correct and has + // been checked against upstream rather than assumed: sortformer + // refuses any input whose rate differs from its feature config + // (frontend.cpp throws), and every TimeSpan it emits is built + // as llround(seconds * 16000.0) (postprocess.cpp). So the span + // domain and the input domain are the same 16 kHz, and this is + // not a place where a resampling family could silently scale + // every timestamp by a constant. out->set_start(audiocpp_backend::samples_to_seconds( turn.span.start_sample, sample_rate)); out->set_end(audiocpp_backend::samples_to_seconds( diff --git a/backend/cpp/audio-cpp/loaded_model.cpp b/backend/cpp/audio-cpp/loaded_model.cpp index 292863f55..ec5b0a7c3 100644 --- a/backend/cpp/audio-cpp/loaded_model.cpp +++ b/backend/cpp/audio-cpp/loaded_model.cpp @@ -276,12 +276,16 @@ std::string resolve_model_path(const std::string &model_path_dir, } LoadedModel::LoadedModel(const std::string &resolved_path, - const ModelOptions &options) - : LoadedModel(resolved_path, options, require_family(resolved_path, options)) {} + const ModelOptions &options, + std::string model_identity) + : LoadedModel(resolved_path, options, require_family(resolved_path, options), + std::move(model_identity)) {} LoadedModel::LoadedModel(const std::string &resolved_path, - const ModelOptions &options, std::string family) - : lane_(family), registry_(engine::runtime::make_default_registry()) { + const ModelOptions &options, std::string family, + std::string model_identity) + : lane_(family), registry_(engine::runtime::make_default_registry()), + identity_(std::move(model_identity)) { if (!registry_.supports_family(family)) { throw ConfigError("audio-cpp: unknown audio.cpp family '" + family + "'"); } @@ -333,7 +337,11 @@ LoadedModel::LoadedModel(const std::string &resolved_path, capabilities_ = to_capabilities(family, engine_caps); } -LoadedModel::Session LoadedModel::session_for(Rpc rpc, const RequestShape &shape) { +LoadedModel::Session LoadedModel::session_for(Rpc rpc, const RequestShape &shape, + const LaneEntry &lane) { + // Proof of holding only. Nothing here reads it, and nothing should: its + // whole job is to make a caller that has not taken the lane fail to compile. + (void)lane; const Route route = resolve_route(rpc, shape, capabilities_); if (!route.ok) { throw CapabilityError(route.error); @@ -423,8 +431,17 @@ LaneEntry LoadedModel::acquire(int requested_timeout_ms) { requested_timeout_ms)); } +std::unique_ptr LoadedModel::acquire_owned(int requested_timeout_ms) { + return std::make_unique( + lane_, + resolve_wait_budget_ms(wait_budget_ceiling_ms_, requested_timeout_ms)); +} + engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, - const engine::runtime::TaskRequest &request) { + const engine::runtime::TaskRequest &request, + const LaneEntry &lane) { + // Proof of holding only, as in session_for. + (void)lane; if (session.offline == nullptr) { throw CapabilityError("audio-cpp: no offline session for this request"); } @@ -432,10 +449,4 @@ engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, return session.offline->run(request); } -std::unique_ptr LoadedModel::acquire_owned(int requested_timeout_ms) { - return std::make_unique( - lane_, - resolve_wait_budget_ms(wait_budget_ceiling_ms_, requested_timeout_ms)); -} - } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/loaded_model.h b/backend/cpp/audio-cpp/loaded_model.h index c2e308091..9f2a4cf0c 100644 --- a/backend/cpp/audio-cpp/loaded_model.h +++ b/backend/cpp/audio-cpp/loaded_model.h @@ -66,12 +66,23 @@ public: // Throws ConfigError when the path does not exist, the family cannot be // determined, or the registry rejects the family. - LoadedModel(const std::string &resolved_path, const ModelOptions &options); + // + // `model_identity` is ModelOptions.Model verbatim: the UNTRANSLATED + // controller-side name. It is a constructor argument rather than a setter + // so identity and model are inseparable. llama-cpp keeps its equivalent in + // a separate global from the model, which leaves a window where a handler + // can read one without the other; here a handler that holds the model + // through snapshot() necessarily holds the identity it was loaded with. + LoadedModel(const std::string &resolved_path, const ModelOptions &options, + std::string model_identity); LoadedModel(const LoadedModel &) = delete; LoadedModel &operator=(const LoadedModel &) = delete; const std::string &family() const noexcept { return capabilities_.family; } + // Empty when the controller predates ModelOptions.ModelIdentity, which the + // identity check reads as "skip". See check_model_identity in grpc-server. + const std::string &identity() const noexcept { return identity_; } const std::string &variant() const noexcept { return variant_; } const std::string &description() const noexcept { return description_; } const std::vector &languages() const noexcept { return languages_; } @@ -92,10 +103,14 @@ public: // runtime_error when it can but the session could not be built, which is an // environment fault rather than a capability answer. // - // Call it only while holding this model's lane. The session cache is not - // itself synchronised, and it does not need to be: the lane admits one - // caller at a time, which is the same constraint the sessions themselves - // impose. + // The `lane` parameter is a PROOF OF HOLDING and is otherwise unused: it + // exists so the rule below is a compile error rather than prose. The + // session cache is an unsynchronised std::map and the sessions themselves + // are not reentrant, so this must only be called with the lane held; the + // lane admits one caller at a time, which is exactly the constraint the + // sessions impose. Pass the LaneEntry from acquire(). It is deliberately + // not a reference to `this`'s own lane member, because a caller can only + // produce a LaneEntry by having taken one. // // STATE CONTRACT, and it is the CALLER'S to honour. Sessions are cached per // (task, mode), so a streaming session is normally the same warm object the @@ -116,7 +131,8 @@ public: // // Offline sessions need no such care: their interface has no reset and // run() takes a whole request. - Session session_for(Rpc rpc, const RequestShape &shape); + Session session_for(Rpc rpc, const RequestShape &shape, + const LaneEntry &lane); // Takes the inference lane, or throws LaneUnavailable. Serializes runs // against this model. `requested_timeout_ms` is a per-request wait hint @@ -144,8 +160,10 @@ private: // The public constructor runs the load gate, then delegates here. The // detour exists because lane_ has to be built from the family in the member // initializer list, and the family is only known after the gate has run. + // Four parameters rather than three so it cannot be confused with the + // public constructor, whose third argument is also a std::string. LoadedModel(const std::string &resolved_path, const ModelOptions &options, - std::string family); + std::string family, std::string model_identity); // MEMBER ORDER IS LOAD-BEARING BELOW THIS LINE. Members are destroyed in // reverse declaration order. @@ -168,6 +186,7 @@ private: std::string description_; std::vector languages_; std::string pinned_task_; + std::string identity_; bool supports_timestamps_ = false; int wait_budget_ceiling_ms_ = 0; }; @@ -178,14 +197,17 @@ private: // the model: a second request with a different sample rate or length would // otherwise run against the first request's contract. // -// Call it only while holding the model's lane, for the same reason session_for -// requires it: the session is not reentrant, and prepare() mutates it. +// `lane` is a PROOF OF HOLDING, unused at runtime, for the same reason +// session_for takes one: the session is not reentrant and prepare() mutates it, +// so running without the lane is a data race. Making it a parameter turns that +// into a compile error instead of a comment. // // Throws CapabilityError when the session is not an offline one. That should be // unreachable through session_for, which already refuses a non-offline session // for an offline route, and is checked anyway because the alternative is a null // dereference. engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, - const engine::runtime::TaskRequest &request); + const engine::runtime::TaskRequest &request, + const LaneEntry &lane); } // namespace audiocpp_backend