1use crate::errors::AppError;
40use crate::extract::llm_embedding::LlmEmbedding;
41use parking_lot::Mutex;
42use std::path::Path;
43use std::sync::Arc;
44use std::sync::OnceLock;
45use tokio::sync::{mpsc, Semaphore};
46use tokio::task::JoinSet;
47use tokio_util::sync::CancellationToken;
48
49static CLAUDE_EMBEDDER: OnceLock<Mutex<LlmEmbedding>> = OnceLock::new();
59static OPENCODE_EMBEDDER: OnceLock<Mutex<LlmEmbedding>> = OnceLock::new();
60static OPENROUTER_CLIENT: OnceLock<crate::embedding_api::OpenRouterClient> = OnceLock::new();
61
62pub fn is_openrouter_initialized() -> bool {
64 OPENROUTER_CLIENT.get().is_some()
65}
66static EMBEDDER: OnceLock<Mutex<LlmEmbedding>> = OnceLock::new();
67
68static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
74
75pub const CHUNK_EMBED_BATCH_SIZE: usize = 8;
79
80pub const ENTITY_EMBED_BATCH_SIZE: usize = 25;
84
85pub const EMBED_BATCH_CALIBRATION_DIM: usize = 64;
87
88fn adaptive_batch_for_dim(base: usize, dim: usize) -> usize {
96 let base = base.max(1);
97 (base * EMBED_BATCH_CALIBRATION_DIM / dim.max(1)).clamp(1, base)
98}
99
100pub fn chunk_embed_batch_size() -> usize {
102 let dim = crate::constants::embedding_dim();
103 let batch = adaptive_batch_for_dim(CHUNK_EMBED_BATCH_SIZE, dim);
104 tracing::debug!(
105 dim,
106 base = CHUNK_EMBED_BATCH_SIZE,
107 batch,
108 "adaptive chunk batch size (G44)"
109 );
110 batch
111}
112
113pub fn entity_embed_batch_size() -> usize {
115 let dim = crate::constants::embedding_dim();
116 let batch = adaptive_batch_for_dim(ENTITY_EMBED_BATCH_SIZE, dim);
117 tracing::debug!(
118 dim,
119 base = ENTITY_EMBED_BATCH_SIZE,
120 batch,
121 "adaptive entity batch size (G44)"
122 );
123 batch
124}
125
126pub(crate) fn shared_runtime() -> Result<&'static tokio::runtime::Runtime, AppError> {
128 if let Some(rt) = RUNTIME.get() {
129 return Ok(rt);
130 }
131 let rt = tokio::runtime::Builder::new_multi_thread()
132 .worker_threads(2)
133 .enable_all()
134 .build()
135 .map_err(|e| AppError::Embedding(format!("tokio runtime init failed: {e}")))?;
136 let _ = RUNTIME.set(rt);
137 Ok(RUNTIME.get().expect("RUNTIME initialised above"))
138}
139
140pub fn get_embedder(_models_dir: &Path) -> Result<&'static Mutex<LlmEmbedding>, AppError> {
142 if let Some(e) = EMBEDDER.get() {
143 return Ok(e);
144 }
145 let backend = LlmEmbedding::detect_available()?;
146 let _ = EMBEDDER.set(Mutex::new(backend));
147 Ok(EMBEDDER.get().expect("EMBEDDER initialised above"))
148}
149
150pub fn get_claude_embedder(
155 claude_binary: Option<&Path>,
156 claude_model: Option<&str>,
157) -> Result<&'static Mutex<LlmEmbedding>, AppError> {
158 if let Some(e) = CLAUDE_EMBEDDER.get() {
159 return Ok(e);
160 }
161 let mut builder = LlmEmbedding::with_claude_builder();
162 if let Some(b) = claude_binary {
163 builder = builder.override_binary(b.to_path_buf());
164 }
165 if let Some(m) = claude_model {
166 builder = builder.override_model(m.to_string());
167 }
168 let backend = builder.build()?;
169 let _ = CLAUDE_EMBEDDER.set(Mutex::new(backend));
170 Ok(CLAUDE_EMBEDDER
171 .get()
172 .expect("CLAUDE_EMBEDDER initialised above"))
173}
174
175pub fn get_opencode_embedder(
180 opencode_binary: Option<&Path>,
181 opencode_model: Option<&str>,
182) -> Result<&'static Mutex<LlmEmbedding>, AppError> {
183 if let Some(e) = OPENCODE_EMBEDDER.get() {
184 return Ok(e);
185 }
186 let mut builder = LlmEmbedding::with_opencode_builder();
187 if let Some(b) = opencode_binary {
188 builder = builder.override_binary(b.to_path_buf());
189 }
190 if let Some(m) = opencode_model {
191 builder = builder.override_model(m.to_string());
192 }
193 let backend = builder.build()?;
194 let _ = OPENCODE_EMBEDDER.set(Mutex::new(backend));
195 Ok(OPENCODE_EMBEDDER
196 .get()
197 .expect("OPENCODE_EMBEDDER initialised above"))
198}
199
200pub fn get_openrouter_embedder(
201 api_key: secrecy::SecretBox<String>,
202 model: &str,
203 dim: usize,
204) -> Result<&'static crate::embedding_api::OpenRouterClient, AppError> {
205 if let Some(c) = OPENROUTER_CLIENT.get() {
206 return Ok(c);
207 }
208 let client = crate::embedding_api::OpenRouterClient::new(api_key, model.to_string(), dim)?;
209 let _ = OPENROUTER_CLIENT.set(client);
210 Ok(OPENROUTER_CLIENT
211 .get()
212 .expect("OPENROUTER_CLIENT initialised above"))
213}
214
215pub fn embed_via_claude_local(
219 _models_dir: &Path,
220 text: &str,
221 claude_binary: Option<&Path>,
222 claude_model: Option<&str>,
223) -> Result<Vec<f32>, AppError> {
224 let _slot_guard = acquire_llm_slot_for_embedding()?;
225 let embedder = get_claude_embedder(claude_binary, claude_model)?;
226 embed_passage(embedder, text)
227}
228
229pub fn embed_via_claude_local_resolved(
234 _models_dir: &Path,
235 text: &str,
236 claude_binary: Option<&Path>,
237 claude_model: Option<&str>,
238) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
239 let _slot_guard = acquire_llm_slot_for_embedding()?;
240 let embedder = get_claude_embedder(claude_binary, claude_model)?;
241 let v = embed_passage(embedder, text)?;
242 Ok((v, LlmBackendKind::Claude))
243}
244
245pub fn embed_via_opencode_local_resolved(
250 _models_dir: &Path,
251 text: &str,
252 opencode_binary: Option<&Path>,
253 opencode_model: Option<&str>,
254) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
255 let _slot_guard = acquire_llm_slot_for_embedding()?;
256 let embedder = get_opencode_embedder(opencode_binary, opencode_model)?;
257 let v = embed_passage(embedder, text)?;
258 Ok((v, LlmBackendKind::Opencode))
259}
260fn clone_client(embedder: &Mutex<LlmEmbedding>) -> LlmEmbedding {
263 embedder.lock().clone()
264}
265
266pub fn embed_passage(embedder: &Mutex<LlmEmbedding>, text: &str) -> Result<Vec<f32>, AppError> {
270 let client = clone_client(embedder);
271 let result = client.embed_passage(text)?;
272 validate_dim(result)
273}
274
275pub fn embed_query(embedder: &Mutex<LlmEmbedding>, text: &str) -> Result<Vec<f32>, AppError> {
279 let client = clone_client(embedder);
280 let result = client.embed_query(text)?;
281 validate_dim(result)
282}
283
284pub fn embed_passages_controlled(
289 embedder: &Mutex<LlmEmbedding>,
290 texts: &[&str],
291 _token_counts: &[usize],
292) -> Result<Vec<Vec<f32>>, AppError> {
293 if texts.is_empty() {
294 return Ok(Vec::new());
295 }
296 let owned: Vec<String> = texts.iter().map(|t| t.to_string()).collect();
297 embed_texts_parallel(embedder, &owned, 1, chunk_embed_batch_size())
298}
299
300pub fn embed_passage_local(models_dir: &Path, text: &str) -> Result<Vec<f32>, AppError> {
301 let _slot_guard = acquire_llm_slot_for_embedding()?;
302 let embedder = get_embedder(models_dir)?;
303 embed_passage(embedder, text)
304}
305
306pub fn should_skip_embedding_on_failure() -> bool {
310 matches!(
311 std::env::var("SQLITE_GRAPHRAG_SKIP_EMBEDDING_ON_FAILURE").as_deref(),
312 Ok("1") | Ok("true")
313 )
314}
315
316pub fn embed_passage_or_skip(
323 models_dir: &Path,
324 text: &str,
325 choice: Option<crate::cli::LlmBackendChoice>,
326) -> Result<Option<Vec<f32>>, AppError> {
327 match embed_passage_with_choice(models_dir, text, choice) {
328 Ok((v, _backend)) => Ok(Some(v)),
329 Err(AppError::Validation(msg)) => Err(AppError::Validation(msg)),
330 Err(e) => {
331 if should_skip_embedding_on_failure() {
332 tracing::warn!(
333 error = %e,
334 "embedding failed but --skip-embedding-on-failure is active; persisting with NULL embedding"
335 );
336 Ok(None)
337 } else {
338 Err(e)
339 }
340 }
341 }
342}
343
344pub fn embed_passage_local_resolved(
350 models_dir: &Path,
351 text: &str,
352) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
353 let _slot_guard = acquire_llm_slot_for_embedding()?;
354 let embedder = get_embedder(models_dir)?;
355 let v = embed_passage(embedder, text)?;
356 let kind = match embedder.lock().flavour() {
357 crate::extract::llm_embedding::EmbeddingFlavour::Codex => LlmBackendKind::Codex,
358 crate::extract::llm_embedding::EmbeddingFlavour::Claude => LlmBackendKind::Claude,
359 crate::extract::llm_embedding::EmbeddingFlavour::Opencode => LlmBackendKind::Opencode,
360 };
361 Ok((v, kind))
362}
363
364pub fn embed_query_local(models_dir: &Path, text: &str) -> Result<Vec<f32>, AppError> {
365 let _slot_guard = acquire_llm_slot_for_embedding()?;
366 let embedder = get_embedder(models_dir)?;
367 embed_query(embedder, text)
368}
369
370pub fn embed_passage_with_choice(
387 models_dir: &Path,
388 text: &str,
389 choice: Option<crate::cli::LlmBackendChoice>,
390) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
391 let _slot_guard = acquire_llm_slot_for_embedding()?;
392 match choice {
393 None => {
394 let embedder = get_embedder(models_dir)?;
395 embed_passage(embedder, text).map(|v| (v, LlmBackendKind::None))
396 }
397 Some(choice) => embed_with_fallback(models_dir, text, &choice.to_chain(), false),
398 }
399}
400
401pub fn embed_passage_with_embedding_choice(
405 models_dir: &Path,
406 text: &str,
407 embedding_backend: crate::cli::EmbeddingBackendChoice,
408 llm_backend: crate::cli::LlmBackendChoice,
409) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
410 let _slot_guard = acquire_llm_slot_for_embedding()?;
411 let chain = embedding_backend.to_chain(llm_backend);
412 embed_with_fallback(models_dir, text, &chain, false)
413}
414
415pub fn try_embed_query_with_choice(
421 models_dir: &Path,
422 text: &str,
423 choice: Option<crate::cli::LlmBackendChoice>,
424) -> Result<(Vec<f32>, LlmBackendKind), FallbackReason> {
425 match embed_passage_with_choice(models_dir, text, choice) {
426 Ok((v, _backend)) if v.is_empty() => Err(FallbackReason::DimZero),
439 Ok((v, backend)) => Ok((v, backend)),
440 Err(e) => Err(classify_embedding_error(e)),
441 }
442}
443pub fn try_embed_query_with_embedding_choice(
448 models_dir: &Path,
449 text: &str,
450 embedding_backend: crate::cli::EmbeddingBackendChoice,
451 llm_backend: crate::cli::LlmBackendChoice,
452) -> Result<(Vec<f32>, LlmBackendKind), FallbackReason> {
453 match embed_passage_with_embedding_choice(models_dir, text, embedding_backend, llm_backend) {
454 Ok((v, _backend)) if v.is_empty() => Err(FallbackReason::DimZero),
455 Ok((v, backend)) => Ok((v, backend)),
456 Err(e) => Err(classify_embedding_error(e)),
457 }
458}
459
460fn acquire_llm_slot_for_embedding() -> Result<crate::llm_slots::LlmSlotGuard, AppError> {
472 use crate::constants::{CLI_LOCK_DEFAULT_WAIT_SECS, LLM_WORKER_RSS_MB};
473 let max = std::env::var("SQLITE_GRAPHRAG_LLM_MAX_HOST_CONCURRENCY")
474 .ok()
475 .and_then(|s| s.parse::<u32>().ok())
476 .filter(|n| *n >= 1)
477 .unwrap_or_else(crate::llm_slots::default_max_concurrency);
478 let wait_secs = if std::env::var("SQLITE_GRAPHRAG_LLM_SLOT_NO_WAIT").is_ok() {
479 0
480 } else {
481 std::env::var("SQLITE_GRAPHRAG_LLM_SLOT_WAIT_SECS")
482 .ok()
483 .and_then(|s| s.parse::<u64>().ok())
484 .unwrap_or(CLI_LOCK_DEFAULT_WAIT_SECS)
485 };
486 let _ = LLM_WORKER_RSS_MB; match crate::llm_slots::acquire_llm_slot(max, wait_secs) {
494 Ok(guard) => Ok(guard),
495 Err(e @ AppError::LockBusy { .. }) if wait_secs > 0 => Err(AppError::Embedding(format!(
496 "slot exhausted: {e} (fall back to FTS5)"
497 ))),
498 Err(e) => Err(e),
499 }
500}
501#[derive(Debug, Clone, Copy, PartialEq, Eq)]
513pub enum EmbeddingErrorKind {
514 OAuth,
516 Quota,
518 SlotExhausted,
520 BackendMismatch,
522 ZeroDimension,
524 Unknown,
526}
527
528impl EmbeddingErrorKind {
529 pub fn classify(msg: &str) -> Self {
538 let m = msg.to_lowercase();
539 if m.contains("oauth") {
540 Self::OAuth
541 } else if m.contains("quota") {
542 Self::Quota
543 } else if m.contains("slot exhausted") {
544 Self::SlotExhausted
545 } else if m.contains("backend mismatch") {
546 Self::BackendMismatch
547 } else if m.contains("dim") && m.contains("zero") {
548 Self::ZeroDimension
549 } else {
550 Self::Unknown
551 }
552 }
553
554 pub fn code(&self) -> &'static str {
556 match self {
557 Self::OAuth => "oauth",
558 Self::Quota => "quota",
559 Self::SlotExhausted => "slot-exhausted",
560 Self::BackendMismatch => "backend-mismatch",
561 Self::ZeroDimension => "zero-dimension",
562 Self::Unknown => "unknown",
563 }
564 }
565}
566
567impl std::fmt::Display for EmbeddingErrorKind {
568 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
569 f.write_str(self.code())
570 }
571}
572
573#[derive(Debug, Clone, PartialEq)]
580pub enum FallbackReason {
581 EmbeddingFailed(String),
585 SlotExhausted,
590 OAuthQuota { backend: &'static str },
594 BackendMismatch {
598 requested: &'static str,
599 resolved: &'static str,
600 },
601 DimZero,
606 Cancelled,
608 Timeout {
611 operation: String,
612 duration_secs: u64,
613 },
614}
615
616impl FallbackReason {
617 pub fn reason_code(&self) -> &'static str {
621 match self {
622 Self::EmbeddingFailed(_) => "embedding_failed",
623 Self::SlotExhausted => "slot_exhausted",
624 Self::OAuthQuota { .. } => "oauth_quota",
625 Self::BackendMismatch { .. } => "backend_mismatch",
626 Self::DimZero => "dim_zero",
627 Self::Cancelled => "cancelled",
628 Self::Timeout { .. } => "timeout",
629 }
630 }
631}
632
633impl std::fmt::Display for FallbackReason {
634 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
635 match self {
636 Self::EmbeddingFailed(msg) => write!(f, "embedding failed: {msg}"),
637 Self::SlotExhausted => write!(
638 f,
639 "slot exhausted: failed to acquire LLM slot after backoff window (max=8 concurrent, total backoff=750ms)"
640 ),
641 Self::OAuthQuota { backend } => {
642 write!(f, "OAuth usage quota exhausted on backend '{backend}'")
643 }
644 Self::BackendMismatch {
645 requested,
646 resolved,
647 } => {
648 write!(
649 f,
650 "backend mismatch: user requested '{requested}' but '{resolved}' was invoked"
651 )
652 }
653 Self::DimZero => write!(f, "embedding returned zero-dimensional vector"),
654 Self::Cancelled => write!(f, "embedding cancelled by external signal"),
655 Self::Timeout {
656 operation,
657 duration_secs,
658 } => {
659 write!(
660 f,
661 "embedding timed out after {duration_secs}s during {operation}"
662 )
663 }
664 }
665 }
666}
667
668impl std::error::Error for FallbackReason {}
669
670pub fn try_embed_query_with_fallback(
678 models_dir: &Path,
679 query: &str,
680) -> Result<(Vec<f32>, LlmBackendKind), FallbackReason> {
681 match embed_query_local(models_dir, query) {
682 Ok(v) => Ok((v, LlmBackendKind::None)),
683 Err(e) => Err(classify_embedding_error(e)),
684 }
685}
686
687pub fn try_embed_query_with_deterministic_fallback(
696 models_dir: &Path,
697 query: &str,
698 choice: Option<crate::cli::LlmBackendChoice>,
699) -> Result<(Vec<f32>, LlmBackendKind), FallbackReason> {
700 match try_embed_query_with_choice(models_dir, query, choice) {
701 Ok(t) => Ok(t),
702 Err(reason @ FallbackReason::OAuthQuota { backend }) => {
703 let alt = match backend {
704 "codex" => Some(crate::cli::LlmBackendChoice::Claude),
705 "claude" => Some(crate::cli::LlmBackendChoice::Codex),
706 "opencode" => Some(crate::cli::LlmBackendChoice::Codex),
707 "openrouter" => Some(crate::cli::LlmBackendChoice::Codex),
708 _ => None,
709 };
710 if let Some(alt_choice) = alt {
711 try_embed_query_with_choice(models_dir, query, Some(alt_choice))
712 } else {
713 Err(reason)
714 }
715 }
716 Err(reason @ FallbackReason::SlotExhausted) => {
717 std::thread::sleep(std::time::Duration::from_millis(750));
718 try_embed_query_with_choice(models_dir, query, choice).or(Err(reason))
719 }
720 Err(other) => Err(other),
721 }
722}
723
724pub fn classify_embedding_error(err: AppError) -> FallbackReason {
732 match err {
733 AppError::Timeout {
734 operation,
735 duration_secs,
736 } => FallbackReason::Timeout {
737 operation,
738 duration_secs,
739 },
740 AppError::Embedding(msg) => match EmbeddingErrorKind::classify(&msg) {
741 EmbeddingErrorKind::SlotExhausted => FallbackReason::SlotExhausted,
750 EmbeddingErrorKind::OAuth => {
751 let backend = if msg.contains("codex") {
752 "codex"
753 } else if msg.contains("claude") || msg.contains("anthropic-ratelimit") {
754 "claude"
759 } else if msg.contains("opencode") {
760 "opencode"
761 } else {
762 "unknown"
763 };
764 FallbackReason::OAuthQuota { backend }
765 }
766 EmbeddingErrorKind::Quota => {
767 let backend = if msg.contains("codex") {
768 "codex"
769 } else if msg.contains("claude") || msg.contains("anthropic-ratelimit") {
770 "claude"
771 } else if msg.contains("opencode") {
772 "opencode"
773 } else {
774 "unknown"
775 };
776 FallbackReason::OAuthQuota { backend }
777 }
778 EmbeddingErrorKind::BackendMismatch => {
779 let (requested, resolved) =
784 if msg.contains("requested claude") && msg.contains("but codex") {
785 ("claude", "codex")
786 } else if msg.contains("requested codex") && msg.contains("but claude") {
787 ("codex", "claude")
788 } else if msg.contains("requested claude") {
789 ("claude", "unknown")
790 } else if msg.contains("requested codex") {
791 ("codex", "unknown")
792 } else {
793 ("unknown", "unknown")
794 };
795 FallbackReason::BackendMismatch {
796 requested,
797 resolved,
798 }
799 }
800 EmbeddingErrorKind::ZeroDimension => FallbackReason::DimZero,
801 EmbeddingErrorKind::Unknown => {
802 if msg.contains("cancelled") {
803 FallbackReason::Cancelled
804 } else {
805 FallbackReason::EmbeddingFailed(msg)
806 }
807 }
808 },
809 e => FallbackReason::EmbeddingFailed(e.to_string()),
810 }
811}
812pub fn embed_with_fallback(
831 models_dir: &Path,
832 text: &str,
833 chain: &[LlmBackendKind],
834 skip_on_failure: bool,
835) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
836 use crate::llm::exit_code_hints::LlmBackendError;
837 let effective: Vec<LlmBackendKind> = if chain.is_empty() {
838 vec![
839 LlmBackendKind::Codex,
840 LlmBackendKind::Claude,
841 LlmBackendKind::Opencode,
842 LlmBackendKind::None,
843 ]
844 } else {
845 chain.to_vec()
846 };
847
848 let mut last_err: Option<AppError> = None;
849 for backend in &effective {
850 match embed_via_backend_strict(
861 models_dir,
862 text,
863 backend,
864 last_err.as_ref(),
865 skip_on_failure,
866 ) {
867 Ok((v, resolved_kind)) => return Ok((v, resolved_kind)),
868 Err(e) => {
869 if matches!(e, AppError::Validation(_)) {
874 return Err(e);
875 }
876 tracing::warn!(
877 target: "embedding",
878 backend = ?backend,
879 error = %e,
880 "embed_with_fallback: backend failed, trying next"
881 );
882 last_err = Some(e);
883 }
884 }
885 }
886 if skip_on_failure {
887 return Ok((Vec::new(), LlmBackendKind::None));
892 }
893 Err(last_err
894 .unwrap_or_else(|| AppError::Embedding(LlmBackendError::NoBackendsAvailable.to_string())))
895}
896
897#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
901pub enum LlmBackendKind {
902 Codex,
904 Claude,
906 Opencode,
908 OpenRouter,
910 None,
912}
913
914impl LlmBackendKind {
915 pub fn as_str(self) -> &'static str {
918 match self {
919 Self::Codex => "codex",
920 Self::Claude => "claude",
921 Self::Opencode => "opencode",
922 Self::OpenRouter => "openrouter",
923 Self::None => "none",
924 }
925 }
926}
927
928pub fn embed_via_backend(
943 models_dir: &Path,
944 text: &str,
945 backend: &LlmBackendKind,
946) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
947 match backend {
948 LlmBackendKind::None => Ok((Vec::new(), LlmBackendKind::None)),
949 LlmBackendKind::Codex => embed_passage_local_resolved(models_dir, text),
950 LlmBackendKind::Claude => {
951 tracing::debug!(
955 target: "embedder",
956 backend = "claude",
957 "embed_via_backend: forcing claude (ADR-0042 / GAP-002 fix)"
958 );
959 embed_via_claude_local_resolved(models_dir, text, None, None)
960 }
961 LlmBackendKind::Opencode => {
962 tracing::debug!(
963 target: "embedder",
964 backend = "opencode",
965 "embed_via_backend: forcing opencode (GAP-OPENCODE-001)"
966 );
967 embed_via_opencode_local_resolved(models_dir, text, None, None)
968 }
969 LlmBackendKind::OpenRouter => {
970 tracing::debug!(
971 target: "embedder",
972 backend = "openrouter",
973 "embed_via_backend: using OpenRouter API (v1.0.93)"
974 );
975 let client = OPENROUTER_CLIENT.get().ok_or_else(|| {
976 AppError::Embedding(
977 "OpenRouter client not initialised; call get_openrouter_embedder first".into(),
978 )
979 })?;
980 let rt = shared_runtime()?;
981 let vec = rt.block_on(client.embed_single(text, client.default_input_type()))?;
982 Ok((vec, LlmBackendKind::OpenRouter))
983 }
984 }
985}
986
987pub fn embed_via_backend_strict(
1000 models_dir: &Path,
1001 text: &str,
1002 backend: &LlmBackendKind,
1003 last_err: Option<&AppError>,
1004 skip_on_failure: bool,
1005) -> Result<(Vec<f32>, LlmBackendKind), AppError> {
1006 use crate::llm::exit_code_hints::LlmBackendError;
1007 match backend {
1008 LlmBackendKind::None => {
1009 if skip_on_failure && last_err.is_none() {
1013 Ok((Vec::new(), LlmBackendKind::None))
1014 } else if last_err.is_some() {
1015 Err(match last_err {
1019 Some(e) => AppError::Embedding(format!("{e}")),
1020 None => AppError::Embedding(LlmBackendError::NoBackendsAvailable.to_string()),
1021 })
1022 } else {
1023 Err(AppError::Embedding(
1026 LlmBackendError::NoBackendsAvailable.to_string(),
1027 ))
1028 }
1029 }
1030 LlmBackendKind::Codex => embed_passage_local_resolved(models_dir, text),
1031 LlmBackendKind::Claude => {
1032 tracing::debug!(
1033 target: "embedder",
1034 backend = "claude",
1035 "embed_via_backend_strict: forcing claude (ADR-0042 / GAP-002 fix)"
1036 );
1037 embed_via_claude_local_resolved(models_dir, text, None, None)
1038 }
1039 LlmBackendKind::Opencode => {
1040 tracing::debug!(
1041 target: "embedder",
1042 backend = "opencode",
1043 "embed_via_backend_strict: forcing opencode (GAP-OPENCODE-001)"
1044 );
1045 embed_via_opencode_local_resolved(models_dir, text, None, None)
1046 }
1047 LlmBackendKind::OpenRouter => embed_via_backend(models_dir, text, backend),
1048 }
1049}
1050
1051pub fn embed_via_backend_legacy(
1056 models_dir: &Path,
1057 text: &str,
1058 backend: &LlmBackendKind,
1059) -> Result<Vec<f32>, AppError> {
1060 embed_via_backend(models_dir, text, backend).map(|(v, _)| v)
1061}
1062
1063pub fn embed_passages_controlled_local(
1064 models_dir: &Path,
1065 texts: &[&str],
1066 token_counts: &[usize],
1067) -> Result<Vec<Vec<f32>>, AppError> {
1068 let embedder = get_embedder(models_dir)?;
1069 embed_passages_controlled(embedder, texts, token_counts)
1070}
1071
1072pub fn embed_passages_parallel_local(
1075 models_dir: &Path,
1076 texts: &[String],
1077 parallelism: usize,
1078 batch_size: usize,
1079) -> Result<Vec<Vec<f32>>, AppError> {
1080 let embedder = get_embedder(models_dir)?;
1081 embed_texts_parallel(embedder, texts, parallelism, batch_size)
1082}
1083
1084pub fn embed_passages_parallel_with_embedding_choice(
1091 models_dir: &Path,
1092 texts: &[String],
1093 parallelism: usize,
1094 batch_size: usize,
1095 embedding_backend: crate::cli::EmbeddingBackendChoice,
1096 llm_backend: crate::cli::LlmBackendChoice,
1097) -> Result<Vec<Vec<f32>>, AppError> {
1098 let chain = embedding_backend.to_chain(llm_backend);
1099 if chain.first() == Some(&LlmBackendKind::OpenRouter) && is_openrouter_initialized() {
1100 let client = OPENROUTER_CLIENT.get().ok_or_else(|| {
1101 AppError::Embedding(
1102 "OpenRouter client not initialised; call get_openrouter_embedder first".into(),
1103 )
1104 })?;
1105 let rt = shared_runtime()?;
1106 let refs: Vec<&str> = texts.iter().map(|s| s.as_str()).collect();
1107 let vecs = rt.block_on(client.embed_batch(&refs, client.default_input_type()))?;
1108 Ok(vecs)
1109 } else {
1110 embed_passages_parallel_local(models_dir, texts, parallelism, batch_size)
1111 }
1112}
1113
1114type EntityEmbedCacheMap = std::collections::HashMap<u64, Arc<Vec<f32>>>;
1126
1127static ENTITY_EMBED_CACHE: OnceLock<parking_lot::Mutex<EntityEmbedCacheMap>> = OnceLock::new();
1128
1129fn entity_embed_cache() -> &'static parking_lot::Mutex<EntityEmbedCacheMap> {
1130 ENTITY_EMBED_CACHE.get_or_init(|| parking_lot::Mutex::new(std::collections::HashMap::new()))
1131}
1132
1133fn entity_cache_key(model: &str, text: &str) -> u64 {
1134 let mut hasher = blake3::Hasher::new();
1135 hasher.update(model.as_bytes());
1136 hasher.update(b"\0");
1137 hasher.update(text.as_bytes());
1138 let h = hasher.finalize();
1139 let bytes = h.as_bytes();
1140 u64::from_le_bytes([
1141 bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
1142 ])
1143}
1144
1145pub fn embed_entity_texts_cached(
1155 models_dir: &Path,
1156 texts: &[String],
1157 parallelism: usize,
1158) -> Result<(Vec<Vec<f32>>, EmbedCacheStats), AppError> {
1159 if texts.is_empty() {
1160 return Ok((Vec::new(), EmbedCacheStats::default()));
1161 }
1162 let embedder = get_embedder(models_dir)?;
1163 let model = embedder.lock().model_label();
1164 let cache = entity_embed_cache();
1165 let mut hits: Vec<Option<Arc<Vec<f32>>>> = vec![None; texts.len()];
1166 let mut miss_indices: Vec<usize> = Vec::with_capacity(texts.len());
1167 {
1168 let guard = cache.lock();
1169 for (i, text) in texts.iter().enumerate() {
1170 let key = entity_cache_key(&model, text);
1171 if let Some(v) = guard.get(&key) {
1172 hits[i] = Some(Arc::clone(v));
1173 } else {
1174 miss_indices.push(i);
1175 }
1176 }
1177 }
1178 let miss_count = miss_indices.len();
1179 if miss_count > 0 {
1180 let miss_texts: Vec<String> = miss_indices.iter().map(|&i| texts[i].clone()).collect();
1181 let miss_vecs = embed_texts_parallel(
1182 embedder,
1183 &miss_texts,
1184 parallelism,
1185 entity_embed_batch_size(),
1186 )?;
1187 let mut guard = cache.lock();
1188 for (slot, &orig_idx) in miss_indices.iter().enumerate() {
1189 let vec = Arc::new(miss_vecs[slot].clone());
1190 let key = entity_cache_key(&model, &texts[orig_idx]);
1191 guard.insert(key, Arc::clone(&vec));
1192 hits[orig_idx] = Some(vec);
1193 }
1194 }
1195 let mut out = Vec::with_capacity(texts.len());
1196 for hit in hits.into_iter() {
1197 let v = hit.ok_or_else(|| {
1198 AppError::Embedding("entity embed cache produced null result".to_string())
1199 })?;
1200 out.push((*v).clone());
1201 }
1202 Ok((
1203 out,
1204 EmbedCacheStats {
1205 requested: texts.len(),
1206 hits: texts.len() - miss_count,
1207 misses: miss_count,
1208 },
1209 ))
1210}
1211
1212#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize)]
1214pub struct EmbedCacheStats {
1215 pub requested: usize,
1216 pub hits: usize,
1217 pub misses: usize,
1218}
1219
1220impl EmbedCacheStats {
1221 pub fn hit_rate(&self) -> f64 {
1223 if self.requested == 0 {
1224 0.0
1225 } else {
1226 self.hits as f64 / self.requested as f64
1227 }
1228 }
1229}
1230
1231pub fn embed_texts_parallel(
1244 embedder: &Mutex<LlmEmbedding>,
1245 texts: &[String],
1246 parallelism: usize,
1247 batch_size: usize,
1248) -> Result<Vec<Vec<f32>>, AppError> {
1249 let mut slots: Vec<Option<Vec<f32>>> = vec![None; texts.len()];
1250 embed_texts_parallel_with(embedder, texts, parallelism, batch_size, |idx, v| {
1251 slots[idx] = Some(v.to_vec());
1252 Ok(())
1253 })?;
1254 let mut out = Vec::with_capacity(slots.len());
1255 for (idx, slot) in slots.into_iter().enumerate() {
1256 out.push(slot.ok_or_else(|| {
1257 AppError::Embedding(format!("embedding fan-out lost item index {idx}"))
1258 })?);
1259 }
1260 Ok(out)
1261}
1262
1263pub fn embed_texts_parallel_with(
1267 embedder: &Mutex<LlmEmbedding>,
1268 texts: &[String],
1269 parallelism: usize,
1270 batch_size: usize,
1271 mut on_result: impl FnMut(usize, &[f32]) -> Result<(), AppError>,
1272) -> Result<(), AppError> {
1273 if texts.is_empty() {
1274 return Ok(());
1275 }
1276 let dim = crate::constants::embedding_dim();
1277 if texts.len() == 1 {
1278 let v = embed_passage(embedder, &texts[0])?;
1279 return on_result(0, &v);
1280 }
1281
1282 let client = clone_client(embedder);
1283 let permits = effective_permits(parallelism);
1284 let batches = build_batches(texts, batch_size.max(1));
1285 let token = crate::cancel_token().clone();
1286
1287 let work = move |batch: Vec<(usize, String)>| {
1288 let client = client.clone();
1289 async move {
1290 client
1291 .embed_batch_async(crate::constants::PASSAGE_PREFIX, &batch)
1292 .await
1293 }
1294 };
1295
1296 let fan_out = run_bounded(batches, permits, dim, token, work, &mut on_result);
1297 match tokio::runtime::Handle::try_current() {
1298 Ok(handle) => tokio::task::block_in_place(|| handle.block_on(fan_out)),
1299 Err(_) => shared_runtime()?.block_on(fan_out),
1300 }
1301}
1302
1303fn build_batches(texts: &[String], batch_size: usize) -> Vec<Vec<(usize, String)>> {
1305 texts
1306 .iter()
1307 .cloned()
1308 .enumerate()
1309 .collect::<Vec<_>>()
1310 .chunks(batch_size)
1311 .map(|c| c.to_vec())
1312 .collect()
1313}
1314
1315pub fn effective_permits(requested: usize) -> usize {
1320 let cpus = std::thread::available_parallelism()
1321 .map(|n| n.get())
1322 .unwrap_or(4);
1323 let by_ram = ((crate::memory_guard::available_memory_mb() / 2)
1324 / crate::constants::LLM_WORKER_RSS_MB)
1325 .max(1) as usize;
1326 requested.clamp(1, 32).min(cpus).min(by_ram).max(1)
1327}
1328
1329async fn run_bounded<F, Fut>(
1339 batches: Vec<Vec<(usize, String)>>,
1340 permits: usize,
1341 dim: usize,
1342 token: CancellationToken,
1343 work: F,
1344 on_result: &mut impl FnMut(usize, &[f32]) -> Result<(), AppError>,
1345) -> Result<(), AppError>
1346where
1347 F: Fn(Vec<(usize, String)>) -> Fut + Clone + Send + 'static,
1348 Fut: std::future::Future<Output = Result<Vec<(usize, Vec<f32>)>, AppError>> + Send,
1349{
1350 let total_batches = batches.len();
1351 let semaphore = Arc::new(Semaphore::new(permits));
1352 let (tx, mut rx) = mpsc::channel::<Result<Vec<(usize, Vec<f32>)>, AppError>>(permits * 2);
1355 let mut set: JoinSet<()> = JoinSet::new();
1356
1357 for (batch_idx, batch) in batches.into_iter().enumerate() {
1358 let sem = Arc::clone(&semaphore);
1359 let token = token.clone();
1360 let tx = tx.clone();
1361 let work = work.clone();
1362 set.spawn(async move {
1363 let wait_start = std::time::Instant::now();
1364 let Ok(_permit) = sem.acquire_owned().await else {
1367 let _ = tx
1368 .send(Err(AppError::Embedding("semaphore closed".to_string())))
1369 .await;
1370 return;
1371 };
1372 let permit_wait_ms = wait_start.elapsed().as_millis() as u64;
1373 let work_start = std::time::Instant::now();
1374 let outcome = if crate::should_obey_shutdown() {
1380 tokio::select! {
1381 res = work(batch) => res,
1382 _ = token.cancelled() => Err(AppError::Embedding(
1383 "embedding cancelled by shutdown signal".to_string(),
1384 )),
1385 }
1386 } else {
1387 work(batch).await
1388 };
1389 tracing::debug!(
1391 target: "embedding",
1392 batch_idx,
1393 permit_wait_ms,
1394 work_ms = work_start.elapsed().as_millis() as u64,
1395 ok = outcome.is_ok(),
1396 "embedding batch finished"
1397 );
1398 let _ = tx.send(outcome).await;
1399 });
1400 }
1401 drop(tx);
1402
1403 let mut completed = 0usize;
1404 let mut failed = 0usize;
1405 let mut cancelled = 0usize;
1406 let mut first_error: Option<AppError> = None;
1407
1408 while let Some(message) = rx.recv().await {
1409 match message {
1410 Ok(items) => {
1411 completed += 1;
1412 if first_error.is_none() {
1413 for (idx, v) in items {
1414 if v.len() != dim {
1415 first_error = Some(AppError::Embedding(format!(
1416 "LLM returned {} dims for item {idx}, expected {dim}; \
1417 refusing to truncate or pad silently (G42/C5)",
1418 v.len()
1419 )));
1420 break;
1421 }
1422 if let Err(e) = on_result(idx, &v) {
1423 first_error = Some(e);
1424 break;
1425 }
1426 }
1427 if first_error.is_some() {
1428 set.shutdown().await;
1431 }
1432 }
1433 }
1434 Err(e) => {
1435 if matches!(&e, AppError::Embedding(msg) if msg.contains("cancelled")) {
1436 cancelled += 1;
1437 } else {
1438 failed += 1;
1439 }
1440 if first_error.is_none() {
1441 first_error = Some(e);
1442 set.shutdown().await;
1443 }
1444 }
1445 }
1446 }
1447
1448 while let Some(join_result) = set.join_next().await {
1451 if let Err(join_err) = join_result {
1452 if join_err.is_panic() {
1453 failed += 1;
1454 if first_error.is_none() {
1455 first_error = Some(AppError::Embedding(format!(
1456 "embedding task panicked: {join_err}"
1457 )));
1458 }
1459 } else {
1460 cancelled += 1;
1461 }
1462 }
1463 }
1464
1465 tracing::debug!(
1475 target: "embedding",
1476 total_batches,
1477 completed,
1478 failed,
1479 cancelled,
1480 "embedding fan-out finished"
1481 );
1482
1483 match first_error {
1484 Some(e) => Err(e),
1485 None => Ok(()),
1486 }
1487}
1488
1489pub fn f32_to_bytes(v: &[f32]) -> Vec<u8> {
1490 let mut out = Vec::with_capacity(v.len() * 4);
1491 for f in v {
1492 out.extend_from_slice(&f.to_le_bytes());
1493 }
1494 out
1495}
1496
1497pub fn bytes_to_f32(bytes: &[u8]) -> Vec<f32> {
1498 let mut out = Vec::with_capacity(bytes.len() / 4);
1499 for chunk in bytes.chunks_exact(4) {
1500 out.push(f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]));
1501 }
1502 out
1503}
1504
1505pub fn embedding_dim() -> usize {
1508 crate::constants::embedding_dim()
1509}
1510
1511fn validate_dim(v: Vec<f32>) -> Result<Vec<f32>, AppError> {
1515 let dim = crate::constants::embedding_dim();
1516 if v.len() != dim {
1517 return Err(AppError::Embedding(format!(
1518 "embedding has {} dims, expected {dim}; \
1519 refusing to truncate or pad silently (G42/C5)",
1520 v.len()
1521 )));
1522 }
1523 Ok(v)
1524}
1525
1526#[cfg(test)]
1527mod tests {
1528 use super::*;
1529 use std::sync::atomic::{AtomicUsize, Ordering};
1530
1531 #[test]
1532 fn f32_to_bytes_roundtrip() {
1533 let input = vec![0.0_f32, 1.5, -2.25, f32::MIN, f32::MAX];
1534 let bytes = f32_to_bytes(&input);
1535 assert_eq!(bytes.len(), input.len() * 4);
1536 let out = bytes_to_f32(&bytes);
1537 assert_eq!(out, input);
1538 }
1539
1540 #[test]
1541 fn validate_dim_rejects_divergent_vectors() {
1542 let dim = crate::constants::embedding_dim();
1545 let long = vec![0.0; dim + 10];
1546 assert!(validate_dim(long).is_err(), "longer vector must error");
1547 let short = vec![0.0; dim.saturating_sub(1).max(1)];
1548 assert!(validate_dim(short).is_err(), "shorter vector must error");
1549 let exact = vec![0.0; dim];
1550 assert_eq!(validate_dim(exact).expect("exact dim must pass").len(), dim);
1551 }
1552
1553 #[test]
1554 fn embedding_dim_matches_constants_source() {
1555 assert_eq!(embedding_dim(), crate::constants::embedding_dim());
1556 }
1557
1558 #[test]
1559 fn build_batches_preserves_global_indices() {
1560 let texts: Vec<String> = (0..10).map(|i| format!("t{i}")).collect();
1561 let batches = build_batches(&texts, 4);
1562 assert_eq!(batches.len(), 3);
1563 assert_eq!(batches[0].len(), 4);
1564 assert_eq!(batches[2].len(), 2);
1565 assert_eq!(batches[2][1].0, 9);
1566 assert_eq!(batches[2][1].1, "t9");
1567 }
1568
1569 #[test]
1570 fn effective_permits_clamps_to_bounds() {
1571 assert!(effective_permits(0) >= 1);
1572 assert!(effective_permits(1000) <= 32);
1573 }
1574
1575 fn test_batches(n: usize) -> Vec<Vec<(usize, String)>> {
1576 (0..n).map(|i| vec![(i, format!("t{i}"))]).collect()
1577 }
1578
1579 fn dummy_vec(dim: usize) -> Vec<f32> {
1580 vec![0.0; dim]
1581 }
1582
1583 #[test]
1586 fn concurrency_peak_never_exceeds_permits() {
1587 let permits = 4usize;
1588 let batches = test_batches(permits * 10);
1589 let dim = crate::constants::embedding_dim();
1590 let current = Arc::new(AtomicUsize::new(0));
1591 let peak = Arc::new(AtomicUsize::new(0));
1592
1593 let current_c = Arc::clone(¤t);
1594 let peak_c = Arc::clone(&peak);
1595 let work = move |batch: Vec<(usize, String)>| {
1596 let current = Arc::clone(¤t_c);
1597 let peak = Arc::clone(&peak_c);
1598 async move {
1599 let now = current.fetch_add(1, Ordering::SeqCst) + 1;
1600 peak.fetch_max(now, Ordering::SeqCst);
1601 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1602 current.fetch_sub(1, Ordering::SeqCst);
1603 Ok(batch
1604 .into_iter()
1605 .map(|(i, _)| (i, dummy_vec(crate::constants::embedding_dim())))
1606 .collect())
1607 }
1608 };
1609
1610 let mut delivered = 0usize;
1611 let rt = tokio::runtime::Builder::new_multi_thread()
1612 .worker_threads(4)
1613 .enable_all()
1614 .build()
1615 .expect("test runtime");
1616 rt.block_on(run_bounded(
1617 batches,
1618 permits,
1619 dim,
1620 CancellationToken::new(),
1621 work,
1622 &mut |_idx, _v| {
1623 delivered += 1;
1624 Ok(())
1625 },
1626 ))
1627 .expect("fan-out must succeed");
1628
1629 assert_eq!(delivered, permits * 10, "every item must be delivered");
1630 assert!(
1631 peak.load(Ordering::SeqCst) <= permits,
1632 "peak concurrency {} exceeded permits {permits}",
1633 peak.load(Ordering::SeqCst)
1634 );
1635 }
1636
1637 #[test]
1640 fn panicking_task_returns_permit_and_surfaces_error() {
1641 let permits = 2usize;
1642 let batches = test_batches(4);
1643 let dim = crate::constants::embedding_dim();
1644
1645 let work = move |batch: Vec<(usize, String)>| async move {
1646 if batch[0].0 == 1 {
1647 panic!("intentional test panic");
1648 }
1649 Ok(batch
1650 .into_iter()
1651 .map(|(i, _)| (i, dummy_vec(crate::constants::embedding_dim())))
1652 .collect())
1653 };
1654
1655 let rt = tokio::runtime::Builder::new_multi_thread()
1656 .worker_threads(2)
1657 .enable_all()
1658 .build()
1659 .expect("test runtime");
1660 let result = rt.block_on(run_bounded(
1661 batches,
1662 permits,
1663 dim,
1664 CancellationToken::new(),
1665 work,
1666 &mut |_idx, _v| Ok(()),
1667 ));
1668
1669 let err = result.expect_err("panic must surface as an error");
1670 assert!(
1671 err.to_string().contains("panicked"),
1672 "error must mention the panic: {err}"
1673 );
1674 }
1675
1676 #[test]
1679 fn cancellation_terminates_fan_out_quickly() {
1680 let permits = 2usize;
1681 let batches = test_batches(8);
1682 let dim = crate::constants::embedding_dim();
1683 let token = CancellationToken::new();
1684
1685 let work = move |batch: Vec<(usize, String)>| async move {
1686 tokio::time::sleep(std::time::Duration::from_secs(30)).await;
1688 Ok(batch
1689 .into_iter()
1690 .map(|(i, _)| (i, dummy_vec(crate::constants::embedding_dim())))
1691 .collect())
1692 };
1693
1694 let rt = tokio::runtime::Builder::new_multi_thread()
1695 .worker_threads(2)
1696 .enable_all()
1697 .build()
1698 .expect("test runtime");
1699 let cancel = token.clone();
1700 let start = std::time::Instant::now();
1701 let result = rt.block_on(async move {
1702 tokio::spawn(async move {
1703 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1704 cancel.cancel();
1705 });
1706 run_bounded(batches, permits, dim, token, work, &mut |_idx, _v| Ok(())).await
1707 });
1708
1709 assert!(result.is_err(), "cancelled fan-out must report an error");
1710 assert!(
1711 start.elapsed() < std::time::Duration::from_secs(10),
1712 "graceful shutdown must finish well under the work duration"
1713 );
1714 }
1715
1716 #[test]
1719 fn fan_out_rejects_divergent_dim() {
1720 let permits = 2usize;
1721 let batches = test_batches(2);
1722 let dim = crate::constants::embedding_dim();
1723
1724 let work = move |batch: Vec<(usize, String)>| async move {
1725 Ok(batch
1726 .into_iter()
1727 .map(|(i, _)| (i, vec![0.0f32; 3]))
1728 .collect::<Vec<(usize, Vec<f32>)>>())
1729 };
1730
1731 let rt = tokio::runtime::Builder::new_multi_thread()
1732 .worker_threads(2)
1733 .enable_all()
1734 .build()
1735 .expect("test runtime");
1736 let result = rt.block_on(run_bounded(
1737 batches,
1738 permits,
1739 dim,
1740 CancellationToken::new(),
1741 work,
1742 &mut |_idx, _v| Ok(()),
1743 ));
1744
1745 let err = result.expect_err("divergent dim must fail the fan-out");
1746 assert!(err.to_string().contains("G42/C5"), "error cites C5: {err}");
1747 }
1748
1749 #[test]
1751 fn adaptive_batch_dim64_keeps_calibrated_sizes() {
1752 assert_eq!(adaptive_batch_for_dim(CHUNK_EMBED_BATCH_SIZE, 64), 8);
1753 assert_eq!(adaptive_batch_for_dim(ENTITY_EMBED_BATCH_SIZE, 64), 25);
1754 }
1755
1756 #[test]
1758 fn adaptive_batch_dim384_shrinks() {
1759 assert_eq!(adaptive_batch_for_dim(CHUNK_EMBED_BATCH_SIZE, 384), 1);
1760 assert_eq!(adaptive_batch_for_dim(ENTITY_EMBED_BATCH_SIZE, 384), 4);
1761 }
1762
1763 #[test]
1765 fn adaptive_batch_intermediate_dims() {
1766 assert_eq!(adaptive_batch_for_dim(8, 128), 4);
1767 assert_eq!(adaptive_batch_for_dim(8, 256), 2);
1768 }
1769
1770 #[test]
1772 fn adaptive_batch_small_dim_clamps_to_base() {
1773 assert_eq!(adaptive_batch_for_dim(8, 8), 8);
1774 }
1775
1776 #[test]
1778 fn adaptive_batch_total_function() {
1779 assert_eq!(adaptive_batch_for_dim(8, 4096), 1);
1780 assert_eq!(adaptive_batch_for_dim(8, 0), 8);
1781 assert_eq!(adaptive_batch_for_dim(0, 64), 1);
1782 }
1783
1784 #[test]
1786 #[serial_test::serial(env)]
1787 fn adaptive_wrappers_follow_env_dim() {
1788 std::env::set_var("SQLITE_GRAPHRAG_EMBEDDING_DIM", "384");
1789 let chunk = chunk_embed_batch_size();
1790 let entity = entity_embed_batch_size();
1791 std::env::remove_var("SQLITE_GRAPHRAG_EMBEDDING_DIM");
1792 crate::constants::set_active_embedding_dim(crate::constants::DEFAULT_EMBEDDING_DIM);
1793 assert_eq!(chunk, 1, "384-dim chunk batch must shrink to 1 (G44)");
1794 assert_eq!(entity, 4, "384-dim entity batch must shrink to 4 (G44)");
1795 }
1796
1797 #[test]
1805 fn embedding_error_kind_classify_oauth_message() {
1806 assert_eq!(
1807 EmbeddingErrorKind::classify("OAuth token expired for claude"),
1808 EmbeddingErrorKind::OAuth,
1809 );
1810 assert_eq!(
1811 EmbeddingErrorKind::classify("oauth authentication failed"),
1812 EmbeddingErrorKind::OAuth,
1813 );
1814 }
1815
1816 #[test]
1819 fn embedding_error_kind_classify_quota_message() {
1820 assert_eq!(
1821 EmbeddingErrorKind::classify("quota exhausted on backend"),
1822 EmbeddingErrorKind::Quota,
1823 );
1824 assert_eq!(
1825 EmbeddingErrorKind::classify("Usage quota limit reached"),
1826 EmbeddingErrorKind::Quota,
1827 );
1828 }
1829
1830 #[test]
1834 fn embedding_error_kind_classify_slot_exhausted_message() {
1835 assert_eq!(
1836 EmbeddingErrorKind::classify(
1837 "slot exhausted: failed to acquire LLM slot after backoff"
1838 ),
1839 EmbeddingErrorKind::SlotExhausted,
1840 );
1841 }
1842
1843 #[test]
1846 fn embedding_error_kind_classify_zero_dimension_message() {
1847 assert_eq!(
1848 EmbeddingErrorKind::classify("embedding returned dim=zero"),
1849 EmbeddingErrorKind::ZeroDimension,
1850 );
1851 assert_eq!(
1852 EmbeddingErrorKind::classify("got zero-dim vector from LLM"),
1853 EmbeddingErrorKind::ZeroDimension,
1854 );
1855 }
1856
1857 #[test]
1861 fn embedding_error_kind_classify_unknown_fallback() {
1862 assert_eq!(
1863 EmbeddingErrorKind::classify("unrelated subprocess error"),
1864 EmbeddingErrorKind::Unknown,
1865 );
1866 assert_eq!(
1867 EmbeddingErrorKind::classify("rate limit hit"),
1868 EmbeddingErrorKind::Unknown,
1869 );
1870 assert_eq!(EmbeddingErrorKind::OAuth.code(), "oauth");
1872 assert_eq!(EmbeddingErrorKind::Quota.code(), "quota");
1873 assert_eq!(EmbeddingErrorKind::SlotExhausted.code(), "slot-exhausted");
1874 assert_eq!(
1875 EmbeddingErrorKind::BackendMismatch.code(),
1876 "backend-mismatch"
1877 );
1878 assert_eq!(EmbeddingErrorKind::ZeroDimension.code(), "zero-dimension");
1879 assert_eq!(EmbeddingErrorKind::Unknown.code(), "unknown");
1880 }
1881
1882 #[test]
1884 fn fallback_reason_display_does_not_panic() {
1885 let _ = FallbackReason::EmbeddingFailed("rate limit".into()).to_string();
1886 let _ = FallbackReason::Cancelled.to_string();
1887 let _ = FallbackReason::Timeout {
1888 operation: "embed_query".into(),
1889 duration_secs: 30,
1890 }
1891 .to_string();
1892 }
1893
1894 #[test]
1897 fn fallback_reason_is_partial_eq() {
1898 assert_eq!(
1899 FallbackReason::EmbeddingFailed("a".into()),
1900 FallbackReason::EmbeddingFailed("a".into())
1901 );
1902 assert_eq!(FallbackReason::Cancelled, FallbackReason::Cancelled);
1903 assert_ne!(
1904 FallbackReason::EmbeddingFailed("a".into()),
1905 FallbackReason::EmbeddingFailed("b".into())
1906 );
1907 assert_ne!(
1908 FallbackReason::Cancelled,
1909 FallbackReason::Timeout {
1910 operation: "x".into(),
1911 duration_secs: 1
1912 }
1913 );
1914 }
1915
1916 #[test]
1919 fn fallback_reason_timeout_preserves_fields() {
1920 let r = FallbackReason::Timeout {
1921 operation: "embed_query_local".into(),
1922 duration_secs: 300,
1923 };
1924 match r {
1925 FallbackReason::Timeout {
1926 operation,
1927 duration_secs,
1928 } => {
1929 assert_eq!(operation, "embed_query_local");
1930 assert_eq!(duration_secs, 300);
1931 }
1932 other => panic!("expected Timeout, got {other:?}"),
1933 }
1934 }
1935
1936 #[test]
1942 #[ignore = "G58 S1 stub: requires env without codex/claude on PATH; tracked as T5 of Fase 2"]
1943 fn try_embed_query_with_fallback_surfaces_embedding_failed_for_missing_binary() {
1944 let bogus = std::path::Path::new("/nonexistent-models-dir-for-g58-fallback-test");
1947 let result = try_embed_query_with_fallback(bogus, "hello world");
1948 match result {
1949 Err(FallbackReason::EmbeddingFailed(msg)) => {
1950 assert!(!msg.is_empty(), "fallback message must not be empty");
1952 }
1953 Err(FallbackReason::Cancelled) => {
1954 panic!("expected EmbeddingFailed, got Cancelled");
1955 }
1956 Err(FallbackReason::Timeout { .. }) => {
1957 panic!("expected EmbeddingFailed, got Timeout");
1958 }
1959 Err(FallbackReason::SlotExhausted) => {
1960 panic!("expected EmbeddingFailed, got SlotExhausted");
1961 }
1962 Err(FallbackReason::OAuthQuota { .. }) => {
1963 panic!("expected EmbeddingFailed, got OAuthQuota");
1964 }
1965 Err(FallbackReason::BackendMismatch { .. }) => {
1966 panic!("expected EmbeddingFailed, got BackendMismatch");
1967 }
1968 Err(FallbackReason::DimZero) => {
1969 panic!("expected EmbeddingFailed, got DimZero");
1970 }
1971 Ok(_) => {
1972 panic!("expected an error, got Ok — embedder must fail for bogus path");
1973 }
1974 }
1975 }
1976
1977 #[test]
1979 fn g56_entity_cache_key_is_stable_and_distinct() {
1980 let k1 = entity_cache_key("codex:default", "sqlite-graphrag");
1981 let k2 = entity_cache_key("codex:default", "sqlite-graphrag");
1982 let k3 = entity_cache_key("codex:default", "claude-code");
1983 let k4 = entity_cache_key("claude:default", "sqlite-graphrag");
1984 assert_eq!(k1, k2, "same model+text must hash identically");
1985 assert_ne!(k1, k3, "different text must hash differently");
1986 assert_ne!(k1, k4, "different model must hash differently");
1987 }
1988
1989 #[test]
1990 fn g56_entity_embed_cache_stats_hit_rate() {
1991 let zero = EmbedCacheStats::default();
1992 assert_eq!(zero.hit_rate(), 0.0);
1993 let half = EmbedCacheStats {
1994 requested: 4,
1995 hits: 2,
1996 misses: 2,
1997 };
1998 assert!((half.hit_rate() - 0.5).abs() < 1e-9);
1999 let all = EmbedCacheStats {
2000 requested: 7,
2001 hits: 7,
2002 misses: 0,
2003 };
2004 assert!((all.hit_rate() - 1.0).abs() < 1e-9);
2005 }
2006
2007 #[test]
2008 fn g56_entity_embed_cache_populates_and_hits() {
2009 let cache = entity_embed_cache();
2013 let model = "test-model";
2014 let text = "sqlite-graphrag";
2015 let key = entity_cache_key(model, text);
2016 let stored = Arc::new(vec![0.42_f32; crate::constants::embedding_dim()]);
2017 cache.lock().insert(key, Arc::clone(&stored));
2018 let guard = cache.lock();
2019 let hit = guard.get(&key).expect("cache must return stored value");
2020 assert_eq!(hit.len(), crate::constants::embedding_dim());
2021 assert!((hit[0] - 0.42).abs() < 1e-6);
2022 }
2023
2024 #[test]
2025 fn g56_empty_texts_short_circuits_with_zero_stats() {
2026 let stats = EmbedCacheStats::default();
2029 assert_eq!(stats.requested, 0);
2030 assert_eq!(stats.hits, 0);
2031 assert_eq!(stats.misses, 0);
2032 assert_eq!(stats.hit_rate(), 0.0);
2033 }
2034}
2035
2036#[cfg(test)]
2040mod embed_with_fallback_tests {
2041 use super::*;
2042 use crate::llm::exit_code_hints::LlmBackendError;
2043
2044 #[test]
2045 fn none_backend_returns_empty_vector_without_calling_llm() {
2046 let (v, kind) = embed_via_backend(
2050 std::path::Path::new("/nonexistent"),
2051 "any text",
2052 &LlmBackendKind::None,
2053 )
2054 .expect("None backend never fails");
2055 assert!(v.is_empty());
2056 assert_eq!(kind, LlmBackendKind::None, "None backend must report None");
2057 }
2058
2059 #[test]
2060 fn empty_chain_defaults_to_codex_claude_none() {
2061 let defaults = [
2065 LlmBackendKind::Codex,
2066 LlmBackendKind::Claude,
2067 LlmBackendKind::None,
2068 ];
2069
2070 #[allow(dead_code)]
2075 fn llm_backend_kind_as_str_is_stable() {
2076 assert_eq!(LlmBackendKind::Codex.as_str(), "codex");
2077 assert_eq!(LlmBackendKind::Claude.as_str(), "claude");
2078 assert_eq!(LlmBackendKind::None.as_str(), "none");
2079 }
2080
2081 #[allow(dead_code)]
2082 fn fallback_reason_reason_code_is_stable() {
2083 assert_eq!(
2084 FallbackReason::EmbeddingFailed("any".into()).reason_code(),
2085 "embedding_failed"
2086 );
2087 assert_eq!(FallbackReason::Cancelled.reason_code(), "cancelled");
2088 assert_eq!(
2089 FallbackReason::Timeout {
2090 operation: "embed_query".into(),
2091 duration_secs: 30
2092 }
2093 .reason_code(),
2094 "timeout"
2095 );
2096 }
2097 assert_eq!(defaults.len(), 3);
2098 }
2099
2100 #[test]
2101 fn embed_with_fallback_chain_of_only_none_aborts_without_skip_on_failure_v1088() {
2102 let chain = vec![LlmBackendKind::None];
2114 let err = embed_with_fallback(
2115 std::path::Path::new("/nonexistent-models-dir-for-gap005-test"),
2116 "hello",
2117 &chain,
2118 false,
2119 )
2120 .expect_err("chain of only [None] without skip_on_failure MUST abort");
2121 let msg = format!("{err}");
2122 assert!(
2123 msg.contains("no LLM backends available"),
2124 "error must mention exhausted chain, got: {msg}"
2125 );
2126 }
2127 #[test]
2128 fn embed_with_fallback_skip_on_failure_with_only_none_returns_empty() {
2129 let chain = vec![LlmBackendKind::None];
2134 let v = embed_with_fallback(
2135 std::path::Path::new("/nonexistent-models-dir-for-gap005-test"),
2136 "hello",
2137 &chain,
2138 true,
2139 )
2140 .expect("None chain is always Ok");
2141 assert!(v.0.is_empty(), "vector must be empty");
2142 assert_eq!(v.1, LlmBackendKind::None);
2143 }
2144 #[allow(dead_code)]
2145 fn llm_backend_error_no_backends_default_message() {
2146 let e = LlmBackendError::NoBackendsAvailable;
2149 let h = e.hint();
2150 assert!(h.contains("--llm-fallback"));
2151 }
2152
2153 #[test]
2154 fn llm_backend_error_nonzero_exit_carries_stderr_tail() {
2155 let e = LlmBackendError::NonZeroExit {
2156 exit_code: Some(137),
2157 signal: Some(9),
2158 stdout_tail: "out".into(),
2159 stderr_tail: "OOM killed".into(),
2160 binary: "codex".into(),
2161 hint: "OOM".into(),
2162 };
2163 let s = e.to_string();
2164 assert!(s.contains("codex"));
2165 assert!(s.contains("OOM killed"));
2166 assert!(s.contains("signal 9") || s.contains("exit 137"));
2167 }
2168}