feat: Mode A hybrid pipeline with graceful degradation to C

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/pipeline.rs

@@ -1,10 +1,14 @@
 use crate::claml;
 use crate::corpus::{self};
-use crate::fusion::cross_segment_dedupe;
+use crate::embed::EmbeddingStore;
+use crate::fusion::{cross_segment_dedupe, rrf_merge};
+use crate::ionos::IonosClient;
 use crate::lexical::LexicalIndex;
 use crate::model::*;
+use crate::rerank;
 use crate::segment::split_segments;
 use crate::tags::tags_for;
+use crate::vector::VectorIndex;
 use std::collections::HashMap;
 use std::time::Instant;
 
@@ -14,6 +18,8 @@ pub struct Pipeline {
     pub lexical: LexicalIndex,
     pub pool_size: usize,
     pub valid_by_code: HashMap<String, bool>,
+    cfg: Config,
+    vector: Option<VectorIndex>,
 }
 
 impl Pipeline {
@@ -31,7 +37,26 @@ impl Pipeline {
             }
             m
         };
-        Ok(Self { entries, meta, lexical, pool_size: cfg.pool_size, valid_by_code })
+        // Build the semantic index only if EVERY corpus entry already has a
+        // cached embedding; otherwise leave it `None` so Mode A degrades to C
+        // rather than mispairing partial vectors.
+        let vector = match EmbeddingStore::open(&cfg.index_dir, &cfg.embed_model) {
+            Ok(store) if entries.iter().all(|e| store.has(&e.text)) => {
+                let vecs: Option<Vec<Vec<f32>>> =
+                    entries.iter().map(|e| store.get(&e.text)).collect();
+                vecs.map(VectorIndex::from_vectors)
+            }
+            _ => None,
+        };
+        Ok(Self {
+            entries,
+            meta,
+            lexical,
+            pool_size: cfg.pool_size,
+            valid_by_code,
+            cfg: cfg.clone(),
+            vector,
+        })
     }
 
     /// Look up ICD metadata: exact code first, then — only for 6-char dotted codes
@@ -83,12 +108,10 @@ impl Pipeline {
         })
     }
 
-    pub fn suggest(&self, dictation: &str, mode: Mode, filter: &Filter, top_k: usize)
+    /// Mode C candidate collection: lexical pool per segment. This is the
-        -> SuggestResult {
+    /// exact original behavior; it is also the degrade fallback for Mode A.
-        let t0 = Instant::now();
+    fn collect_mode_c(&self, segments: &[String]) -> Vec<Candidate> {
-        let segments = split_segments(dictation);
         let mut candidates: Vec<Candidate> = Vec::new();
-
         for (si, seg) in segments.iter().enumerate() {
             let lex = self.lexical.search(seg, self.pool_size).unwrap_or_default();
             for (entry_idx, score) in lex {
@@ -105,6 +128,88 @@ impl Pipeline {
                 }
             }
         }
+        candidates
+    }
+
+    /// Mode A candidate collection: per segment, fuse a lexical pool with a
+    /// semantic pool (IONOS embedding + in-RAM vector search) via RRF, then
+    /// cross-encoder rerank the fused pool. ANY error (IONOS unreachable, no
+    /// vector index, embed/rerank failure) propagates so `suggest` degrades.
+    fn collect_mode_a(&self, segments: &[String], calls: &mut usize)
+        -> Result<Vec<Candidate>, AppError> {
+        let client = IonosClient::new(&self.cfg.ionos_base_url, &self.cfg.token_path)?;
+        let vector = self.vector.as_ref()
+            .ok_or_else(|| AppError::Ionos("no vector index".into()))?;
+        let mut candidates: Vec<Candidate> = Vec::new();
+
+        for (si, seg) in segments.iter().enumerate() {
+            // Lexical pool (entry indices).
+            let lex = self.lexical.search(seg, self.pool_size).unwrap_or_default();
+            let lex_ids: Vec<usize> = lex.iter().map(|(i, _)| *i).collect();
+
+            // Semantic pool: embed the segment, then exact cosine search.
+            let body = serde_json::json!({
+                "model": self.cfg.embed_model, "input": [seg]
+            });
+            let resp: serde_json::Value = client.post_json("/embeddings", &body)?;
+            *calls += 1;
+            let q: Vec<f32> = resp["data"][0]["embedding"].as_array()
+                .ok_or_else(|| AppError::Ionos("embeddings: no data[0].embedding".into()))?
+                .iter()
+                .map(|x| x.as_f64()
+                    .ok_or_else(|| AppError::Ionos("embeddings: non-numeric component".into())))
+                .collect::<Result<Vec<f64>, _>>()?
+                .into_iter().map(|f| f as f32).collect();
+            let sem = vector.top_k(&q, self.pool_size);
+            let sem_ids: Vec<usize> = sem.iter().map(|(i, _)| *i).collect();
+
+            // Fuse the two ranked index lists; cap at pool_size.
+            let mut fused = rrf_merge(&lex_ids, &sem_ids, 60);
+            fused.truncate(self.pool_size);
+
+            // Cross-encoder rerank the fused pool.
+            let docs: Vec<String> = fused.iter()
+                .map(|&i| self.entries[i].text.clone())
+                .collect();
+            let scores = rerank::rerank(&client, &self.cfg.rerank_model, seg, &docs)?;
+            *calls += 1;
+
+            for (rank, &entry_idx) in fused.iter().enumerate() {
+                let e = &self.entries[entry_idx];
+                if let Some(code) = corpus::primary_code(e) {
+                    candidates.push(Candidate {
+                        icd_code: code.to_string(),
+                        alpha_text: e.text.clone(),
+                        segment_idx: si,
+                        lexical: None,
+                        semantic: None,
+                        rerank: scores.get(rank).copied(),
+                    });
+                }
+            }
+        }
+        Ok(candidates)
+    }
+
+    pub fn suggest(&self, dictation: &str, mode: Mode, filter: &Filter, top_k: usize)
+        -> SuggestResult {
+        let t0 = Instant::now();
+        let segments = split_segments(dictation);
+        let mut ionos_calls = 0usize;
+        let mut degraded = false;
+
+        let candidates: Vec<Candidate> = match mode {
+            Mode::C => self.collect_mode_c(&segments),
+            Mode::A => match self.collect_mode_a(&segments, &mut ionos_calls) {
+                Ok(c) => c,
+                Err(_) => {
+                    // Total degradation: any failure → Mode C fallback,
+                    // request still answered, never panics.
+                    degraded = true;
+                    self.collect_mode_c(&segments)
+                }
+            },
+        };
 
         let merged = cross_segment_dedupe(candidates);
         let mut suggestions = Vec::new();
@@ -123,9 +228,9 @@ impl Pipeline {
             suggestions,
             diagnostics: Diagnostics {
                 mode: format!("{:?}", mode),
-                degraded: false,
+                degraded,
                 segment_count: segments.len(),
-                ionos_calls: 0,
+                ionos_calls,
                 millis: t0.elapsed().as_millis(),
             },
         }

tests/pipeline_mode_a_tests.rs

@@ -0,0 +1,13 @@
+use alpha_id::pipeline::Pipeline;
+use alpha_id::model::{Config, Filter, Mode};
+
+#[test]
+fn mode_a_degrades_to_c_when_ionos_unreachable() {
+    let mut cfg = Config::load("config/default.toml").unwrap();
+    cfg.ionos_base_url = "http://127.0.0.1:1".into(); // refused
+    let p = Pipeline::load(&cfg).unwrap();
+    let res = p.suggest("Diabetes mellitus Typ 2", Mode::A, &Filter::default(), 10);
+    assert!(res.diagnostics.degraded);                 // fell back
+    assert!(!res.suggestions.is_empty());              // still answered
+    assert!(res.suggestions.iter().any(|s| s.icd_code.starts_with("E11")));
+}