Phase 0: shared search core + reproducible corpus

Scaffold the StreamSearch repo and build the reusable heart with no infra:
- embed.py: singleton MiniLM (all-MiniLM-L6-v2, 384-d), batched embed_texts
- search.py: hybrid BM25 + dense (Qdrant) fused with RRF (k=60)
- schema.py: Document contract + Cassandra/Qdrant definitions
- data/corpus_sample.jsonl: 3000 arXiv abstracts (HF mirror, reproducible)
- tests: RRF, embedding determinism, single model-load, hybrid ranking (10 pass)
- pinned requirements.txt (Python 3.11), Makefile, MIT LICENSE, README

Deployment: unified live site over managed free tiers (Qdrant Cloud, Astra
Cassandra, Confluent/Redpanda Kafka); Spark shown via local compose + video.

.gitattributes

@@ -0,0 +1,5 @@
+# Normalize line endings to LF in the repo (avoids CRLF churn on Windows).
+* text=auto eol=lf
+*.png binary
+*.jpg binary
+*.snapshot binary

.gitignore

@@ -0,0 +1,27 @@
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+venv/
+*.egg-info/
+.pytest_cache/
+
+# Model caches (large; downloaded at runtime)
+.cache/
+models/
+*.safetensors
+
+# Baked demo index snapshots (rebuilt via scripts/build_index.py)
+app/prebuilt/*
+!app/prebuilt/.gitkeep
+
+# Local infra volumes / data
+qdrant_storage/
+cassandra_data/
+redpanda_data/
+*.log
+
+# OS / editor
+.DS_Store
+.idea/
+.vscode/

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Aryan Sharma
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

@@ -0,0 +1,35 @@
+# StreamSearch — task runner.
+# On Windows without `make`, run the underlying commands directly (see README),
+# or install make (`winget install GnuWin32.Make`). The demo/proof hosts are
+# Linux, where these targets run as-is.
+
+PY ?= .venv/bin/python
+ifeq ($(OS),Windows_NT)
+	PY = .venv/Scripts/python.exe
+endif
+
+.PHONY: setup test build-index demo produce stream record all
+
+setup:            ## create venv deps
+	$(PY) -m pip install --upgrade pip
+	$(PY) -m pip install -r requirements.txt
+
+test:             ## run the test suite
+	$(PY) -m pytest -q
+
+build-index:      ## build the baked-in demo index from the sample corpus
+	$(PY) scripts/build_index.py
+
+demo:             ## run the demo tier locally (FastAPI + Streamlit)
+	$(PY) -m streamlit run app/ui.py
+
+produce:          ## stream the corpus into the Kafka topic (proof tier)
+	$(PY) producer/produce.py
+
+stream:           ## run the Spark Structured Streaming job (proof tier)
+	$(PY) streaming/spark_job.py
+
+record:           ## print the proof-tier recording steps
+	@cat scripts/record_demo.md
+
+all: build-index demo   ## build the demo index then launch the local demo

README.md

@@ -0,0 +1,79 @@
+# StreamSearch — Real-Time Semantic Search over a Streaming Corpus
+
+> **🔗 Live demo:** _coming in Phase 1_ · **📹 Pipeline video:** _coming in Phase 4_
+
+Documents stream through **Kafka**, are processed and embedded by **Spark
+Structured Streaming**, and land in **Cassandra** (metadata) and a **Qdrant**
+vector database (embeddings) for continuous indexing. Hybrid search — BM25 +
+dense embeddings fused with Reciprocal Rank Fusion — is served over the live
+index through a FastAPI + Streamlit app.
+
+## Deployment model — unified live site over managed free tiers
+
+One clickable site a recruiter can actually use, backed by **real** managed
+infrastructure on free tiers:
+
+- **Live tier** — FastAPI + Streamlit app (Render / HF Spaces) over **Qdrant
+  Cloud** (vectors) and **DataStax Astra** (Cassandra metadata). The ingest box
+  publishes to **real Confluent/Redpanda Cloud Kafka**; a small always-on
+  consumer writes to Cassandra + Qdrant, so a pasted document genuinely flows
+  through Kafka → store → search results. *(Phases 1–3)*
+- **Proof tier** — the full **Spark Structured Streaming** job (the one piece
+  with no free always-on host) runs against the same stack via local
+  `docker compose`, captured on video. This makes the Spark resume bullet
+  literally true. *(Phases 3–4)*
+
+The shared embedding + search core (`src/streamsearch/`) is written once and
+used by both the live consumer and the Spark job.
+
+## Architecture
+
+```
+                      ┌───────── shared core (src/streamsearch) ──────────┐
+                      │  embed.py · search.py (BM25+dense+RRF k=60) · schema │
+                      └───────────────┬───────────────────┬────────────────┘
+ ingest box ─┐                        │                   │
+             ▼                        ▼                   ▼
+   Kafka (Confluent/Redpanda Cloud)   │        LIVE: small consumer worker
+             │                        │          ├─→ Astra (Cassandra) metadata
+             ├──────────────────────► │          └─→ Qdrant Cloud   vectors
+             │                        │
+   PROOF: Spark Structured Streaming (docker compose, foreachBatch: embed once)
+             └─────────────────────────────► same Cassandra + Qdrant
+                                                     │
+                          FastAPI /search /ingest  ·  Streamlit UI  (Render/HF)
+```
+
+## Quickstart (local, Windows)
+
+```powershell
+py -3.11 -m venv .venv
+.venv\Scripts\python.exe -m pip install -r requirements.txt
+.venv\Scripts\python.exe -m pytest -q          # Phase 0 tests
+```
+
+> **Note:** built and pinned for **Python 3.11**. `make` targets mirror these
+> commands and run on the Linux demo/proof hosts; on Windows either install
+> `make` or run the `.venv\Scripts\python.exe ...` commands directly.
+
+## Corpus
+
+`data/corpus_sample.jsonl` — a fixed set of **3,000 arXiv abstracts** (ML/CS)
+built from the `CShorten/ML-ArXiv-Papers` Hugging Face dataset via
+`scripts/fetch_corpus.py` and committed for reproducibility. (We use the HF
+mirror because the live arXiv API rate-limits bursty fetches with HTTP 429.)
+The demo is never empty. `datasets` is a build-time-only dependency.
+
+## Scope (honest notes)
+
+Single-broker Redpanda, single-node Cassandra, at-least-once micro-batch
+delivery. This is a portfolio piece, not a production platform.
+
+## Status
+
+- [x] Phase 0 — shared search core + tests
+- [ ] Phase 1 — demo tier deployed (live URL)
+- [ ] Phase 2 — Kafka + producer
+- [ ] Phase 3 — Spark → Cassandra + Qdrant
+- [ ] Phase 4 — unify + record video
+- [ ] Phase 5 — tests, README, polish

requirements.txt

@@ -0,0 +1,19 @@
+# --- Core: shared embed + search (Phase 0), reused by both tiers ---
+sentence-transformers==3.0.1
+qdrant-client==1.9.2
+rank-bm25==0.2.2
+numpy==1.26.4
+
+# --- Demo tier (Phase 1): FastAPI + Streamlit ---
+fastapi==0.111.0
+uvicorn[standard]==0.30.1
+streamlit==1.36.0
+pydantic==2.7.4
+
+# --- Proof tier (Phases 2-3): Kafka producer, Spark, Cassandra ---
+confluent-kafka==2.4.0
+pyspark==3.5.1
+cassandra-driver==3.29.1
+
+# --- Tests (Phase 5) ---
+pytest==8.2.2

scripts/fetch_corpus.py

@@ -0,0 +1,65 @@
+"""Build a fixed, reproducible corpus_sample.jsonl of arXiv abstracts.
+
+Source: the `CShorten/ML-ArXiv-Papers` Hugging Face dataset (arXiv titles +
+abstracts). We use a stable HF mirror rather than the live arXiv API because the
+API rate-limits (HTTP 429) hard on bursty fetches. The output is checked into git
+so runs are deterministic and the demo is never empty (spec, non-negotiable #3).
+
+Requires: pip install datasets   (build-time only; not a runtime dependency)
+
+Usage:
+    python scripts/fetch_corpus.py --target 3000
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+
+HF_DATASET = "CShorten/ML-ArXiv-Papers"
+_WS = re.compile(r"\s+")
+
+
+def clean(text: str) -> str:
+    return _WS.sub(" ", (text or "")).strip()
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--target", type=int, default=3000, help="number of documents")
+    ap.add_argument("--out", type=Path, default=Path("data/corpus_sample.jsonl"))
+    args = ap.parse_args()
+
+    from datasets import load_dataset
+
+    print(f"Loading {HF_DATASET} ...")
+    ds = load_dataset(HF_DATASET, split="train")
+
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    written = 0
+    seen: set[str] = set()
+    with args.out.open("w", encoding="utf-8") as f:
+        for i, row in enumerate(ds):
+            title = clean(row.get("title"))
+            abstract = clean(row.get("abstract"))
+            if not abstract or title in seen:
+                continue
+            seen.add(title)
+            rec = {
+                "id": f"arxiv:{i}",
+                "title": title,
+                "text": abstract,
+                "source": "arxiv/ml",
+            }
+            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+            written += 1
+            if written >= args.target:
+                break
+
+    print(f"Wrote {written} docs to {args.out}")
+
+
+if __name__ == "__main__":
+    main()

src/streamsearch/__init__.py

@@ -0,0 +1,2 @@
+from .embed import embed_texts, get_model
+from .search import HybridSearcher, reciprocal_rank_fusion

src/streamsearch/embed.py

@@ -0,0 +1,47 @@
+"""Shared embedding: load the small model ONCE, embed in batches.
+
+The model is a module-level singleton. It must never be reloaded per document
+(spec, Phase 0 and Phase 3 gotcha). Both the demo tier and the Spark foreachBatch
+handler import `embed_texts` from here.
+"""
+
+from __future__ import annotations
+
+import logging
+from functools import lru_cache
+
+import numpy as np
+
+from .schema import MODEL_NAME, EMBED_DIM
+
+log = logging.getLogger(__name__)
+
+
+@lru_cache(maxsize=1)
+def get_model():
+    """Load and cache the sentence-transformer. Logged so we can prove single-load."""
+    from sentence_transformers import SentenceTransformer
+
+    log.info("Loading embedding model %s (one-time)", MODEL_NAME)
+    model = SentenceTransformer(MODEL_NAME)
+    return model
+
+
+def embed_texts(texts: list[str], batch_size: int = 64) -> np.ndarray:
+    """Batch-embed texts into an (n, EMBED_DIM) float32 array of unit vectors.
+
+    Normalized embeddings mean cosine similarity == dot product, which is what
+    Qdrant is configured for (Distance.COSINE).
+    """
+    if not texts:
+        return np.empty((0, EMBED_DIM), dtype=np.float32)
+
+    model = get_model()
+    vectors = model.encode(
+        texts,
+        batch_size=batch_size,
+        normalize_embeddings=True,
+        convert_to_numpy=True,
+        show_progress_bar=False,
+    )
+    return vectors.astype(np.float32)

src/streamsearch/schema.py

@@ -0,0 +1,77 @@
+"""Shared record schema and store definitions.
+
+One record shape flows through the whole system: the producer emits it as JSON
+onto Kafka, Spark parses it, Cassandra stores the metadata, and Qdrant stores the
+embedding keyed by the same id. Keeping the contract here means the demo tier and
+proof tier cannot drift apart.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+from typing import Any
+
+# Embedding model config. all-MiniLM-L6-v2 is 384-dim, ~90 MB, chosen so the
+# free demo host does not run out of memory (see spec, non-negotiable #2).
+MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
+EMBED_DIM = 384
+
+# Qdrant collection holding the dense vectors.
+QDRANT_COLLECTION = "documents"
+
+# Kafka topic the producer streams onto and Spark consumes from.
+KAFKA_TOPIC = "documents"
+
+
+@dataclass
+class Document:
+    """A single corpus document. `id` is the join key across Kafka/Cassandra/Qdrant."""
+
+    id: str
+    title: str
+    text: str
+    source: str = "corpus_sample"
+
+    def to_json(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_json(cls, obj: dict[str, Any]) -> "Document":
+        return cls(
+            id=str(obj["id"]),
+            title=obj.get("title", ""),
+            text=obj["text"],
+            source=obj.get("source", "corpus_sample"),
+        )
+
+    @property
+    def embed_input(self) -> str:
+        """Text fed to the embedder: title + body so titles carry weight."""
+        return f"{self.title}\n\n{self.text}".strip() if self.title else self.text
+
+
+# --- Cassandra ---------------------------------------------------------------
+# Single-node keyspace/table for document metadata. The vector lives in Qdrant;
+# Cassandra is the system of record for the raw text + metadata.
+CASSANDRA_KEYSPACE = "streamsearch"
+CASSANDRA_TABLE = "documents"
+
+CREATE_KEYSPACE_CQL = (
+    f"CREATE KEYSPACE IF NOT EXISTS {CASSANDRA_KEYSPACE} "
+    "WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};"
+)
+
+CREATE_TABLE_CQL = f"""
+CREATE TABLE IF NOT EXISTS {CASSANDRA_KEYSPACE}.{CASSANDRA_TABLE} (
+    id text PRIMARY KEY,
+    title text,
+    text text,
+    source text,
+    ingested_at timestamp
+);
+"""
+
+INSERT_CQL = f"""
+INSERT INTO {CASSANDRA_KEYSPACE}.{CASSANDRA_TABLE} (id, title, text, source, ingested_at)
+VALUES (?, ?, ?, ?, toTimestamp(now()));
+"""

src/streamsearch/search.py

@@ -0,0 +1,162 @@
+"""Shared hybrid search: BM25 (lexical) + dense (Qdrant) fused with RRF.
+
+Written once here and reused by both tiers:
+  - demo tier serves it over a baked-in Qdrant snapshot,
+  - proof tier serves it over the Qdrant populated by the Spark job.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass, field
+from typing import Iterable
+
+from qdrant_client import QdrantClient
+from qdrant_client.http import models as qmodels
+from rank_bm25 import BM25Okapi
+
+from .embed import embed_texts
+from .schema import EMBED_DIM, QDRANT_COLLECTION
+
+log = logging.getLogger(__name__)
+
+RRF_K = 60  # standard Reciprocal Rank Fusion constant (spec: k = 60)
+
+_TOKEN_RE = re.compile(r"[a-z0-9]+")
+
+
+def tokenize(text: str) -> list[str]:
+    return _TOKEN_RE.findall(text.lower())
+
+
+@dataclass
+class SearchResult:
+    id: str
+    title: str
+    text: str
+    score: float           # fused RRF score
+    bm25_rank: int | None = None
+    dense_rank: int | None = None
+    payload: dict = field(default_factory=dict)
+
+
+def reciprocal_rank_fusion(
+    ranked_lists: dict[str, list[str]], k: int = RRF_K
+) -> dict[str, float]:
+    """Fuse several ranked id-lists into one score map.
+
+    ranked_lists maps a stage name -> ids ordered best-first. RRF score for a
+    doc is sum over stages of 1 / (k + rank), rank being 1-based. Pure function,
+    unit-tested independently of any index.
+    """
+    scores: dict[str, float] = {}
+    for ids in ranked_lists.values():
+        for rank, doc_id in enumerate(ids, start=1):
+            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
+    return scores
+
+
+class HybridSearcher:
+    """Holds a Qdrant client for dense search and an in-memory BM25 index.
+
+    BM25 is kept in memory (rank_bm25) built from the same docs indexed in
+    Qdrant. `add_documents` keeps both stages in sync so the demo ingest box
+    reflects new docs in both lexical and dense results.
+    """
+
+    def __init__(self, client: QdrantClient, collection: str = QDRANT_COLLECTION):
+        self.client = client
+        self.collection = collection
+        self._ids: list[str] = []
+        self._corpus_tokens: list[list[str]] = []
+        self._bm25: BM25Okapi | None = None
+        self._meta: dict[str, dict] = {}
+
+    # --- index construction -------------------------------------------------
+    def ensure_collection(self) -> None:
+        existing = {c.name for c in self.client.get_collections().collections}
+        if self.collection not in existing:
+            self.client.create_collection(
+                collection_name=self.collection,
+                vectors_config=qmodels.VectorParams(
+                    size=EMBED_DIM, distance=qmodels.Distance.COSINE
+                ),
+            )
+
+    def add_documents(self, docs: Iterable) -> int:
+        """Embed + upsert docs to Qdrant and (re)build the BM25 index.
+
+        `docs` are schema.Document instances. Returns count added.
+        """
+        docs = list(docs)
+        if not docs:
+            return 0
+
+        vectors = embed_texts([d.embed_input for d in docs])
+        points = [
+            qmodels.PointStruct(
+                id=_qdrant_id(d.id),
+                vector=vectors[i].tolist(),
+                payload={"doc_id": d.id, "title": d.title, "text": d.text,
+                         "source": d.source},
+            )
+            for i, d in enumerate(docs)
+        ]
+        self.client.upsert(collection_name=self.collection, points=points)
+
+        for d in docs:
+            self._ids.append(d.id)
+            self._corpus_tokens.append(tokenize(d.embed_input))
+            self._meta[d.id] = {"title": d.title, "text": d.text, "source": d.source}
+        self._bm25 = BM25Okapi(self._corpus_tokens)
+        log.info("Indexed %d docs (total %d)", len(docs), len(self._ids))
+        return len(docs)
+
+    # --- query --------------------------------------------------------------
+    def search(self, query: str, top_k: int = 10, candidates: int = 50) -> list[SearchResult]:
+        dense_ids = self._dense_search(query, candidates)
+        bm25_ids = self._bm25_search(query, candidates)
+
+        fused = reciprocal_rank_fusion({"dense": dense_ids, "bm25": bm25_ids})
+        dense_rank = {doc_id: i + 1 for i, doc_id in enumerate(dense_ids)}
+        bm25_rank = {doc_id: i + 1 for i, doc_id in enumerate(bm25_ids)}
+
+        ordered = sorted(fused.items(), key=lambda kv: kv[1], reverse=True)[:top_k]
+        results = []
+        for doc_id, score in ordered:
+            meta = self._meta.get(doc_id, {})
+            results.append(
+                SearchResult(
+                    id=doc_id,
+                    title=meta.get("title", ""),
+                    text=meta.get("text", ""),
+                    score=score,
+                    bm25_rank=bm25_rank.get(doc_id),
+                    dense_rank=dense_rank.get(doc_id),
+                    payload=meta,
+                )
+            )
+        return results
+
+    def _dense_search(self, query: str, limit: int) -> list[str]:
+        qvec = embed_texts([query])[0].tolist()
+        hits = self.client.search(
+            collection_name=self.collection, query_vector=qvec, limit=limit,
+            with_payload=True,
+        )
+        return [h.payload.get("doc_id", str(h.id)) for h in hits]
+
+    def _bm25_search(self, query: str, limit: int) -> list[str]:
+        if self._bm25 is None or not self._ids:
+            return []
+        scores = self._bm25.get_scores(tokenize(query))
+        top = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:limit]
+        return [self._ids[i] for i in top if scores[i] > 0]
+
+
+def _qdrant_id(doc_id: str) -> int:
+    """Qdrant point ids must be int or UUID. Hash string ids to a stable int."""
+    import hashlib
+
+    return int(hashlib.md5(doc_id.encode()).hexdigest()[:15], 16)

tests/conftest.py

@@ -0,0 +1,8 @@
+import sys
+from pathlib import Path
+
+# Make src/ importable without an editable install.
+ROOT = Path(__file__).resolve().parents[1]
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))

tests/fixtures/mini_corpus.jsonl

@@ -0,0 +1,6 @@
+{"id": "d1", "title": "Transformer models for language", "text": "Attention based neural networks have transformed natural language processing and machine translation.", "source": "test"}
+{"id": "d2", "title": "Convolutional networks for vision", "text": "Deep convolutional neural networks achieve strong results on image classification benchmarks.", "source": "test"}
+{"id": "d3", "title": "Distributed stream processing", "text": "Apache Kafka and Spark Structured Streaming enable scalable real time data pipelines.", "source": "test"}
+{"id": "d4", "title": "Vector databases and search", "text": "Approximate nearest neighbor indexes like HNSW power fast semantic vector search at scale.", "source": "test"}
+{"id": "d5", "title": "Reinforcement learning basics", "text": "An agent learns a policy by maximizing cumulative reward through interaction with an environment.", "source": "test"}
+{"id": "d6", "title": "Cassandra wide column store", "text": "Cassandra is a distributed NoSQL database offering tunable consistency and high write throughput.", "source": "test"}

tests/test_embed.py

@@ -0,0 +1,33 @@
+import numpy as np
+
+from streamsearch import embed
+from streamsearch.schema import EMBED_DIM
+
+
+def test_embed_shape_and_norm():
+    vecs = embed.embed_texts(["hello world", "a second document"])
+    assert vecs.shape == (2, EMBED_DIM)
+    assert vecs.dtype == np.float32
+    # normalize_embeddings=True -> unit vectors
+    norms = np.linalg.norm(vecs, axis=1)
+    np.testing.assert_allclose(norms, 1.0, atol=1e-4)
+
+
+def test_embed_is_deterministic():
+    a = embed.embed_texts(["reproducible embedding"])
+    b = embed.embed_texts(["reproducible embedding"])
+    np.testing.assert_allclose(a, b, atol=1e-6)
+
+
+def test_empty_input_returns_empty():
+    vecs = embed.embed_texts([])
+    assert vecs.shape == (0, EMBED_DIM)
+
+
+def test_model_loads_only_once():
+    # get_model is lru_cached: repeated calls return the same object.
+    embed.get_model.cache_clear()
+    m1 = embed.get_model()
+    m2 = embed.get_model()
+    assert m1 is m2
+    assert embed.get_model.cache_info().hits >= 1

tests/test_search.py

@@ -0,0 +1,74 @@
+import json
+from pathlib import Path
+
+import pytest
+from qdrant_client import QdrantClient
+
+from streamsearch.schema import Document
+from streamsearch.search import HybridSearcher, reciprocal_rank_fusion
+
+FIXTURE = Path(__file__).parent / "fixtures" / "mini_corpus.jsonl"
+
+
+def load_fixture() -> list[Document]:
+    docs = []
+    with FIXTURE.open(encoding="utf-8") as f:
+        for line in f:
+            docs.append(Document.from_json(json.loads(line)))
+    return docs
+
+
+@pytest.fixture(scope="module")
+def searcher() -> HybridSearcher:
+    client = QdrantClient(":memory:")
+    s = HybridSearcher(client)
+    s.ensure_collection()
+    s.add_documents(load_fixture())
+    return s
+
+
+# --- RRF is a pure function; test it without any index -----------------------
+def test_rrf_rewards_agreement_across_stages():
+    fused = reciprocal_rank_fusion(
+        {"dense": ["a", "b", "c"], "bm25": ["b", "a", "d"]}, k=60
+    )
+    # 'b' is rank1(bm25)+rank2(dense); 'a' is rank1(dense)+rank2(bm25) -> tie-ish,
+    # but both should top 'c' and 'd' which appear in only one list.
+    assert fused["a"] > fused["c"]
+    assert fused["b"] > fused["d"]
+
+
+def test_rrf_score_formula():
+    fused = reciprocal_rank_fusion({"only": ["x", "y"]}, k=60)
+    assert fused["x"] == pytest.approx(1 / 61)
+    assert fused["y"] == pytest.approx(1 / 62)
+
+
+# --- hybrid search over a real (in-memory) Qdrant index ----------------------
+def test_semantic_query_returns_relevant_doc(searcher):
+    results = searcher.search("neural attention for translation", top_k=3)
+    ids = [r.id for r in results]
+    assert "d1" in ids  # transformer/attention doc should surface
+    assert results[0].score > 0
+
+
+def test_lexical_query_matches_exact_terms(searcher):
+    results = searcher.search("Kafka Spark streaming pipeline", top_k=3)
+    assert results[0].id == "d3"
+
+
+def test_results_carry_per_stage_ranks(searcher):
+    results = searcher.search("vector search HNSW index", top_k=5)
+    top = results[0]
+    assert top.id == "d4"
+    # at least one stage should have ranked the top result
+    assert top.dense_rank is not None or top.bm25_rank is not None
+
+
+def test_ingested_doc_becomes_searchable(searcher):
+    searcher.add_documents([
+        Document(id="new1", title="Quantum computing",
+                 text="Qubits leverage superposition and entanglement for computation.")
+    ])
+    results = searcher.search("superposition entanglement qubits", top_k=3)
+    assert "new1" in [r.id for r in results]