feat(evidence): use server-side Qdrant BM25 retrieval

harness/tests/test_corpus_pipeline.py

@@ -277,7 +277,7 @@ def test_active_searcher_without_active_fails_closed_for_evidence(tmp_path):
     from tht.evidence.search import active_searcher
 
     class Delegate:
-        def search(self, embedding, top_n=10, kinds=None, metadata_filter=None):
+        def search(self, embedding, top_n=10, kinds=None, metadata_filter=None, **kwargs):
             return ["legacy"]
 
     cfg = SimpleNamespace(paths=SimpleNamespace(artifacts=tmp_path / "artifacts"))
@@ -332,14 +332,14 @@ def test_active_evidence_query_holds_lock_against_publish(tmp_path):
     published = threading.Event()
 
     class Delegate:
-        def search(self, embedding, top_n=10, kinds=None, metadata_filter=None):
+        def search(self, embedding, top_n=10, kinds=None, metadata_filter=None, **kwargs):
             entered.set()
             assert release.wait(5)
             return [SimpleNamespace(id="active", similarity=1.0)]
 
     search = threading.Thread(
         target=lambda: ActiveEvidenceSearcher(store, Delegate()).search(
-            [1.0], kinds=["evidence"]
+            [1.0], kinds=["evidence"], query_text="old"
         )
     )
     search.start()
@@ -360,6 +360,17 @@ def test_active_evidence_query_holds_lock_against_publish(tmp_path):
     assert published.is_set()
 
 
+def test_active_evidence_search_refuses_a_dense_only_fallback(tmp_path):
+    from tht.evidence.search import ActiveEvidenceSearcher
+    from tht.ports.vector import VectorStoreError
+
+    current = pipeline(tmp_path, Source([(item("one", "a"), "cardiomiopatia")]), vectors=Vectors())
+    current.run()
+
+    with pytest.raises(VectorStoreError, match="hybrid query text"):
+        ActiveEvidenceSearcher(current.store, object()).search([1.0], kinds=["evidence"])
+
+
 def test_pipeline_result_dump_does_not_deepcopy_frozen_metadata():
     manifest = CorpusManifest(metadata={"nested": {"value": ["safe"]}})
     payload = PipelineResult(
@@ -716,6 +727,16 @@ def test_dimension_mismatch_fails_before_vector_write_and_publish(tmp_path):
     assert candidate.store.active_generation() is None
 
 
+def test_pipeline_marks_each_evidence_fragment_for_server_side_italian_bm25(tmp_path):
+    vectors = Vectors()
+
+    pipeline(tmp_path, Source([(item("one", "a"), "ricovero cardiologico")]), vectors=vectors).run()
+
+    assert [(record.sparse_text, record.sparse_language) for record in vectors.records] == [
+        ("ricovero cardiologico", "italian"),
+    ]
+
+
 def test_dry_run_and_failed_acquire_never_change_active(tmp_path):
     one = item("one", "a")
     active = pipeline(tmp_path, Source([(one, "old")])).run().generation