feat(embedding): apply asymmetric query/document prefixes to embed calls

packages/core/src/repowise/core/analysis/decisions/semantic_match.py

@@ -357,7 +357,13 @@ async def find_duplicate_decision(
     if not query:
         return None
     try:
-        results = await store.search(query, limit=SEARCH_FETCH)
+        # kind="document": this is decision text compared against other
+        # decision text (near-duplicate matching), not a natural-language
+        # question against stored documents. Stored decision vectors always
+        # embed at the document prefix (see upsert_decision_vectors); an
+        # asymmetric-prefix embedder that got the default "query" here would
+        # search with the wrong framing and silently miss real duplicates.
+        results = await store.search(query, limit=SEARCH_FETCH, kind="document")
     except Exception:
         return None
 
@@ -397,7 +403,8 @@ async def find_related_decisions(
     if not query:
         return []
     try:
-        results = await store.search(query, limit=SEARCH_FETCH)
+        # kind="document" — same reasoning as find_duplicate_decision above.
+        results = await store.search(query, limit=SEARCH_FETCH, kind="document")
     except Exception:
         return []
     return _related_from_results(results, lo=lo, hi=hi, exclude_ids=exclude_ids, limit=limit)
@@ -452,7 +459,10 @@ async def find_related_decisions_many(
     try:
         # Empty query texts still occupy their slot (keeps results aligned)
         # but must not reach the embedder — some providers reject "".
-        all_results = await store.search_many([q or " " for q in queries], limit=SEARCH_FETCH)
+        # kind="document" — same reasoning as find_duplicate_decision above.
+        all_results = await store.search_many(
+            [q or " " for q in queries], limit=SEARCH_FETCH, kind="document"
+        )
     except Exception:
         return [[] for _ in items]
     if len(all_results) != len(items):

packages/core/src/repowise/core/persistence/vector_store/_base.py

@@ -195,15 +195,27 @@ async def embed_batch(self, items: list[tuple[str, str, dict]]) -> None:
         for page_id, text, metadata in items:
             await self.embed_and_upsert(page_id, text, metadata)
 
-    async def embed_texts(self, texts: list[str]) -> list[list[float]] | None:
+    async def embed_texts(
+        self, texts: list[str], *, kind: str = "document"
+    ) -> list[list[float]] | None:
         """Embed *texts* in batched embedder requests, without upserting.
 
         Lets a caller that needs the raw vectors (e.g. decision dedup, which
-        searches *and* upserts the same text) pay for one batched embedding
-        instead of one round-trip per item. Returns ``None`` when the backend
-        holds no embedder — callers must fall back to the per-item text APIs.
-        Chunked so a large input can't blow the embedder's per-request token
-        cap; each text is capped at :data:`EMBED_TEXT_MAX_CHARS`.
+        searches *and* upserts the same text — pass ``kind="document"``, the
+        default, so both directions embed identically) pay for one batched
+        embedding instead of one round-trip per item. Returns ``None`` when
+        the backend holds no embedder — callers must fall back to the
+        per-item text APIs. Chunked so a large input can't blow the
+        embedder's per-request token cap; each text is capped at
+        :data:`EMBED_TEXT_MAX_CHARS`.
+
+        Args:
+            kind: ``"query"`` or ``"document"`` — forwarded to the embedder
+                so a directional model (see
+                :func:`repowise.core.providers.embedding.base.resolve_embed_prefix`)
+                applies the right framing. A caller embedding a natural-
+                language question to search against stored pages must pass
+                ``kind="query"``.
         """
         embedder = getattr(self, "_embedder", None)
         if embedder is None:
@@ -212,7 +224,7 @@ async def embed_texts(self, texts: list[str]) -> list[list[float]] | None:
             return []
         out: list[list[float]] = []
         for _chunk, capped_texts in iter_embed_chunks([("", t, {}) for t in texts]):
-            out.extend(await embedder.embed(capped_texts))
+            out.extend(await embedder.embed(capped_texts, kind=kind))
         return out
 
     async def search_by_vector(
@@ -239,11 +251,26 @@ async def upsert_vectors(self, items: list[tuple[str, list[float], dict]]) -> bo
         return False
 
     @abstractmethod
-    async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
-        """Embed *query* and return the *limit* nearest pages."""
+    async def search(
+        self, query: str, limit: int = 10, *, kind: str = "query"
+    ) -> list[SearchResult]:
+        """Embed *query* and return the *limit* nearest pages.
+
+        Args:
+            kind: ``"query"`` or ``"document"``, forwarded to the embedder
+                (see :meth:`embed_texts`). The default matches what every
+                existing caller wants — *query* is a natural-language
+                question being matched against stored documents. Decision
+                near-duplicate lookup is the one caller that isn't: it is
+                comparing decision text to other decision text, symmetric by
+                construction, so it passes ``kind="document"`` to match the
+                prefix the candidate text was — or will be — stored under.
+        """
         ...
 
-    async def search_many(self, queries: list[str], limit: int = 10) -> list[list[SearchResult]]:
+    async def search_many(
+        self, queries: list[str], limit: int = 10, *, kind: str = "query"
+    ) -> list[list[SearchResult]]:
         """Batch variant of :meth:`search` — one result list per query, aligned
         by index.
 
@@ -253,13 +280,16 @@ async def search_many(self, queries: list[str], limit: int = 10) -> list[list[Se
         Backends override this to embed *all* queries in a single embedder
         call — the network round-trip dominates each search, so batching the
         embedding turns N round-trips into 1.
+
+        Args:
+            kind: See :meth:`search`; forwarded unchanged to every query.
         """
         import asyncio as _asyncio
 
         if not queries:
             return []
         results = await _asyncio.gather(
-            *(self.search(q, limit=limit) for q in queries), return_exceptions=True
+            *(self.search(q, limit=limit, kind=kind) for q in queries), return_exceptions=True
         )
         return [r if isinstance(r, list) else [] for r in results]
 

packages/core/src/repowise/core/persistence/vector_store/in_memory.py

@@ -66,24 +66,28 @@
             self._store[page_id] = (list(vector), dict(metadata))
         return True
 
-    async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
+    async def search(
+        self, query: str, limit: int = 10, *, kind: str = "query"
+    ) -> list[SearchResult]:
         if not self._store:
             return []
-        q_vecs = await self._embedder.embed([query])
+        q_vecs = await self._embedder.embed([query], kind=kind)
         return self._search_by_vector(q_vecs[0], limit)
 
     async def search_by_vector(self, vector: list[float], limit: int = 10) -> list[SearchResult]:
         if not self._store:
             return []
         return self._search_by_vector(vector, limit)
 
-    async def search_many(self, queries: list[str], limit: int = 10) -> list[list[SearchResult]]:
+    async def search_many(
+        self, queries: list[str], limit: int = 10, *, kind: str = "query"
+    ) -> list[list[SearchResult]]:
         """One embedder call for all queries, then local scoring per query."""
         if not queries:
             return []
         if not self._store:
             return [[] for _ in queries]
-        q_vecs = await self._embedder.embed(list(queries))
+        q_vecs = await self._embedder.embed(list(queries), kind=kind)
         return [self._search_by_vector(q_vec, limit) for q_vec in q_vecs]
 
     async def delete(self, page_id: str) -> None:

packages/core/src/repowise/core/persistence/vector_store/lancedb_store.py

@@ -263,12 +263,14 @@ async def upsert_vectors(self, items: list[tuple[str, list[float], dict]]) -> bo
         await self._upsert_rows(rows)
         return True
 
-    async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
+    async def search(
+        self, query: str, limit: int = 10, *, kind: str = "query"
+    ) -> list[SearchResult]:
         await self._ensure_connected()
         if self._table is None:
             return []
 
-        q_vecs = await self._embedder.embed([query])
+        q_vecs = await self._embedder.embed([query], kind=kind)
         return await self._search_by_vector([float(v) for v in q_vecs[0]], limit, query=query)
 
     async def search_by_vector(self, vector: list[float], limit: int = 10) -> list[SearchResult]:
@@ -277,14 +279,16 @@ async def search_by_vector(self, vector: list[float], limit: int = 10) -> list[S
             return []
         return await self._search_by_vector([float(v) for v in vector], limit)
 
-    async def search_many(self, queries: list[str], limit: int = 10) -> list[list[SearchResult]]:
+    async def search_many(
+        self, queries: list[str], limit: int = 10, *, kind: str = "query"
+    ) -> list[list[SearchResult]]:
         """One embedder call for all queries; the vector lookups are local."""
         if not queries:
             return []
         await self._ensure_connected()
         if self._table is None:
             return [[] for _ in queries]
-        q_vecs = await self._embedder.embed(list(queries))
+        q_vecs = await self._embedder.embed(list(queries), kind=kind)
         out: list[list[SearchResult]] = []
         for query, q_vec in zip(queries, q_vecs, strict=True):
             try:

packages/core/src/repowise/core/persistence/vector_store/pgvector_store.py

@@ -114,8 +114,10 @@ async def upsert_vectors(self, items: list[tuple[str, list[float], dict]]) -> bo
             await session.commit()
         return True
 
-    async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
-        q_vecs = await self._embedder.embed([query])
+    async def search(
+        self, query: str, limit: int = 10, *, kind: str = "query"
+    ) -> list[SearchResult]:
+        q_vecs = await self._embedder.embed([query], kind=kind)
         return await self.search_by_vector(q_vecs[0], limit)
 
     async def search_by_vector(self, vector: list[float], limit: int = 10) -> list[SearchResult]:
@@ -150,11 +152,13 @@ async def search_by_vector(self, vector: list[float], limit: int = 10) -> list[S
             for r in raw
         ]
 
-    async def search_many(self, queries: list[str], limit: int = 10) -> list[list[SearchResult]]:
+    async def search_many(
+        self, queries: list[str], limit: int = 10, *, kind: str = "query"
+    ) -> list[list[SearchResult]]:
         """One embedder call for all queries; per-query SELECTs share a session."""
         if not queries:
             return []
-        q_vecs = await self._embedder.embed(list(queries))
+        q_vecs = await self._embedder.embed(list(queries), kind=kind)
 
         from sqlalchemy.sql import text as sa_text
 

packages/core/src/repowise/core/providers/embedding/base.py

@@ -66,6 +66,38 @@ def resolve_embedding_timeout(
     return selected
 
 
+_EMBED_PREFIX_ENV: dict[str, str] = {
+    "query": "REPOWISE_EMBED_QUERY_PREFIX",
+    "document": "REPOWISE_EMBED_DOC_PREFIX",
+}
+
+
+def resolve_embed_prefix(kind: str) -> str:
+    """The literal text to prepend before embedding a batch of *kind*.
+
+    Some embedding models (e.g. asymmetric-instruction models like
+    NVIDIA's Nemotron-3-Embed) are trained with a fixed ``"query: "`` /
+    ``"passage: "`` (or similar) framing and never apply it themselves — the
+    caller has to. Most models (OpenAI, Gemini, ...) want the raw text and
+    would be actively hurt by an uninvited prefix.
+
+    Reading ``REPOWISE_EMBED_QUERY_PREFIX`` / ``REPOWISE_EMBED_DOC_PREFIX``
+    keeps this opt-in and per-deployment: unset (the default) resolves to
+    ``""`` for both kinds, so an embedder that calls this is inert unless the
+    operator explicitly configures a model that needs it.
+
+    Args:
+        kind: ``"query"`` or ``"document"`` — which side of retrieval *texts*
+            are on. Anything else is a caller bug, not a misconfiguration, so
+            it raises rather than silently defaulting.
+    """
+    try:
+        var = _EMBED_PREFIX_ENV[kind]
+    except KeyError:
+        raise ValueError(f"kind must be 'query' or 'document', got {kind!r}") from None
+    return os.environ.get(var, "")
+
+
 def _report_invalid_timeout(var: str, raw: str, default: float) -> None:
     """Say it where the user will actually see it.
 
@@ -96,11 +128,17 @@ def dimensions(self) -> int:
         """Number of dimensions in the embedding vector."""
         ...
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
+    async def embed(self, texts: list[str], *, kind: str = "document") -> list[list[float]]:
         """Embed a batch of texts.
 
         Args:
             texts: Non-empty list of strings to embed.
+            kind: ``"query"`` or ``"document"`` — which side of retrieval
+                *texts* are on. Implementations that don't need the
+                distinction (most do not) accept and ignore it; it exists so
+                an asymmetric-instruction model can apply the right prefix
+                (see :func:`resolve_embed_prefix`) without every call site
+                needing to know which models care.
 
         Returns:
             List of unit-length float vectors, one per input string.
@@ -125,7 +163,8 @@ class MockEmbedder:
 
     dimensions: int = 8
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
+    async def embed(self, texts: list[str], *, kind: str = "document") -> list[list[float]]:
+        del kind  # deterministic hash is direction-agnostic
         results: list[list[float]] = []
         for text in texts:
             digest = hashlib.sha256(text.encode()).digest()

packages/core/src/repowise/core/providers/embedding/edenai.py

@@ -101,12 +101,18 @@
     def dimensions(self) -> int:
         return self._dimensions
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
+    async def embed(self, texts: list[str], *, kind: str = "document") -> list[list[float]]:
         """Embed a batch of texts using Eden AI.
 
         Runs the synchronous SDK call in a thread pool to avoid blocking the
         asyncio event loop.
+
+        Args:
+            texts: Non-empty list of strings to embed.
+            kind: Accepted for Embedder-protocol parity; unused. None of the
+                models in ``_DIMS`` are configured with a directional prefix.
         """
+        del kind
         if not texts:
             return []
 

packages/core/src/repowise/core/providers/embedding/gemini.py

@@ -79,18 +79,23 @@
     def dimensions(self) -> int:
         return self._output_dimensionality
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
+    async def embed(self, texts: list[str], *, kind: str = "document") -> list[list[float]]:
         """Embed a batch of texts using Gemini.
 
         Runs the synchronous SDK call in a thread pool to avoid blocking the
         asyncio event loop.
 
         Args:
             texts: Non-empty list of strings to embed.
+            kind: Accepted for Embedder-protocol parity with directional
+                embedders; Gemini's ``task_type`` already encodes
+                query-vs-document framing (see ``__init__``), so this is
+                ignored here rather than doubly applied.
 
         Returns:
             List of unit-length (L2-normalized) float vectors.
         """
+        del kind
         if not texts:
             return []
 

packages/core/src/repowise/core/providers/embedding/ollama.py

@@ -84,8 +84,16 @@
     def dimensions(self) -> int:
         return self._dimensions
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
-        """Embed a batch of texts using Ollama's native API."""
+    async def embed(self, texts: list[str], *, kind: str = "document") -> list[list[float]]:
+        """Embed a batch of texts using Ollama's native API.
+
+        Args:
+            texts: Non-empty list of strings to embed.
+            kind: Accepted for Embedder-protocol parity; unused. Ollama
+                embedding models exposed here have no configured directional
+                prefix, so nothing is applied.
+        """
+        del kind
         if not texts:
             return []