Merge feature/embeddings: embed feedback, errors, events and LLM text

EmbeddingSink queues text from each batch on its own bounded queue and
upserts vectors into a new {prefix}embeddings table (pgvector on Postgres,
JSON elsewhere). mw.similar() and mw.similar_feedback() search it.
submit_feedback records a feedback.submitted event; record_llm_call keeps
prompt and completion only with capture_llm_text=True.

CHANGELOG.md

@@ -5,6 +5,73 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Calendar Versioning](https://calver.org/) (YYYY.MM.DD).
 
+## [2026.09.30] - 2026-09-30
+
+Embed feedback, tool errors, event text and (opt-in) LLM text, and search
+them by meaning: "has anyone reported this before?"
+
+### Added
+- `EmbeddingSink(embedder, database_sink, *, sources=("feedback",
+  "call_error", "event", "llm"), event_text_keys=("error_excerpt", "error",
+  "message", "reason"), max_chars=4000, max_queue=1000, batch_size=32,
+  timeout=30.0, scan_limit=5000)`. `write` extracts text and puts it on the
+  sink's own bounded queue, returning at once, so a slow embedding backend
+  never reaches the dispatcher's `sink_timeout` or a tool call. A background
+  task embeds in batches and upserts through the `DatabaseSink`'s engine. A
+  full queue drops and counts (`dropped`); embedder and database failures are
+  logged and counted (`errors`), never raised. An unchanged text (same
+  `text_hash`) is not embedded again; a changed one replaces its row. Text is
+  redacted with the middleware's `Redactor` (adopted when the sink is passed
+  to `instrument`) and truncated before it leaves the process. `aclose()`
+  drains for up to `timeout`, then drops and counts the rest.
+- `Embedder` protocol (`model`, `dim`, `async embed(texts)`) and
+  `OpenAIEmbedder(base_url, api_key=None, model="mxbai-embed-large", dim=1024,
+  batch_size=32, timeout=30.0)` for any OpenAI-compatible `/v1/embeddings`
+  endpoint. It uses `httpx`, or `httpx2` under FastMCP 4; no new dependency.
+  A vector of the wrong length raises `ValueError`.
+- `await mw.similar(text, *, source_types=None, k=10, min_score=None)` and
+  `await mw.similar_feedback(feedback_ref, *, k=10, source_types=None,
+  min_score=None)`. Results carry `source_type`, `source_id`, `text`, `score`
+  (1 - cosine distance, clamped to [0, 1]) and `created_at`. PostgreSQL ranks
+  with `embedding <=> :q` on an HNSW index; other databases compare in Python
+  against the newest `scan_limit` rows. `similar_feedback` reuses the stored
+  vector and leaves the item itself out. Neither raises; failures log and
+  return [].
+- **Feedback flows as an event.** With `instrumentation=`, `submit_feedback`
+  records `feedback.submitted` keyed by the feedback id, with `type`, `title`,
+  `description` and `submitter`. Hosts with their own feedback tools record
+  the same event with their own ref (`bug-7Q2X`).
+- `InstrumentationMiddleware(capture_llm_text=False, max_llm_text_chars=None)`
+  and `record_llm_call(..., prompt=None, completion=None)`. Prompt and
+  completion are stored (redacted, cut to `8 * max_error_chars`, 4000 by
+  default) only when `capture_llm_text=True`.
+- `embeddings` extra: `pgvector>=0.5.0`. Postgres hosts also need `asyncpg`.
+- `build_metadata(prefix, metadata=None, embedding_dim=1024)` and
+  `DatabaseSink(embedding_dim=1024)`. `embedding_dim=None` in `build_metadata`
+  leaves the embeddings table out.
+
+### Changed
+- **New table `{prefix}embeddings`** (`id`, `source_type`, `source_id`,
+  `model`, `dim`, `text_hash`, `text`, `created_at`, `embedding`), unique on
+  (`source_type`, `source_id`, `model`). `embedding` is pgvector `vector(dim)`
+  with an HNSW `vector_cosine_ops` index on PostgreSQL, a JSON float list
+  elsewhere. `build_metadata()` now includes it, so Alembic hosts need a
+  migration; the README has the SQL, starting with
+  `CREATE EXTENSION IF NOT EXISTS vector`. With `create_tables=True` the table
+  is created when an `EmbeddingSink` first writes, not at startup, so hosts
+  that do not embed need neither pgvector nor the extension. On PostgreSQL
+  that step first runs `CREATE EXTENSION IF NOT EXISTS vector`; without the
+  privilege it logs a warning, keeps the sink running, and retries at most
+  once a minute.
+- `DatabaseSink.prune()` also deletes embeddings older than the cutoff,
+  except those of feedback, and returns the total from all three tables.
+- `record_llm_call` now drops `prompt` and `completion` keys passed in
+  `attrs` unless `capture_llm_text=True`. Earlier versions stored them.
+
+### Documentation
+- README: `DatabaseSink` needs the `instrumentation` extra even with
+  SQLAlchemy installed, since SQLAlchemy 2.1 no longer installs `greenlet`.
+
 ## [2026.09.29.1] - 2026-09-29
 
 ### Fixed

CONTRIBUTING.md

@@ -127,6 +127,20 @@ needs permission to create and drop databases:
 FFB_TEST_DATABASE_URL=postgresql+asyncpg://user:pass@host:5432/postgres uv run pytest -m postgres
 ```
 
+### Embedding tests against a real endpoint
+
+`tests/unit/test_embeddings.py` uses a deterministic fake embedder. One test
+embeds through a real OpenAI-compatible endpoint when you point it at one, and
+skips otherwise:
+
+```bash
+FFB_TEST_EMBED_URL=https://api.example.com/v1 FFB_TEST_EMBED_KEY=... \
+  uv run pytest tests/unit/test_embeddings.py -k live
+```
+
+`FFB_TEST_EMBED_MODEL` (default `mxbai-embed-large`) and `FFB_TEST_EMBED_DIM`
+(default 1024) pick the model.
+
 ### Writing Tests
 
 - Place unit tests in `tests/unit/`

Makefile

@@ -59,4 +59,5 @@ test-all: test test-pg
 
 lint:
 	uv run --no-sync ruff check src/ examples/ tests/_compat.py tests/_databases.py \
-		tests/unit/test_test_databases.py tests/unit/test_instrumentation.py
+		tests/unit/test_test_databases.py tests/unit/test_instrumentation.py \
+		tests/unit/test_embeddings.py

README.md

@@ -181,6 +181,8 @@ instrument(app, [
 
 `DatabaseSink` needs the `instrumentation` extra
 (`uv add "fastmcp-feedback[instrumentation]"`), plus `asyncpg` for Postgres.
+Install the extra even if you already have SQLAlchemy: SQLAlchemy 2.1 no longer
+pulls in `greenlet` by itself, and its async engine cannot run without it.
 
 ### Using your own database and migrations
 
@@ -506,6 +508,26 @@ on SQLite):
 Anything in `attrs` is kept alongside, but these names win. `key=` works as
 for `record_event`, for example a chat turn id.
 
+The prompt and the answer are not stored unless you opt in, since chat text
+can hold anything a user typed. With `capture_llm_text=True` on the
+middleware, `record_llm_call(..., prompt=..., completion=...)` stores them as
+`prompt` and `completion` attributes, redacted and cut to `max_llm_text_chars`
+(by default `8 * max_error_chars`, so 4000 characters):
+
+```python
+from fastmcp_feedback.instrumentation import instrument
+
+mw = instrument(app, capture_llm_text=True)
+mw.record_llm_call(
+    model="claude-opus-4",
+    prompt="Why does the render time out?",
+    completion="The scene has 40 million polygons.",
+)
+```
+
+With the default `capture_llm_text=False` both are discarded, and so are
+`prompt` and `completion` keys passed in `attrs`.
+
 #### Storage and sinks
 
 Events go to a `ffb_events` table: `id`, `occurred_at`, `kind`, `key`,
@@ -534,6 +556,157 @@ includes it). `create_tables=True` creates it on first use, since it creates
 missing tables. Until the table exists, event writes fail and those events
 are dropped (logged); tool calls and pruning are unaffected.
 
+### Embeddings: finding similar reports and errors
+
+"Has anyone reported this before?" is hard to answer with `LIKE`. An
+`EmbeddingSink` turns the text in the records into vectors and stores them, so
+you can search by meaning. It embeds four kinds of text:
+
+| `source_type` | Text | `source_id` |
+|---------------|------|-------------|
+| `feedback` | title and description of a `feedback.submitted` event | the feedback ref |
+| `call_error` | `"{tool}: {error_type}: {error_message}"` of calls with outcome `error` or `soft_error` | the call id |
+| `event` | the values of `error_excerpt`, `error`, `message`, `reason` in an event's attrs (`event_text_keys=`) | the event id |
+| `llm` | `"prompt: ...\ncompletion: ..."` of an `llm.call` event, when `capture_llm_text=True` | the event id |
+
+An `llm.call` without prompt or completion counts as a plain event, so a
+failed LLM call's `error` is still embedded.
+
+It needs a `DatabaseSink` to store into and an embedder. `OpenAIEmbedder`
+talks to any OpenAI-compatible `/v1/embeddings` endpoint (OpenAI, Ollama,
+LiteLLM, vLLM):
+
+```python
+import os
+
+from fastmcp_feedback import add_feedback_tools
+from fastmcp_feedback.instrumentation import (
+    DatabaseSink,
+    EmbeddingSink,
+    OpenAIEmbedder,
+    instrument,
+)
+
+embedder = OpenAIEmbedder(
+    os.environ["EMBED_URL"],               # e.g. https://api.example.com/v1
+    api_key=os.environ["EMBED_API_KEY"],
+    model="mxbai-embed-large",
+    dim=1024,
+)
+db = DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True, embedding_dim=1024)
+mw = instrument(app, [db, EmbeddingSink(embedder, db)])
+add_feedback_tools(app, database_url="sqlite:///feedback.db", instrumentation=mw)
+```
+
+Then search:
+
+```python
+for hit in await mw.similar("render timed out", k=5):
+    print(f"{hit['score']:.2f}", hit["source_type"], hit["source_id"], hit["text"][:60])
+
+# Reports and errors like feedback 42, leaving 42 itself out.
+related = await mw.similar_feedback("42", source_types=["feedback", "call_error"])
+```
+
+Each hit has `source_type`, `source_id`, `text` (what was embedded),
+`score` and `created_at`. `score` is 1 minus the cosine distance, clamped to
+[0, 1]: 1 is the same direction, 0 unrelated or opposite. `min_score=` drops
+weaker hits, and `source_types=` limits the search. Only vectors from the
+embedder's `model` are compared. `similar_feedback` uses the stored vector of
+that feedback, so it embeds nothing, and returns [] until the feedback has
+been embedded. Neither ever raises; on any failure they log and return [].
+
+On PostgreSQL the search runs in the database (`ORDER BY embedding <=> :q`)
+on an HNSW index. pgvector applies `source_types` and the model filter after
+the index scan, so a narrow filter over a large table can return fewer than
+`k` hits; `SET hnsw.iterative_scan = relaxed_order` (pgvector 0.8) fixes that.
+Elsewhere it compares in Python against the newest `scan_limit` (5000) rows,
+and logs once when there are more.
+
+**It stays off the call path.** `write` only picks out the text and puts it on
+the sink's own bounded queue (`max_queue`, 1000), so a slow embedding backend
+never holds up the dispatcher or a tool call. A background task embeds up to
+`batch_size` (32) texts per request, each request limited to `timeout` (30 s),
+and upserts the rows. A full queue drops texts and counts them in
+`sink.dropped`; failures of the embedder or the database are logged and
+counted in `sink.errors`, never raised. A text whose row already holds the
+same text (by `text_hash`) is not embedded again; a changed text replaces its
+row. `await sink.flush()` waits for the queue, and `mw.aclose()` lets it drain
+for up to `timeout` before stopping.
+
+**Nothing new leaves the process unredacted.** Text is redacted with the
+middleware's `Redactor` (the sink adopts it when passed to `instrument`) and
+cut to `max_chars` (4000) before it is sent to the embedder or stored, on top
+of the redaction every record already had.
+
+**Feedback flows as an event.** With `instrumentation=mw`, `submit_feedback`
+records a `feedback.submitted` event keyed by the feedback id, with `type`,
+`title`, `description` and `submitter` attributes. If you have your own
+feedback tools, record the same event with your own ref and it is embedded
+the same way:
+
+```python
+mw.record_event(
+    "feedback.submitted",
+    key="bug-7Q2X",
+    attrs={"type": "bug", "title": "Render hangs", "description": "Stuck at 99%"},
+)
+```
+
+**Storage.** Vectors go to `ffb_embeddings`: `id`, `source_type`,
+`source_id`, `model`, `dim`, `text_hash` (sha256 of `text`), `text`,
+`created_at` and `embedding`, unique on (`source_type`, `source_id`,
+`model`). `embedding` is a pgvector `vector(dim)` on PostgreSQL and a JSON
+list of floats elsewhere. `DatabaseSink(embedding_dim=...)` sets `dim` and must
+equal the embedder's `dim`, or `EmbeddingSink` raises `ValueError`. Retention
+covers it: `prune()` deletes embeddings older than the cutoff, **except those
+of feedback**, which are kept like the calls linked to feedback.
+
+On PostgreSQL this needs the `embeddings` extra (`pgvector`) and `asyncpg`:
+
+```bash
+uv add "fastmcp-feedback[instrumentation,embeddings]" asyncpg
+```
+
+With `create_tables=True`, the table is created when an `EmbeddingSink` first
+writes, not before, so hosts that do not embed never need pgvector. On
+PostgreSQL it first runs `CREATE EXTENSION IF NOT EXISTS vector`. Creating an
+extension needs a superuser or the database owner on most setups; without
+that it logs a warning, and so does the table creation if the extension is
+still missing. The sink keeps working and counts the failed writes, and
+retries the creation at most once a minute.
+
+**Upgrading:** `ffb_embeddings` is new in 2026.09.30. `build_metadata()`
+includes it (pass `embedding_dim=None` to leave it out). Hosts on Alembic need
+a migration; for PostgreSQL, with the default prefix and 1024 dimensions:
+
+```sql
+CREATE EXTENSION IF NOT EXISTS vector;
+
+CREATE TABLE ffb_embeddings (
+    id VARCHAR(36) NOT NULL,
+    source_type VARCHAR(16) NOT NULL,
+    source_id VARCHAR(255) NOT NULL,
+    model VARCHAR(128) NOT NULL,
+    dim INTEGER NOT NULL,
+    text_hash VARCHAR(64) NOT NULL,
+    text TEXT NOT NULL,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    embedding VECTOR(1024) NOT NULL,
+    PRIMARY KEY (id),
+    CONSTRAINT ffb_embeddings_source_uq UNIQUE (source_type, source_id, model)
+);
+CREATE INDEX ffb_embeddings_hnsw ON ffb_embeddings USING hnsw (embedding vector_cosine_ops);
+CREATE INDEX ix_ffb_embeddings_created_at ON ffb_embeddings (created_at);
+CREATE INDEX ix_ffb_embeddings_source_id ON ffb_embeddings (source_id);
+CREATE INDEX ix_ffb_embeddings_source_type ON ffb_embeddings (source_type);
+```
+
+Wrap it in `op.execute(...)` in the migration's `upgrade()`, and
+`DROP TABLE ffb_embeddings` in `downgrade()`. Autogenerate also works but
+renders the column as `fastmcp_feedback.instrumentation.storage.EmbeddingVector(dim=1024)`,
+which ties the migration to this package.
+
 ### Middleware order and shutdown
 
 FastMCP runs the first middleware added as the outermost. Call `instrument()`

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "fastmcp-feedback"
-version = "2026.09.29.1"  # YYYYMMDD.patch versioning - update for each release
+version = "2026.09.30"  # YYYYMMDD.patch versioning - update for each release
 description = "Production-ready feedback collection system for FastMCP servers"
 readme = "README.md"
 license = "MIT"
@@ -52,6 +52,8 @@ dependencies = [
 postgresql = ["psycopg2-binary>=2.9.0"]
 # Async database sink for per-call instrumentation. Postgres hosts add asyncpg.
 instrumentation = ["sqlalchemy[asyncio]>=2.0.43", "aiosqlite>=0.22.1"]
+# Vector column for EmbeddingSink on Postgres. Postgres hosts add asyncpg.
+embeddings = ["pgvector>=0.5.0"]
 mysql = ["pymysql>=1.1.0"]
 dev = [
     "sqlalchemy[asyncio]>=2.0.43",

src/fastmcp_feedback/__init__.py

@@ -44,7 +44,7 @@ Quick Start:
 import logging
 
 # Version information - CalVer YYYY.MM.DD
-__version__ = "2026.09.29.1"
+__version__ = "2026.09.30"
 __author__ = "Ryan Malloy"
 __email__ = "ryan@supported.systems"
 

src/fastmcp_feedback/instrumentation/__init__.py

@@ -6,6 +6,10 @@ mw = instrument(app)  # every tool call as JSON lines on stderr
 mw = instrument(app, [DatabaseSink(engine)], identity_resolver=who)
 mw = instrument(app, result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)))
 mw.record_event("job.finished", key=job_id, attrs={"duration_s": 3600})
+
+db = DatabaseSink(engine)
+mw = instrument(app, [db, EmbeddingSink(OpenAIEmbedder(url, key), db)])
+await mw.similar("render timed out")
 """
 
 from .classify import (
@@ -14,6 +18,14 @@ from .classify import (
     soft_error_classifier,
 )
 from .dispatch import RecordDispatcher
+from .embeddings import (
+    DEFAULT_EVENT_TEXT_KEYS,
+    Embedder,
+    EmbeddingSink,
+    EmbeddingText,
+    OpenAIEmbedder,
+    extract_texts,
+)
 from .events import EventRecord
 from .middleware import MODE_ENV_VAR, MODES, InstrumentationMiddleware, instrument
 from .record import ToolCallRecord
@@ -24,21 +36,36 @@ from .redaction import (
     Redactor,
 )
 from .sinks import CallbackSink, JsonLinesSink, MemorySink, Sink
-from .storage import DEFAULT_TABLE_PREFIX, DatabaseSink, build_metadata
+from .storage import (
+    DEFAULT_EMBEDDING_DIM,
+    DEFAULT_TABLE_PREFIX,
+    EMBEDDING_SOURCES,
+    DatabaseSink,
+    EmbeddingVector,
+    build_metadata,
+)
 
 __all__ = [
     "CallbackSink",
     "DatabaseSink",
+    "DEFAULT_EMBEDDING_DIM",
+    "DEFAULT_EVENT_TEXT_KEYS",
     "DEFAULT_RESULT_CLASSIFIER",
     "DEFAULT_SECRET_KEYS",
     "DEFAULT_SECRET_PATTERNS",
     "DEFAULT_TABLE_PREFIX",
+    "EMBEDDING_SOURCES",
+    "Embedder",
+    "EmbeddingSink",
+    "EmbeddingText",
+    "EmbeddingVector",
     "EventRecord",
     "InstrumentationMiddleware",
     "JsonLinesSink",
     "MemorySink",
     "MODE_ENV_VAR",
     "MODES",
+    "OpenAIEmbedder",
     "RecordDispatcher",
     "REDACTED",
     "Redactor",
@@ -46,6 +73,7 @@ __all__ = [
     "Sink",
     "ToolCallRecord",
     "build_metadata",
+    "extract_texts",
     "instrument",
     "soft_error_classifier",
 ]

src/fastmcp_feedback/instrumentation/embeddings.py

@@ -0,0 +1,614 @@
+"""Embed feedback, tool errors, event text and LLM text for similarity search.
+
+EmbeddingSink sits next to a DatabaseSink. It picks text out of each batch,
+queues it on its own bounded queue and returns at once, so a slow embedding
+backend never holds up the dispatcher. A background task embeds and stores
+the vectors in ``{prefix}embeddings``.
+"""
+
+import asyncio
+import contextlib
+import hashlib
+import logging
+import math
+import operator
+import uuid
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Protocol, runtime_checkable
+
+from sqlalchemy import Float, and_, literal, not_, select, tuple_
+
+from .events import EventRecord
+from .record import ToolCallRecord
+from .redaction import Redactor
+from .sinks import Record
+from .storage import EMBEDDING_SOURCES, EmbeddingVector
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_EVENT_TEXT_KEYS = ("error_excerpt", "error", "message", "reason")
+
+
+@runtime_checkable
+class Embedder(Protocol):
+    """Turns texts into vectors of length ``dim``, one per text, in order."""
+
+    model: str
+    dim: int
+
+    async def embed(self, texts: list[str]) -> list[list[float]]: ...
+
+
+def _http_module() -> Any:
+    # FastMCP 2.x and 3.x depend on httpx, FastMCP 4 on its successor httpx2.
+    try:
+        import httpx
+
+        return httpx
+    except ImportError:
+        pass
+    try:
+        import httpx2
+
+        return httpx2
+    except ImportError as exc:
+        raise ImportError("OpenAIEmbedder needs httpx or httpx2") from exc
+
+
+class OpenAIEmbedder:
+    """An OpenAI-compatible ``POST {base_url}/embeddings`` endpoint.
+
+    Args:
+        base_url: up to and including the version, e.g.
+            ``https://api.example.com/v1``.
+        api_key: sent as ``Authorization: Bearer ...`` when given.
+        model: the ``model`` field of each request.
+        dim: expected vector length. A response with any other length raises
+            ValueError, since it cannot be stored or compared.
+        batch_size: texts per request.
+        timeout: seconds per request.
+        transport: passed to the HTTP client. For tests.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        api_key: str | None = None,
+        model: str = "mxbai-embed-large",
+        dim: int = 1024,
+        batch_size: int = 32,
+        timeout: float = 30.0,
+        transport: Any = None,
+    ):
+        if batch_size < 1:
+            raise ValueError(f"batch_size must be at least 1, got {batch_size!r}")
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.model = model
+        self.dim = dim
+        self.batch_size = batch_size
+        self.timeout = timeout
+        self.transport = transport
+
+    async def embed(self, texts: list[str]) -> list[list[float]]:
+        if not texts:
+            return []
+        http = _http_module()
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["Authorization"] = f"Bearer {self.api_key}"
+        vectors: list[list[float]] = []
+        kwargs: dict[str, Any] = {"timeout": self.timeout}
+        if self.transport is not None:
+            kwargs["transport"] = self.transport
+        async with http.AsyncClient(**kwargs) as client:
+            for start in range(0, len(texts), self.batch_size):
+                chunk = texts[start : start + self.batch_size]
+                response = await client.post(
+                    f"{self.base_url}/embeddings",
+                    json={"model": self.model, "input": chunk},
+                    headers=headers,
+                )
+                response.raise_for_status()
+                data = sorted(
+                    response.json()["data"], key=lambda item: item.get("index", 0)
+                )
+                if len(data) != len(chunk):
+                    raise ValueError(
+                        f"asked for {len(chunk)} embeddings, got {len(data)}"
+                    )
+                for item in data:
+                    vector = [float(v) for v in item["embedding"]]
+                    if len(vector) != self.dim:
+                        raise ValueError(
+                            f"{self.model} returned a {len(vector)}-dim vector, "
+                            f"expected {self.dim}"
+                        )
+                    vectors.append(vector)
+        return vectors
+
+
+@dataclass
+class EmbeddingText:
+    """One text waiting to be embedded, and where it came from."""
+
+    source_type: str
+    source_id: str
+    text: str
+
+    @property
+    def text_hash(self) -> str:
+        return hashlib.sha256(self.text.encode()).hexdigest()
+
+
+def _clean(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = value if isinstance(value, str) else str(value)
+    text = text.strip()
+    return text or None
+
+
+def extract_texts(
+    records: Iterable[Record],
+    *,
+    sources: Iterable[str] = EMBEDDING_SOURCES,
+    event_text_keys: Sequence[str] = DEFAULT_EVENT_TEXT_KEYS,
+) -> list[EmbeddingText]:
+    """The texts worth embedding in a batch of records, unredacted and untruncated.
+
+    ``call_error``: calls with outcome ``error`` or ``soft_error``, as
+    ``"{tool}: {error_type}: {error_message}"``, keyed by the call id.
+    ``feedback``: ``feedback.submitted`` events, title and description,
+    keyed by the event's ``key`` (the feedback ref).
+    ``llm``: ``llm.call`` events with ``prompt`` or ``completion`` attrs.
+    ``event``: any other event with one of ``event_text_keys`` in its attrs,
+    those values joined by newlines, keyed by the event id. An ``llm.call``
+    without prompt or completion counts as a plain event, so a failed LLM
+    call's ``error`` is still embedded.
+    """
+    wanted = frozenset(sources)
+    out: list[EmbeddingText] = []
+    for record in records:
+        if isinstance(record, ToolCallRecord):
+            if "call_error" in wanted and record.outcome in ("error", "soft_error"):
+                parts = [record.tool, record.error_type, _clean(record.error_message)]
+                out.append(
+                    EmbeddingText(
+                        "call_error", record.id, ": ".join(p for p in parts if p)
+                    )
+                )
+            continue
+        if not isinstance(record, EventRecord):
+            continue
+        attrs = record.attrs or {}
+        if record.kind == "feedback.submitted":
+            if "feedback" in wanted:
+                body = "\n".join(
+                    t
+                    for t in (
+                        _clean(attrs.get("title")),
+                        _clean(attrs.get("description")),
+                    )
+                    if t
+                )
+                if body:
+                    out.append(EmbeddingText("feedback", record.key or record.id, body))
+            continue
+        if record.kind == "llm.call":
+            prompt, completion = (
+                _clean(attrs.get("prompt")),
+                _clean(attrs.get("completion")),
+            )
+            if prompt or completion:
+                if "llm" in wanted:
+                    lines = []
+                    if prompt:
+                        lines.append(f"prompt: {prompt}")
+                    if completion:
+                        lines.append(f"completion: {completion}")
+                    out.append(EmbeddingText("llm", record.id, "\n".join(lines)))
+                continue
+        if "event" in wanted:
+            values = [_clean(attrs.get(k)) for k in event_text_keys]
+            body = "\n".join(v for v in values if v)
+            if body:
+                out.append(EmbeddingText("event", record.id, body))
+    return out
+
+
+def _dot(a: Sequence[float], b: Sequence[float]) -> float:
+    sumprod = getattr(math, "sumprod", None)  # 3.12+
+    if sumprod is not None:
+        return sumprod(a, b)
+    return sum(map(operator.mul, a, b))
+
+
+def cosine_score(a: Sequence[float], b: Sequence[float]) -> float | None:
+    """1 - cosine distance, clamped to [0, 1]. None for a zero vector."""
+    na, nb = math.sqrt(_dot(a, a)), math.sqrt(_dot(b, b))
+    if not na or not nb:
+        return None
+    return _clamp(_dot(a, b) / (na * nb))
+
+
+def _clamp(score: float) -> float:
+    return min(1.0, max(0.0, score))
+
+
+class EmbeddingSink:
+    """Embed text from records and store the vectors through a DatabaseSink.
+
+    ``write`` only queues, so it returns at once whatever the embedder is
+    doing. A background task embeds up to ``batch_size`` texts at a time and
+    upserts them into ``{prefix}embeddings`` on ``database_sink.engine``. A
+    full queue drops the text and counts it in ``dropped``; embedder and
+    database failures are logged and counted in ``errors``, never raised.
+    A text whose (source_type, source_id, model) row already holds the same
+    text is not embedded again; a changed text replaces the row.
+
+    Text is redacted with ``redactor`` (the middleware's, when this sink is
+    passed to ``instrument``) and cut to ``max_chars`` before it is embedded
+    or stored.
+
+    Args:
+        embedder: anything with ``model``, ``dim`` and ``async embed(texts)``.
+        database_sink: the DatabaseSink whose engine and tables to use. Its
+            ``embedding_dim`` must equal ``embedder.dim``.
+        sources: which of ``feedback``, ``call_error``, ``event`` and ``llm``
+            to embed.
+        event_text_keys: attrs whose values make up an ``event`` text.
+        max_chars: longest text embedded.
+        max_queue: texts waiting to be embedded before new ones are dropped.
+        batch_size: texts per embed call.
+        timeout: seconds one embed call may take, and how long ``aclose``
+            waits for the queue to drain.
+        scan_limit: without pgvector, ``similar`` compares against the newest
+            this many rows.
+    """
+
+    def __init__(
+        self,
+        embedder: Embedder,
+        database_sink: Any,
+        *,
+        sources: Iterable[str] = EMBEDDING_SOURCES,
+        event_text_keys: Sequence[str] = DEFAULT_EVENT_TEXT_KEYS,
+        max_chars: int = 4000,
+        max_queue: int = 1000,
+        batch_size: int = 32,
+        timeout: float = 30.0,
+        scan_limit: int = 5000,
+        redactor: Redactor | None = None,
+    ):
+        self.sources = frozenset(sources)
+        unknown = self.sources - set(EMBEDDING_SOURCES)
+        if unknown:
+            raise ValueError(
+                f"unknown sources {sorted(unknown)}; choose from {EMBEDDING_SOURCES}"
+            )
+        if embedder.dim != database_sink.embedding_dim:
+            raise ValueError(
+                f"embedder {embedder.model!r} makes {embedder.dim}-dim vectors but "
+                f"the DatabaseSink's embeddings table holds "
+                f"{database_sink.embedding_dim}; pass embedding_dim={embedder.dim} "
+                "to DatabaseSink"
+            )
+        if max_chars < 1 or max_queue < 1 or batch_size < 1 or scan_limit < 1:
+            raise ValueError(
+                "max_chars, max_queue, batch_size and scan_limit must be positive"
+            )
+        self.embedder = embedder
+        self.database_sink = database_sink
+        self.table = database_sink.embeddings
+        self.event_text_keys = tuple(event_text_keys)
+        self.max_chars = max_chars
+        self.max_queue = max_queue
+        self.batch_size = batch_size
+        self.timeout = timeout
+        self.scan_limit = scan_limit
+        self.redactor = redactor
+        self._default_redactor = Redactor()
+        self.dropped = 0
+        self.errors = 0
+        self.embedded = 0
+        self.skipped = 0
+        self._queue: asyncio.Queue[EmbeddingText] | None = None
+        self._task: asyncio.Task | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._warned_scan = False
+
+    # Writing
+
+    async def write(self, records: Sequence[Record]) -> None:
+        try:
+            items = extract_texts(
+                records, sources=self.sources, event_text_keys=self.event_text_keys
+            )
+        except Exception:
+            self.errors += 1
+            logger.warning("Failed to extract texts to embed", exc_info=True)
+            return
+        for item in items:
+            self._enqueue(self._prepare(item))
+
+    def _prepare(self, item: EmbeddingText) -> EmbeddingText:
+        redactor = self.redactor or self._default_redactor
+        item.text = redactor.redact_text(item.text)[: self.max_chars]
+        return item
+
+    def _enqueue(self, item: EmbeddingText) -> None:
+        loop = asyncio.get_running_loop()
+        if self._queue is None or self._loop is not loop:
+            self._loop = loop
+            self._queue = asyncio.Queue(maxsize=self.max_queue)
+            self._task = loop.create_task(self._run(self._queue))
+        try:
+            self._queue.put_nowait(item)
+        except asyncio.QueueFull:
+            self.dropped += 1
+            if self.dropped == 1 or self.dropped % 1000 == 0:
+                logger.warning(
+                    "Embedding queue full; %d texts dropped so far", self.dropped
+                )
+
+    async def _run(self, queue: asyncio.Queue) -> None:
+        while True:
+            batch = [await queue.get()]
+            while len(batch) < self.batch_size:
+                try:
+                    batch.append(queue.get_nowait())
+                except asyncio.QueueEmpty:
+                    break
+            try:
+                await self._process(batch)
+            except Exception:
+                self.errors += 1
+                logger.warning(
+                    "Failed to embed %d texts with %s",
+                    len(batch),
+                    self.embedder.model,
+                    exc_info=True,
+                )
+            finally:
+                for _ in batch:
+                    queue.task_done()
+
+    async def _process(self, batch: list[EmbeddingText]) -> None:
+        # The last text for a source wins within a batch.
+        latest = {(i.source_type, i.source_id): i for i in batch}
+        items = list(latest.values())
+        db = self.database_sink
+        await db._ready()
+        await db.ensure_embeddings()
+        t = self.table
+        model = self.embedder.model
+        async with db.engine.connect() as conn:
+            rows = (
+                await conn.execute(
+                    select(t.c.source_type, t.c.source_id, t.c.text_hash).where(
+                        t.c.model == model,
+                        tuple_(t.c.source_type, t.c.source_id).in_(list(latest)),
+                    )
+                )
+            ).all()
+        stored = {(r.source_type, r.source_id): r.text_hash for r in rows}
+        todo = [
+            i for i in items if stored.get((i.source_type, i.source_id)) != i.text_hash
+        ]
+        self.skipped += len(batch) - len(todo)
+        if not todo:
+            return
+        vectors = await asyncio.wait_for(
+            self.embedder.embed([i.text for i in todo]), self.timeout
+        )
+        if len(vectors) != len(todo):
+            raise ValueError(
+                f"embedder returned {len(vectors)} vectors for {len(todo)}"
+            )
+        for vector in vectors:
+            if len(vector) != db.embedding_dim:
+                raise ValueError(
+                    f"embedder returned a {len(vector)}-dim vector, "
+                    f"expected {db.embedding_dim}"
+                )
+        now = datetime.now(UTC)
+        values = [
+            {
+                "id": str(uuid.uuid4()),
+                "source_type": i.source_type,
+                "source_id": i.source_id,
+                "model": model,
+                "dim": len(v),
+                "text_hash": i.text_hash,
+                "text": i.text,
+                "created_at": now,
+                "embedding": [float(x) for x in v],
+            }
+            for i, v in zip(todo, vectors, strict=True)
+        ]
+        await self._upsert(values, replace=set(stored))
+        self.embedded += len(values)
+
+    async def _upsert(self, values: list[dict[str, Any]], replace: set) -> None:
+        t = self.table
+        engine = self.database_sink.engine
+        dialect = engine.dialect.name
+        changing = ("text_hash", "text", "created_at", "embedding", "dim")
+        if dialect in ("postgresql", "sqlite"):
+            if dialect == "postgresql":
+                from sqlalchemy.dialects.postgresql import insert
+            else:
+                from sqlalchemy.dialects.sqlite import insert
+            stmt = insert(t)
+            stmt = stmt.on_conflict_do_update(
+                index_elements=[t.c.source_type, t.c.source_id, t.c.model],
+                set_={c: stmt.excluded[c] for c in changing},
+            )
+            async with engine.begin() as conn:
+                await conn.execute(stmt, values)
+            return
+        async with engine.begin() as conn:
+            for row in values:
+                key = (row["source_type"], row["source_id"])
+                if key in replace:
+                    await conn.execute(
+                        t.update()
+                        .where(
+                            t.c.source_type == row["source_type"],
+                            t.c.source_id == row["source_id"],
+                            t.c.model == row["model"],
+                        )
+                        .values({c: row[c] for c in changing})
+                    )
+                else:
+                    await conn.execute(t.insert(), [row])
+
+    async def flush(self) -> None:
+        """Wait until every queued text has been embedded (or failed)."""
+        if self._queue is not None and self._loop is asyncio.get_running_loop():
+            await self._queue.join()
+
+    async def aclose(self) -> None:
+        """Drain the queue for up to ``timeout`` seconds, then stop.
+
+        Texts still queued after that are dropped and counted. Disposes the
+        DatabaseSink's engine again if it owns it, since the dispatcher may
+        have closed that sink first.
+        """
+        if self._queue is not None and self._loop is asyncio.get_running_loop():
+            try:
+                await asyncio.wait_for(self._queue.join(), self.timeout)
+            except TimeoutError:
+                left = self._queue.qsize()
+                self.dropped += left
+                logger.warning(
+                    "Embedding queue did not drain in %ss; %d texts dropped",
+                    self.timeout,
+                    left,
+                )
+        if self._task is not None:
+            self._task.cancel()
+            with contextlib.suppress(asyncio.CancelledError, RuntimeError):
+                await self._task
+            self._task = None
+            self._queue = None
+        if getattr(self.database_sink, "_owns_engine", False):
+            await self.database_sink.engine.dispose()
+
+    # Searching
+
+    async def similar(
+        self,
+        text: str,
+        *,
+        source_types: Iterable[str] | None = None,
+        k: int = 10,
+        min_score: float | None = None,
+    ) -> list[dict[str, Any]]:
+        """Stored texts closest to ``text``. Raises on failure; see the middleware."""
+        text = (self.redactor or self._default_redactor).redact_text(text)[
+            : self.max_chars
+        ]
+        [vector] = await asyncio.wait_for(self.embedder.embed([text]), self.timeout)
+        return await self.nearest(
+            vector, source_types=source_types, k=k, min_score=min_score
+        )
+
+    async def vector_for(self, source_type: str, source_id: str) -> list[float] | None:
+        """The stored vector for a source under this embedder's model, if any."""
+        t = self.table
+        async with self.database_sink.engine.connect() as conn:
+            row = (
+                await conn.execute(
+                    select(t.c.embedding).where(
+                        t.c.source_type == source_type,
+                        t.c.source_id == str(source_id),
+                        t.c.model == self.embedder.model,
+                    )
+                )
+            ).first()
+        return None if row is None else list(row.embedding)
+
+    async def nearest(
+        self,
+        vector: Sequence[float],
+        *,
+        source_types: Iterable[str] | None = None,
+        k: int = 10,
+        min_score: float | None = None,
+        exclude: tuple[str, str] | None = None,
+    ) -> list[dict[str, Any]]:
+        """Rows closest to ``vector`` by cosine, best first, as dicts with
+        ``source_type``, ``source_id``, ``text``, ``score`` and ``created_at``.
+
+        ``score`` is 1 - cosine distance, clamped to [0, 1]. Only rows made by
+        this embedder's model are compared. ``exclude`` is a
+        (source_type, source_id) left out of the results.
+        """
+        if k < 1:
+            return []
+        t = self.table
+        conditions = [t.c.model == self.embedder.model]
+        if source_types is not None:
+            conditions.append(t.c.source_type.in_(list(source_types)))
+        if exclude is not None:
+            conditions.append(
+                not_(and_(t.c.source_type == exclude[0], t.c.source_id == exclude[1]))
+            )
+        engine = self.database_sink.engine
+        cols = (t.c.source_type, t.c.source_id, t.c.text, t.c.created_at)
+        results: list[dict[str, Any]] = []
+        if engine.dialect.name == "postgresql":
+            distance = t.c.embedding.op("<=>", return_type=Float)(
+                literal(list(vector), type_=EmbeddingVector(len(vector)))
+            )
+            query = (
+                select(*cols, distance.label("distance"))
+                .where(*conditions)
+                .order_by(distance)
+                .limit(k)
+            )
+            async with engine.connect() as conn:
+                for row in (await conn.execute(query)).mappings():
+                    row = dict(row)
+                    distance_value = row.pop("distance")
+                    if distance_value is None or math.isnan(distance_value):
+                        continue
+                    row["score"] = _clamp(1.0 - distance_value)
+                    results.append(row)
+        else:
+            query = (
+                select(*cols, t.c.embedding)
+                .where(*conditions)
+                .order_by(t.c.created_at.desc())
+                .limit(self.scan_limit + 1)
+            )
+            async with engine.connect() as conn:
+                rows = [dict(r) for r in (await conn.execute(query)).mappings()]
+            if len(rows) > self.scan_limit:
+                rows = rows[: self.scan_limit]
+                if not self._warned_scan:
+                    self._warned_scan = True
+                    logger.info(
+                        "Similarity search without pgvector compares only the "
+                        "newest %d embeddings (scan_limit)",
+                        self.scan_limit,
+                    )
+            for row in rows:
+                score = cosine_score(vector, row.pop("embedding"))
+                if score is None:
+                    continue
+                row["score"] = score
+                results.append(row)
+            results.sort(key=lambda r: r["score"], reverse=True)
+            results = results[:k]
+        if min_score is not None:
+            results = [r for r in results if r["score"] >= min_score]
+        for row in results:
+            created = row["created_at"]
+            if created is not None and created.tzinfo is None:
+                row["created_at"] = created.replace(tzinfo=UTC)
+        return results

src/fastmcp_feedback/instrumentation/middleware.py

@@ -27,6 +27,7 @@ from .correlation import (
     select_calls,
 )
 from .dispatch import RecordDispatcher
+from .embeddings import EmbeddingSink
 from .events import (
     MAX_KEY_CHARS,
     MAX_KIND_CHARS,
@@ -161,6 +162,12 @@ class InstrumentationMiddleware(Middleware):
             when it returns less than the tool's rate. For tests.
         recent_events: how many recent events to keep in memory for
             ``events_for``, including ones still waiting to reach the sinks.
+        capture_llm_text: store ``prompt`` and ``completion`` given to
+            ``record_llm_call``. Off by default, since chat text can hold
+            anything a user typed; when off they are discarded, including
+            ``prompt``/``completion`` keys passed in its ``attrs``.
+        max_llm_text_chars: longest prompt or completion kept. Defaults to
+            ``8 * max_error_chars`` (4000).
 
     Exclusion and sampling apply to tool calls only; events are always
     recorded. A sampled tool's call is kept whenever an event was recorded
@@ -189,6 +196,8 @@ class InstrumentationMiddleware(Middleware):
         sample_rates: Mapping[str, float] | None = None,
         random: Callable[[], float] = _random.random,
         recent_events: int = 2000,
+        capture_llm_text: bool = False,
+        max_llm_text_chars: int | None = None,
     ):
         mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
         if mode not in MODES:
@@ -219,8 +228,19 @@ class InstrumentationMiddleware(Middleware):
         self._call_scope: contextvars.ContextVar[CallScope | None] = (
             contextvars.ContextVar(f"ffb_call_scope_{id(self):x}", default=None)
         )
+        self.capture_llm_text = capture_llm_text
+        self.max_llm_text_chars = (
+            max_llm_text_chars
+            if max_llm_text_chars is not None
+            else 8 * max_error_chars
+        )
+        sinks = list(sinks) if sinks is not None else [JsonLinesSink()]
+        for sink in sinks:
+            # Embedded text gets this middleware's redaction rules too.
+            if isinstance(sink, EmbeddingSink) and sink.redactor is None:
+                sink.redactor = self.redactor
         self.dispatcher = RecordDispatcher(
-            sinks if sinks is not None else [JsonLinesSink()],
+            sinks,
             max_queue=max_queue,
             batch_size=batch_size,
             sink_timeout=sink_timeout,
@@ -469,6 +489,8 @@ class InstrumentationMiddleware(Middleware):
         error: str | BaseException | None = None,
         key: Any = None,
         attrs: Mapping[str, Any] | None = None,
+        prompt: str | None = None,
+        completion: str | None = None,
     ) -> bool:
         """Record one LLM request as an ``llm.call`` event. Never raises.
 
@@ -477,9 +499,22 @@ class InstrumentationMiddleware(Middleware):
         ``error`` (redacted, truncated to ``max_error_chars``) plus
         ``error_type`` when ``error`` is an exception. ``ok`` defaults to
         ``error is None``. Extra ``attrs`` are kept, but these names win.
+
+        ``prompt`` and ``completion`` are stored, redacted and truncated to
+        ``max_llm_text_chars``, only with ``capture_llm_text=True``;
+        otherwise they are discarded.
         """
         try:
             fields: dict[str, Any] = dict(attrs or {})
+            fields.pop("prompt", None)
+            fields.pop("completion", None)
+            if self.capture_llm_text:
+                for name, value in (("prompt", prompt), ("completion", completion)):
+                    if value is None and attrs is not None:
+                        value = attrs.get(name)
+                    if value is not None:
+                        text = self.redactor.redact_text(str(value))
+                        fields[name] = text[: self.max_llm_text_chars]
             fields["model"] = None if model is None else str(model)
             for name, value in (
                 ("provider", provider),
@@ -533,6 +568,77 @@ class InstrumentationMiddleware(Middleware):
         rows.sort(key=lambda r: r["occurred_at"])
         return rows[-limit:] if limit > 0 else []
 
+    @property
+    def embedding_sink(self) -> EmbeddingSink | None:
+        """The first EmbeddingSink among the sinks, if any."""
+        return next(
+            (s for s in self.dispatcher.sinks if isinstance(s, EmbeddingSink)), None
+        )
+
+    async def similar(
+        self,
+        text: str,
+        *,
+        source_types: Iterable[str] | None = None,
+        k: int = 10,
+        min_score: float | None = None,
+    ) -> list[dict[str, Any]]:
+        """Embedded texts most like ``text``, best first. Never raises.
+
+        Needs an EmbeddingSink. Each result has ``source_type``,
+        ``source_id``, ``text``, ``score`` (1 - cosine distance, in [0, 1])
+        and ``created_at``. ``source_types`` limits the search to some of
+        ``feedback``, ``call_error``, ``event`` and ``llm``; ``min_score``
+        drops weaker matches. Returns [] on any failure, logged.
+        """
+        sink = self.embedding_sink
+        if sink is None:
+            logger.warning("similar() needs an EmbeddingSink")
+            return []
+        try:
+            return await sink.similar(
+                str(text),
+                source_types=None if source_types is None else list(source_types),
+                k=k,
+                min_score=min_score,
+            )
+        except Exception:
+            logger.warning("Similarity search failed", exc_info=True)
+            return []
+
+    async def similar_feedback(
+        self,
+        feedback_ref: Any,
+        *,
+        k: int = 10,
+        source_types: Iterable[str] | None = None,
+        min_score: float | None = None,
+    ) -> list[dict[str, Any]]:
+        """Texts most like an embedded feedback item, excluding the item itself.
+
+        Uses the feedback's stored vector, so nothing is embedded. Returns []
+        when the feedback is not embedded yet, and on any failure. Never raises.
+        """
+        sink = self.embedding_sink
+        if sink is None:
+            logger.warning("similar_feedback() needs an EmbeddingSink")
+            return []
+        ref = str(feedback_ref)
+        try:
+            vector = await sink.vector_for("feedback", ref)
+            if vector is None:
+                return []
+            return await sink.nearest(
+                vector,
+                source_types=None if source_types is None else list(source_types),
+                k=k,
+                min_score=min_score,
+                exclude=("feedback", ref),
+            )
+        except Exception:
+            logger.warning("similar_feedback(%r) failed", ref, exc_info=True)
+            return []
+
     @property
     def database_sink(self) -> DatabaseSink | None:
         """The first DatabaseSink among the sinks, if any."""

src/fastmcp_feedback/instrumentation/storage.py

@@ -1,4 +1,5 @@
-"""Async SQLAlchemy storage for tool-call records, events and feedback links.
+"""Async SQLAlchemy storage for tool-call records, events, feedback links and
+embeddings.
 
 Hosts that manage their own schema include ``build_metadata(prefix)`` in their
 Alembic ``env.py`` (``target_metadata``) and pass their existing AsyncEngine to
@@ -19,21 +20,25 @@ from sqlalchemy import (
     Column,
     DateTime,
     Float,
+    Index,
     Integer,
     MetaData,
     String,
     Table,
     Text,
+    UniqueConstraint,
     and_,
     delete,
     exists,
     inspect,
     or_,
     select,
+    text,
 )
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.exc import DBAPIError
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+from sqlalchemy.types import TypeDecorator
 
 from .events import MAX_KEY_CHARS, MAX_KIND_CHARS, EventRecord
 from .record import ToolCallRecord
@@ -46,19 +51,62 @@ DEFAULT_TABLE_PREFIX = "ffb_"
 # none_as_null: a missing payload is SQL NULL, so "WHERE args IS NULL" works.
 _JSON = JSON(none_as_null=True).with_variant(JSONB(none_as_null=True), "postgresql")
 
+DEFAULT_EMBEDDING_DIM = 1024
+# source_type values in {prefix}embeddings. Feedback rows survive pruning.
+EMBEDDING_SOURCES = ("feedback", "call_error", "event", "llm")
+FEEDBACK_SOURCE = "feedback"
+
+
+class EmbeddingVector(TypeDecorator):
+    """pgvector ``VECTOR(dim)`` on PostgreSQL, a JSON list of floats elsewhere.
+
+    pgvector is imported only when a PostgreSQL dialect needs the type, so
+    SQLite hosts never need it installed.
+    """
+
+    impl = JSON
+    cache_ok = True
+
+    def __init__(self, dim: int = DEFAULT_EMBEDDING_DIM):
+        super().__init__()
+        self.dim = dim
+
+    def load_dialect_impl(self, dialect: Any) -> Any:
+        if dialect.name == "postgresql":
+            try:
+                from pgvector.sqlalchemy import VECTOR
+            except ImportError as exc:
+                raise ImportError(
+                    "Embeddings on PostgreSQL need pgvector: install "
+                    '"fastmcp-feedback[embeddings]"'
+                ) from exc
+            return dialect.type_descriptor(VECTOR(self.dim))
+        return dialect.type_descriptor(JSON())
+
+    def process_bind_param(self, value: Any, dialect: Any) -> Any:
+        return None if value is None else [float(v) for v in value]
+
+    def process_result_value(self, value: Any, dialect: Any) -> Any:
+        return None if value is None else [float(v) for v in value]
+
 
 def build_metadata(
-    prefix: str = DEFAULT_TABLE_PREFIX, metadata: MetaData | None = None
+    prefix: str = DEFAULT_TABLE_PREFIX,
+    metadata: MetaData | None = None,
+    embedding_dim: int | None = DEFAULT_EMBEDDING_DIM,
 ) -> MetaData:
-    """Return MetaData holding ``{prefix}tool_calls``, ``{prefix}events`` and
-    ``{prefix}feedback_call_links``.
+    """Return MetaData holding ``{prefix}tool_calls``, ``{prefix}events``,
+    ``{prefix}feedback_call_links`` and ``{prefix}embeddings``.
 
     Pass an existing MetaData to add the tables to it instead.
+    ``embedding_dim`` sizes the vector column; None leaves the embeddings
+    table out, for hosts that do not embed and have no pgvector.
     """
     metadata = metadata if metadata is not None else MetaData()
     calls = f"{prefix}tool_calls"
     events = f"{prefix}events"
     links = f"{prefix}feedback_call_links"
+    embeddings = f"{prefix}embeddings"
     if calls not in metadata.tables:
         Table(
             calls,
@@ -119,6 +167,34 @@ def build_metadata(
             Column("rule", String(32), nullable=False),
             Column("linked_at", DateTime(timezone=True), nullable=False),
         )
+    if embedding_dim is not None and embeddings not in metadata.tables:
+        table = Table(
+            embeddings,
+            metadata,
+            Column("id", String(36), primary_key=True),
+            # feedback, call_error, event or llm
+            Column("source_type", String(16), nullable=False, index=True),
+            # The feedback ref, call id or event id the text came from.
+            Column("source_id", String(255), nullable=False, index=True),
+            Column("model", String(128), nullable=False),
+            Column("dim", Integer, nullable=False),
+            # sha256 hex of ``text``; an unchanged text is not embedded again.
+            Column("text_hash", String(64), nullable=False),
+            # What was embedded: redacted and truncated.
+            Column("text", Text, nullable=False),
+            Column("created_at", DateTime(timezone=True), nullable=False, index=True),
+            Column("embedding", EmbeddingVector(embedding_dim), nullable=False),
+            UniqueConstraint(
+                "source_type", "source_id", "model", name=f"{embeddings}_source_uq"
+            ),
+        )
+        # Cosine HNSW index; other databases search by brute force.
+        Index(
+            f"{embeddings}_hnsw",
+            table.c.embedding,
+            postgresql_using="hnsw",
+            postgresql_ops={"embedding": "vector_cosine_ops"},
+        ).ddl_if(dialect="postgresql")
     return metadata
 
 
@@ -140,6 +216,8 @@ class DatabaseSink:
         prune_interval: minimum time between automatic prunes.
         prune_batch: rows deleted per statement, so no single delete holds
             locks for long.
+        embedding_dim: vector size of ``{prefix}embeddings``; must match the
+            embedder of any EmbeddingSink writing through this sink.
     """
 
     def __init__(
@@ -151,6 +229,7 @@ class DatabaseSink:
         retention: timedelta | None = None,
         prune_interval: timedelta = timedelta(hours=1),
         prune_batch: int = 5000,
+        embedding_dim: int = DEFAULT_EMBEDDING_DIM,
     ):
         if retention is not None and retention <= timedelta(0):
             raise ValueError(f"retention must be positive, got {retention!r}")
@@ -158,12 +237,19 @@ class DatabaseSink:
             raise ValueError(f"prune_batch must be at least 1, got {prune_batch!r}")
         self._owns_engine = isinstance(engine, str)
         self.engine = create_async_engine(engine) if isinstance(engine, str) else engine
-        self.metadata = build_metadata(prefix)
+        self.metadata = build_metadata(prefix, embedding_dim=embedding_dim)
         self.table = self.metadata.tables[f"{prefix}tool_calls"]
         self.events = self.metadata.tables[f"{prefix}events"]
         self.links = self.metadata.tables[f"{prefix}feedback_call_links"]
+        self.embeddings = self.metadata.tables[f"{prefix}embeddings"]
+        self.embedding_dim = embedding_dim
         self.create_tables = create_tables
         self._tables_ready = False
+        # The embeddings table (and on Postgres the vector extension) is only
+        # created once an EmbeddingSink uses this sink; see ensure_embeddings.
+        self._embeddings_ready = False
+        self._embeddings_failed_at: float | None = None
+        self._embeddings_lock = asyncio.Lock()
         # The background writer and link_feedback can both reach _ready() at
         # once; without the lock both run create_all and one fails.
         self._tables_lock = asyncio.Lock()
@@ -196,8 +282,73 @@ class DatabaseSink:
         self._tables_ready = True
 
     async def _create_all(self) -> None:
+        # The embeddings table is left to ensure_embeddings, so hosts that do
+        # not embed never need pgvector or the vector extension.
+        tables = [t for t in self.metadata.sorted_tables if t is not self.embeddings]
         async with self.engine.begin() as conn:
-            await conn.run_sync(self.metadata.create_all)
+            await conn.run_sync(self.metadata.create_all, tables=tables)
+
+    async def ensure_embeddings(self) -> bool:
+        """Create ``{prefix}embeddings`` if missing. Only when create_tables=True.
+
+        On PostgreSQL this first runs ``CREATE EXTENSION IF NOT EXISTS vector``.
+        Never raises: a missing privilege, a missing pgvector package or any
+        other failure is logged, and False is returned. A failed attempt is
+        retried at most once a minute. Called by EmbeddingSink before it writes.
+        """
+        if not self.create_tables or self._embeddings_ready:
+            return True
+        async with self._embeddings_lock:
+            if self._embeddings_ready:
+                return True
+            now = time.monotonic()
+            if (
+                self._embeddings_failed_at is not None
+                and now - self._embeddings_failed_at < 60
+            ):
+                return False
+            try:
+                await self._ready()
+                if self.engine.dialect.name == "postgresql":
+                    await self._create_vector_extension()
+                try:
+                    await self._create_embeddings_table()
+                except DBAPIError:
+                    # Another process creating it at the same moment.
+                    await asyncio.sleep(_random.uniform(0.05, 0.25))
+                    await self._create_embeddings_table()
+            except Exception as exc:
+                self._embeddings_failed_at = now
+                logger.warning(
+                    "Could not create %s (%s: %s); embeddings are not stored until "
+                    "it exists. Create it with a migration (see the README) or "
+                    "fix the cause; retrying in a minute.",
+                    self.embeddings.name,
+                    type(exc).__name__,
+                    str(exc).splitlines()[0] if str(exc) else "",
+                )
+                return False
+            self._embeddings_ready = True
+            return True
+
+    async def _create_vector_extension(self) -> None:
+        try:
+            async with self.engine.begin() as conn:
+                await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+        except DBAPIError as exc:
+            # Usually a missing privilege. The table step then says whether
+            # the extension exists anyway (a DBA may have created it).
+            logger.warning(
+                "CREATE EXTENSION vector failed (%s); a superuser or database "
+                "owner must run it once in this database",
+                str(exc.orig).splitlines()[0] if exc.orig else type(exc).__name__,
+            )
+
+    async def _create_embeddings_table(self) -> None:
+        async with self.engine.begin() as conn:
+            await conn.run_sync(
+                lambda sync: self.embeddings.create(sync, checkfirst=True)
+            )
 
     async def _ready(self) -> None:
         if self.create_tables and not self._tables_ready:
@@ -238,15 +389,16 @@ class DatabaseSink:
             logger.warning("Pruning %s failed", self.table.name, exc_info=True)
 
     async def prune(self, older_than: datetime | None = None) -> int:
-        """Delete tool-call and event rows from before ``older_than``.
+        """Delete tool-call, event and embedding rows from before ``older_than``.
 
         Defaults to now minus ``retention``; with neither, deletes nothing.
         Calls linked to feedback in ``{prefix}feedback_call_links`` are kept,
         and links are never deleted. Events go by ``occurred_at``, whatever
-        call they belong to. Deletes ``prune_batch`` rows per transaction until
-        none are left. Returns the total rows deleted from both tables and
-        logs each table's count. A database without the events table (not yet
-        migrated) has its calls pruned and the events step skipped.
+        call they belong to. Embeddings go by ``created_at``, except those of
+        feedback (``source_type="feedback"``), which are kept. Deletes
+        ``prune_batch`` rows per transaction until none are left. Returns the
+        total rows deleted from all three tables and logs each table's count.
+        A table that does not exist yet (not migrated) is skipped.
         """
         if older_than is None:
             if self.retention is None:
@@ -267,15 +419,26 @@ class DatabaseSink:
             events = await self._delete_batches(
                 e, select(e.c.id).where(e.c.occurred_at < older_than)
             )
-        if calls or events:
+        vectors = 0
+        if await self._has_table(self.embeddings.name):
+            v = self.embeddings
+            vectors = await self._delete_batches(
+                v,
+                select(v.c.id).where(
+                    v.c.created_at < older_than, v.c.source_type != FEEDBACK_SOURCE
+                ),
+            )
+        if calls or events or vectors:
             logger.info(
-                "Pruned %d rows from %s and %d from %s",
+                "Pruned %d rows from %s, %d from %s and %d from %s",
                 calls,
                 t.name,
                 events,
                 self.events.name,
+                vectors,
+                self.embeddings.name,
             )
-        return calls + events
+        return calls + events + vectors
 
     async def _delete_batches(self, table: Any, ids_query: Any) -> int:
         batch = ids_query.limit(self.prune_batch)

src/fastmcp_feedback/mixins/submission.py

@@ -21,7 +21,7 @@ class SubmissionMixin(BaseFeedbackMixin):
     """Mixin for feedback submission functionality.
 
     Pass an InstrumentationMiddleware to link each submission to the tool
-    calls that led up to it.
+    calls that led up to it and record a ``feedback.submitted`` event for it.
     """
 
     def __init__(
@@ -86,6 +86,17 @@ class SubmissionMixin(BaseFeedbackMixin):
 
             linked_calls = 0
             if self.instrumentation is not None:
+                # Feeds an EmbeddingSink, so similar reports can be found.
+                self.instrumentation.record_event(
+                    "feedback.submitted",
+                    key=feedback_id,
+                    attrs={
+                        "type": request.type,
+                        "title": request.title,
+                        "description": request.description,
+                        "submitter": request.submitter,
+                    },
+                )
                 links = await self.instrumentation.capture_recent_calls()
                 if await self.instrumentation.link_feedback(feedback_id, links):
                     linked_calls = len(links)

tests/unit/test_embeddings.py

@@ -0,0 +1,848 @@
+"""Tests for EmbeddingSink, OpenAIEmbedder and similarity search.
+
+DB tests take the ``db_url`` fixture, so they run on SQLite and, with
+FFB_TEST_DATABASE_URL set, on Postgres with pgvector.
+"""
+
+import asyncio
+import contextlib
+import hashlib
+import json
+import logging
+import math
+import os
+import re
+import time
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from fastmcp import Client, FastMCP
+from fastmcp.exceptions import ToolError
+from sqlalchemy import func, select, text, update
+
+from fastmcp_feedback import add_feedback_tools
+from fastmcp_feedback.instrumentation import (
+    REDACTED,
+    DatabaseSink,
+    EmbeddingSink,
+    EventRecord,
+    OpenAIEmbedder,
+    Redactor,
+    ToolCallRecord,
+    build_metadata,
+    extract_texts,
+    instrument,
+)
+from fastmcp_feedback.instrumentation.embeddings import cosine_score
+
+DIM = 256
+SECRET = "sk-abcdefghijklmnopqrstuvwx"
+
+
+def token_vector(text_: str, dim: int = DIM) -> list[float]:
+    """Bag of hashed words, normalized: texts sharing words are close."""
+    v = [0.0] * dim
+    for tok in re.findall(r"[a-z0-9]+", text_.lower()):
+        h = int.from_bytes(hashlib.sha256(tok.encode()).digest()[:4], "big")
+        v[h % dim] += 1.0
+    n = math.sqrt(sum(x * x for x in v)) or 1.0
+    return [x / n for x in v]
+
+
+class FakeEmbedder:
+    def __init__(
+        self,
+        dim=DIM,
+        model="fake-embed",
+        *,
+        fail=False,
+        delay=0.0,
+        gate=None,
+        out_dim=None,
+    ):
+        self.dim = dim
+        self.model = model
+        self.fail = fail
+        self.delay = delay
+        self.gate = gate
+        self.out_dim = out_dim or dim
+        self.calls: list[list[str]] = []
+
+    @property
+    def seen(self) -> list[str]:
+        return [t for batch in self.calls for t in batch]
+
+    async def embed(self, texts):
+        self.calls.append(list(texts))
+        if self.gate is not None:
+            await self.gate.wait()
+        if self.delay:
+            await asyncio.sleep(self.delay)
+        if self.fail:
+            raise RuntimeError("embedding backend down")
+        return [token_vector(t, self.out_dim) for t in texts]
+
+
+@pytest.fixture
+async def db(db_url):
+    made = []
+
+    def factory(name="calls", **kwargs):
+        kwargs.setdefault("create_tables", True)
+        kwargs.setdefault("embedding_dim", DIM)
+        sink = DatabaseSink(db_url(name), **kwargs)
+        made.append(sink)
+        return sink
+
+    yield factory
+    for sink in made:
+        await sink.engine.dispose()
+
+
+def call_record(tool="render", outcome="error", message="render timed out", **kw):
+    return ToolCallRecord(
+        tool=tool,
+        started_at=datetime.now(UTC),
+        duration_ms=1.0,
+        ok=outcome == "ok",
+        outcome=outcome,
+        mode="meta",
+        error_type=None if outcome == "ok" else kw.pop("error_type", "TimeoutError"),
+        error_message=None if outcome == "ok" else message,
+        **kw,
+    )
+
+
+def event(kind, key=None, **attrs):
+    return EventRecord(kind=kind, occurred_at=datetime.now(UTC), key=key, attrs=attrs)
+
+
+async def rows(sink, *where):
+    t = sink.embeddings
+    if not await sink._has_table(t.name):  # created on the first embedding
+        return []
+    async with sink.engine.connect() as conn:
+        result = await conn.execute(select(t).where(*where).order_by(t.c.source_id))
+        return [dict(r) for r in result.mappings()]
+
+
+async def settle(mw, emb):
+    await mw.flush()
+    await emb.flush()
+
+
+class TestExtractTexts:
+    def test_call_errors(self):
+        ok = call_record(outcome="ok")
+        err = call_record(message="render timed out", error_type="TimeoutError")
+        soft = call_record(
+            tool="find",
+            outcome="soft_error",
+            message="status=unknown_target",
+            error_type="SoftError",
+        )
+        items = extract_texts([ok, err, soft])
+        assert [(i.source_type, i.source_id, i.text) for i in items] == [
+            ("call_error", err.id, "render: TimeoutError: render timed out"),
+            ("call_error", soft.id, "find: SoftError: status=unknown_target"),
+        ]
+
+    def test_feedback(self):
+        ev = event(
+            "feedback.submitted",
+            key="bug-7Q2X",
+            title="Render hangs",
+            description="the render tool hung on scene a",
+            type="bug",
+        )
+        [item] = extract_texts([ev])
+        assert (item.source_type, item.source_id, item.text) == (
+            "feedback",
+            "bug-7Q2X",
+            "Render hangs\nthe render tool hung on scene a",
+        )
+
+    def test_event_text_keys(self):
+        ev = event(
+            "job.failed",
+            key="job-1",
+            error_excerpt="solver diverged",
+            reason="residual NaN",
+            nproc=8,
+        )
+        quiet = event("job.finished", key="job-2", duration_s=3)
+        [item] = extract_texts([ev, quiet])
+        assert (item.source_type, item.source_id, item.text) == (
+            "event",
+            ev.id,
+            "solver diverged\nresidual NaN",
+        )
+        [custom] = extract_texts([quiet], event_text_keys=("duration_s",))
+        assert custom.text == "3"
+
+    def test_llm(self):
+        ev = event("llm.call", model="m", prompt="why red?", completion="the light")
+        [item] = extract_texts([ev])
+        assert (item.source_type, item.source_id, item.text) == (
+            "llm",
+            ev.id,
+            "prompt: why red?\ncompletion: the light",
+        )
+
+    def test_llm_without_text_is_a_plain_event(self):
+        ev = event("llm.call", model="m", ok=False, error="upstream 529 overloaded")
+        [item] = extract_texts([ev])
+        assert (item.source_type, item.text) == ("event", "upstream 529 overloaded")
+
+    def test_sources_filter(self):
+        records = [
+            call_record(),
+            event("feedback.submitted", key="1", title="t", description="d"),
+            event("job.failed", error="boom"),
+            event("llm.call", prompt="p"),
+        ]
+        assert {i.source_type for i in extract_texts(records)} == {
+            "call_error",
+            "feedback",
+            "event",
+            "llm",
+        }
+        assert [
+            i.source_type for i in extract_texts(records, sources=["feedback"])
+        ] == ["feedback"]
+
+
+class TestEmbeddingSink:
+    async def test_every_source_lands(self, db):
+        sink = db()
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(fake, sink)
+        err = call_record()
+        records = [
+            err,
+            event(
+                "feedback.submitted",
+                key="42",
+                title="Render hangs",
+                description="scene a",
+            ),
+            event("job.failed", key="job-1", error_excerpt="solver diverged"),
+            event("llm.call", prompt="hello", completion="hi"),
+        ]
+        await emb.write(records)
+        await emb.flush()
+        stored = await rows(sink)
+        assert {r["source_type"] for r in stored} == {
+            "call_error",
+            "feedback",
+            "event",
+            "llm",
+        }
+        for r in stored:
+            assert r["dim"] == DIM and r["model"] == "fake-embed"
+            assert len(r["embedding"]) == DIM
+            assert r["text_hash"] == hashlib.sha256(r["text"].encode()).hexdigest()
+            assert r["created_at"] is not None
+        assert emb.embedded == 4 and emb.errors == 0
+        await emb.aclose()
+
+    async def test_submit_feedback_is_embedded(self, db, tmp_path):
+        app = FastMCP("app")
+        sink = db("fb")
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(fake, sink)
+        mw = instrument(app, [sink, emb])
+        add_feedback_tools(
+            app, database_url=f"sqlite:///{tmp_path}/fb.db", instrumentation=mw
+        )
+        async with Client(app) as client:
+            result = await client.call_tool(
+                "submit_feedback",
+                {
+                    "request": {
+                        "type": "bug",
+                        "title": "Render hangs",
+                        "description": "the render tool hung on scene a",
+                        "submitter": "model",
+                    }
+                },
+            )
+        ref = result.data["feedback_id"]
+        await settle(mw, emb)
+        [row] = await rows(sink, sink.embeddings.c.source_type == "feedback")
+        assert row["source_id"] == ref
+        assert row["text"] == "Render hangs\nthe render tool hung on scene a"
+        [ev] = await mw.events_for(ref, kinds=["feedback.submitted"])
+        assert ev["attrs"]["type"] == "bug" and ev["attrs"]["submitter"] == "model"
+        await mw.aclose()
+
+    async def test_redaction(self, db):
+        sink = db()
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(
+            fake, sink, redactor=Redactor(extra_patterns=[r"acme_\w{8}"])
+        )
+        # Straight to the sink, so upstream redaction cannot hide a gap here.
+        await emb.write(
+            [
+                call_record(message=f"401 with api_key={SECRET} and acme_12345678"),
+                event("job.failed", error=f"Bearer {SECRET}"),
+            ]
+        )
+        await emb.flush()
+        assert fake.seen and all(
+            SECRET not in t and "acme_12345678" not in t for t in fake.seen
+        )
+        stored = await rows(sink)
+        assert len(stored) == 2
+        for r in stored:
+            assert SECRET not in r["text"] and REDACTED in r["text"]
+        await emb.aclose()
+
+    async def test_instrument_hands_over_its_redactor(self, db):
+        app = FastMCP("app")
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        redactor = Redactor(extra_patterns=[r"acme_\w{8}"])
+        mw = instrument(app, [sink, emb], redactor=redactor)
+        assert emb.redactor is redactor
+        await mw.aclose()
+
+    async def test_max_chars(self, db):
+        sink = db()
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(fake, sink, max_chars=50)
+        await emb.write([event("job.failed", error="word " * 100)])
+        await emb.flush()
+        [row] = await rows(sink)
+        assert len(row["text"]) == 50 and fake.seen == [row["text"]]
+        await emb.aclose()
+
+    async def test_dedupe_and_replace(self, db):
+        sink = db()
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(fake, sink)
+        ev = event("feedback.submitted", key="7", title="Render hangs", description="a")
+        await emb.write([ev])
+        await emb.flush()
+        await emb.write([ev, ev])
+        await emb.flush()
+        assert fake.seen == ["Render hangs\na"]
+        assert emb.skipped == 2
+        changed = event(
+            "feedback.submitted", key="7", title="Render hangs", description="b"
+        )
+        await emb.write([changed])
+        await emb.flush()
+        [row] = await rows(sink)
+        assert row["text"] == "Render hangs\nb"
+        assert fake.seen == ["Render hangs\na", "Render hangs\nb"]
+        # Another model's vectors are kept apart.
+        other = EmbeddingSink(FakeEmbedder(model="other"), sink)
+        await other.write([changed])
+        await other.flush()
+        assert len(await rows(sink)) == 2
+        await emb.aclose()
+        await other.aclose()
+
+    async def test_queue_full_drops(self, db):
+        sink = db()
+        gate = asyncio.Event()
+        fake = FakeEmbedder(gate=gate)
+        emb = EmbeddingSink(fake, sink, max_queue=2, batch_size=1)
+        start = time.perf_counter()
+        await emb.write([event("job.failed", error=f"boom {i}") for i in range(6)])
+        assert time.perf_counter() - start < 0.5
+        await asyncio.sleep(0.05)  # the worker takes one and blocks on the gate
+        await emb.write([event("job.failed", error=f"late {i}") for i in range(3)])
+        assert emb.dropped >= 4
+        gate.set()
+        await emb.flush()
+        assert emb.errors == 0
+        assert len(await rows(sink)) == 9 - emb.dropped
+        await emb.aclose()
+
+    async def test_embedder_errors_never_reach_tool_calls(self, db):
+        app = FastMCP("app")
+
+        @app.tool
+        def render(scene: str) -> str:
+            raise RuntimeError(f"render of {scene} timed out")
+
+        @app.tool
+        def ping() -> str:
+            return "pong"
+
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(fail=True), sink)
+        mw = instrument(app, [sink, emb])
+        async with Client(app) as client:
+            with contextlib.suppress(ToolError):
+                await client.call_tool("render", {"scene": "a"})
+            assert (await client.call_tool("ping", {})).data == "pong"
+        await settle(mw, emb)
+        assert emb.errors == 1
+        assert mw.dispatcher.sink_errors == 0
+        assert await rows(sink) == []
+        async with sink.engine.connect() as conn:
+            calls = (
+                await conn.execute(select(func.count()).select_from(sink.table))
+            ).scalar_one()
+        assert calls == 2
+        assert await mw.similar("render timed out") == []
+        await mw.aclose()
+
+    async def test_slow_embedder_stays_off_the_call_path(self, db):
+        app = FastMCP("app")
+
+        @app.tool
+        def render() -> str:
+            raise RuntimeError("render timed out")
+
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(delay=2.0), sink, timeout=5.0)
+        mw = instrument(app, [sink, emb], sink_timeout=0.5)
+        async with Client(app) as client:
+            start = time.perf_counter()
+            for _ in range(3):
+                with contextlib.suppress(ToolError):
+                    await client.call_tool("render", {})
+            elapsed = time.perf_counter() - start
+        await mw.flush()
+        assert elapsed < 1.5
+        assert mw.dispatcher.sink_errors == 0
+        await emb.flush()
+        assert emb.embedded == 3
+        await mw.aclose()
+
+    async def test_wrong_vector_length_is_an_error(self, db):
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(out_dim=DIM + 1), sink)
+        await emb.write([event("job.failed", error="boom")])
+        await emb.flush()
+        assert emb.errors == 1 and await rows(sink) == []
+        await emb.aclose()
+
+    async def test_dim_mismatch_refused_at_construction(self, db):
+        with pytest.raises(ValueError, match="embedding_dim=512"):
+            EmbeddingSink(FakeEmbedder(dim=512), db())
+
+    async def test_unknown_source_refused(self, db):
+        with pytest.raises(ValueError, match="unknown sources"):
+            EmbeddingSink(FakeEmbedder(), db(), sources=["feedback", "chat"])
+
+    async def test_aclose_drains(self, db):
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(delay=0.1), sink, batch_size=1)
+        await emb.write([event("job.failed", error=f"e{i}") for i in range(3)])
+        await emb.aclose()
+        assert emb.embedded == 3 and emb.dropped == 0
+
+    async def test_aclose_timeout_drops(self, db):
+        sink = db()
+        emb = EmbeddingSink(
+            FakeEmbedder(gate=asyncio.Event()), sink, batch_size=1, timeout=0.2
+        )
+        await emb.write([event("job.failed", error=f"e{i}") for i in range(3)])
+        await emb.aclose()
+        assert emb.dropped == 2 and emb.embedded == 0
+
+
+class TestLlmText:
+    def app(self, sinks, **kwargs):
+        app = FastMCP("llm")
+        holder = {}
+
+        @app.tool
+        async def ask(q: str) -> str:
+            holder["mw"].record_llm_call(
+                model="m",
+                prompt=f"{q} token={SECRET}",
+                completion="because",
+                attrs={"prompt": "smuggled", "temperature": 0.2},
+            )
+            return "because"
+
+        holder["mw"] = mw = instrument(app, sinks, **kwargs)
+        return app, mw
+
+    async def test_dropped_by_default(self, db):
+        sink = db()
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(fake, sink)
+        app, mw = self.app([sink, emb])
+        async with Client(app) as client:
+            await client.call_tool("ask", {"q": "why is the sky blue"})
+        await settle(mw, emb)
+        [ev] = [e for e in mw.recent_events if e.kind == "llm.call"]
+        assert "prompt" not in ev.attrs and "completion" not in ev.attrs
+        assert ev.attrs["temperature"] == 0.2
+        assert fake.seen == []
+        assert await rows(sink) == []
+        await mw.aclose()
+
+    async def test_embedded_when_captured(self, db):
+        sink = db()
+        fake = FakeEmbedder()
+        emb = EmbeddingSink(fake, sink)
+        app, mw = self.app([sink, emb], capture_llm_text=True, max_error_chars=10)
+        async with Client(app) as client:
+            await client.call_tool("ask", {"q": "why is the sky blue " * 10})
+        await settle(mw, emb)
+        [ev] = [e for e in mw.recent_events if e.kind == "llm.call"]
+        assert len(ev.attrs["prompt"]) == 80  # 8 * max_error_chars
+        assert ev.attrs["completion"] == "because"
+        [row] = await rows(sink)
+        assert row["source_type"] == "llm" and row["source_id"] == ev.id
+        assert row["text"].startswith("prompt: why is the sky blue")
+        assert row["text"].endswith("completion: because")
+        await mw.aclose()
+
+    async def test_captured_text_is_redacted(self, db):
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        app, mw = self.app([sink, emb], capture_llm_text=True)
+        async with Client(app) as client:
+            await client.call_tool("ask", {"q": "q"})
+        await settle(mw, emb)
+        [ev] = [e for e in mw.recent_events if e.kind == "llm.call"]
+        assert SECRET not in ev.attrs["prompt"] and REDACTED in ev.attrs["prompt"]
+        [row] = await rows(sink)
+        assert SECRET not in row["text"]
+        await mw.aclose()
+
+
+FIXTURE = [
+    ("call_error", "c1", "render: TimeoutError: render timed out after 30 s"),
+    ("call_error", "c2", "upload: OSError: disk full on volume data"),
+    ("event", "e1", "render job timed out waiting for the gpu"),
+    ("feedback", "bug-1", "Render hangs\nthe render tool timed out on scene a"),
+    ("feedback", "bug-2", "Dark mode please\nthe ui is too bright at night"),
+    ("llm", "l1", "prompt: summarise the log\ncompletion: nothing notable"),
+]
+
+
+async def seed(sink, emb):
+    from fastmcp_feedback.instrumentation import EmbeddingText
+
+    for source_type, source_id, text_ in FIXTURE:
+        emb._enqueue(EmbeddingText(source_type, source_id, text_))
+    await emb.flush()
+    assert len(await rows(sink)) == len(FIXTURE)
+
+
+class TestSimilar:
+    async def test_ranking_matches_brute_force(self, db, db_backend):
+        sink = db()
+        app = FastMCP("s")
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        mw = instrument(app, [sink, emb])
+        await seed(sink, emb)
+
+        query = "render timed out"
+        q = token_vector(query)
+        expected = sorted(
+            ((cosine_score(q, token_vector(t)), sid) for _, sid, t in FIXTURE),
+            reverse=True,
+        )
+        found = await mw.similar(query, k=len(FIXTURE))
+        # Rows sharing no word with the query all score 0; their order is free.
+        positive = [sid for score, sid in expected if score > 1e-6]
+        assert len(positive) >= 3
+        assert [r["source_id"] for r in found][: len(positive)] == positive
+        assert {r["source_id"] for r in found} == {sid for _, sid in expected}
+        want = {sid: score for score, sid in expected}
+        for r in found:
+            assert r["score"] == pytest.approx(want[r["source_id"]], abs=1e-4)
+            assert 0.0 <= r["score"] <= 1.0
+            assert r["created_at"].tzinfo is not None
+            assert set(r) == {"source_type", "source_id", "text", "score", "created_at"}
+        assert found[0]["source_id"] == "c1"
+
+        top = await mw.similar(query, k=2)
+        assert [r["source_id"] for r in top] == positive[:2]
+        only = await mw.similar(query, source_types=["feedback"])
+        assert [r["source_id"] for r in only] == ["bug-1", "bug-2"]
+        strong = await mw.similar(query, min_score=0.4)
+        assert strong and all(r["score"] >= 0.4 for r in strong)
+        assert len(strong) < len(FIXTURE)
+        await mw.aclose()
+
+    async def test_similar_feedback_excludes_itself(self, db):
+        sink = db()
+        app = FastMCP("s")
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        mw = instrument(app, [sink, emb])
+        await seed(sink, emb)
+        found = await mw.similar_feedback("bug-1", k=3)
+        assert "bug-1" not in [r["source_id"] for r in found]
+        assert found[0]["source_id"] in ("c1", "e1")
+        feedback_only = await mw.similar_feedback("bug-1", source_types=["feedback"])
+        assert [r["source_id"] for r in feedback_only] == ["bug-2"]
+        assert await mw.similar_feedback("bug-404") == []
+        await mw.aclose()
+
+    async def test_scan_limit_without_pgvector(self, db, db_backend, caplog):
+        if db_backend == "postgres":
+            pytest.skip("brute force only runs without pgvector")
+        sink = db()
+        app = FastMCP("s")
+        emb = EmbeddingSink(FakeEmbedder(), sink, scan_limit=2)
+        mw = instrument(app, [sink, emb])
+        await seed(sink, emb)
+        caplog.set_level(logging.INFO, "fastmcp_feedback.instrumentation.embeddings")
+        assert len(await mw.similar("render", k=10)) <= 2
+        await mw.similar("render", k=10)
+        assert len([r for r in caplog.records if "scan_limit" in r.getMessage()]) == 1
+        await mw.aclose()
+
+    async def test_never_raises(self, db):
+        app = FastMCP("s")
+        mw = instrument(app, [])
+        assert await mw.similar("x") == []
+        assert await mw.similar_feedback("1") == []
+        await mw.aclose()
+        sink = db()
+        emb = EmbeddingSink(FakeEmbedder(fail=True), sink)
+        mw = instrument(FastMCP("t"), [sink, emb])
+        assert await mw.similar("x") == []
+        # No table yet: the lookup itself fails.
+        assert await mw.similar_feedback("1") == []
+        await mw.aclose()
+
+
+class TestPrune:
+    async def test_keeps_feedback_embeddings(self, db):
+        sink = db(prune_batch=2)
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        await seed(sink, emb)
+        t = sink.embeddings
+        old = datetime.now(UTC) - timedelta(days=40)
+        async with sink.engine.begin() as conn:
+            await conn.execute(
+                update(t).where(t.c.source_id != "c2").values(created_at=old)
+            )
+        deleted = await sink.prune(older_than=datetime.now(UTC) - timedelta(days=30))
+        assert deleted == 3
+        left = sorted(r["source_id"] for r in await rows(sink))
+        assert left == ["bug-1", "bug-2", "c2"]
+        await emb.aclose()
+
+    async def test_prune_without_embeddings_table(self, db):
+        sink = db()
+        await sink.write([call_record()])
+        assert not await sink._has_table(sink.embeddings.name)
+        assert await sink.prune(older_than=datetime.now(UTC) + timedelta(days=1)) == 1
+
+
+class TestCreateTables:
+    async def test_embeddings_table_only_with_an_embedding_sink(self, db):
+        sink = db()
+        await sink.write([call_record()])
+        assert not await sink._has_table(sink.embeddings.name)
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        await emb.write([call_record()])
+        await emb.flush()
+        assert await sink._has_table(sink.embeddings.name)
+        await emb.aclose()
+
+    async def test_create_tables_false_creates_nothing(self, db):
+        sink = db(create_tables=False)
+        assert await sink.ensure_embeddings() is True
+        assert not await sink._has_table(sink.embeddings.name)
+
+    async def test_failure_is_logged_not_raised(self, db, monkeypatch, caplog):
+        sink = db()
+
+        async def boom():
+            raise RuntimeError("no room")
+
+        monkeypatch.setattr(sink, "_create_embeddings_table", boom)
+        caplog.set_level(logging.WARNING)
+        emb = EmbeddingSink(FakeEmbedder(), sink)
+        assert await sink.ensure_embeddings() is False
+        assert any("Could not create" in r.getMessage() for r in caplog.records)
+        await emb.write([call_record()])
+        await emb.flush()
+        assert emb.errors == 1  # the insert fails; counted, not raised
+        await emb.aclose()
+
+    @pytest.mark.postgres
+    async def test_postgres_extension_table_and_index(self, db, db_backend):
+        if db_backend != "postgres":
+            pytest.skip("Postgres only")
+        sink = db()
+        await sink._ready()
+        async with sink.engine.begin() as conn:
+            await conn.execute(text("DROP EXTENSION IF EXISTS vector"))
+        assert await sink.ensure_embeddings() is True
+        async with sink.engine.connect() as conn:
+            ext = (
+                await conn.execute(
+                    text("SELECT count(*) FROM pg_extension WHERE extname = 'vector'")
+                )
+            ).scalar_one()
+            index = (
+                await conn.execute(
+                    text(
+                        "SELECT indexdef FROM pg_indexes "
+                        "WHERE indexname = 'ffb_embeddings_hnsw'"
+                    )
+                )
+            ).scalar_one()
+            coltype = (
+                await conn.execute(
+                    text(
+                        "SELECT format_type(atttypid, atttypmod) FROM pg_attribute "
+                        "WHERE attrelid = 'ffb_embeddings'::regclass "
+                        "AND attname = 'embedding'"
+                    )
+                )
+            ).scalar_one()
+        assert ext == 1
+        assert "hnsw" in index and "vector_cosine_ops" in index
+        assert coltype == f"vector({DIM})"
+
+    @pytest.mark.postgres
+    async def test_postgres_without_privilege(self, db, db_backend, db_url, caplog):
+        if db_backend != "postgres":
+            pytest.skip("Postgres only")
+        from sqlalchemy.engine import make_url
+
+        admin = db("priv", embedding_dim=DIM)
+        role = f"ffb_limited_{uuid.uuid4().hex[:8]}"
+        password = uuid.uuid4().hex
+        url = make_url(db_url("priv"))
+        async with admin.engine.begin() as conn:
+            await conn.execute(text("DROP EXTENSION IF EXISTS vector"))
+            await conn.execute(text(f"CREATE ROLE {role} LOGIN PASSWORD '{password}'"))
+            await conn.execute(text(f"GRANT ALL ON SCHEMA public TO {role}"))
+        limited = DatabaseSink(
+            url.set(username=role, password=password).render_as_string(
+                hide_password=False
+            ),
+            create_tables=True,
+            embedding_dim=DIM,
+        )
+        try:
+            caplog.set_level(logging.WARNING)
+            emb = EmbeddingSink(FakeEmbedder(), limited)
+            assert await limited.ensure_embeddings() is False
+            messages = [r.getMessage() for r in caplog.records]
+            assert any("CREATE EXTENSION vector failed" in m for m in messages)
+            assert any("Could not create" in m for m in messages)
+            await limited.write([call_record()])  # the rest still works
+            await emb.write([call_record()])
+            await emb.flush()
+            assert emb.errors == 1
+            await emb.aclose()
+        finally:
+            await limited.engine.dispose()
+            async with admin.engine.begin() as conn:
+                await conn.execute(text(f"DROP OWNED BY {role}"))
+                await conn.execute(text(f"DROP ROLE {role}"))
+
+
+class TestMetadata:
+    def test_build_metadata_tables(self):
+        md = build_metadata("x_", embedding_dim=768)
+        t = md.tables["x_embeddings"]
+        assert t.c.embedding.type.dim == 768
+        assert {c.name for c in t.c} == {
+            "id",
+            "source_type",
+            "source_id",
+            "model",
+            "dim",
+            "text_hash",
+            "text",
+            "created_at",
+            "embedding",
+        }
+        without = build_metadata("y_", embedding_dim=None)
+        assert "y_embeddings" not in without.tables
+        assert "y_tool_calls" in without.tables
+
+
+class TestOpenAIEmbedder:
+    def transport(self, handler):
+        from fastmcp_feedback.instrumentation.embeddings import _http_module
+
+        return _http_module().MockTransport(handler)
+
+    async def test_request_batches_and_order(self):
+        seen = []
+
+        def handler(request):
+            body = json.loads(request.content)
+            seen.append((request.url.path, request.headers.get("authorization"), body))
+            data = [
+                {"index": i, "embedding": [float(len(t)), 0.0, 1.0]}
+                for i, t in enumerate(body["input"])
+            ]
+            from fastmcp_feedback.instrumentation.embeddings import _http_module
+
+            return _http_module().Response(200, json={"data": list(reversed(data))})
+
+        e = OpenAIEmbedder(
+            "https://gw.example/v1/",
+            api_key="k",
+            model="m",
+            dim=3,
+            batch_size=2,
+            transport=self.transport(handler),
+        )
+        out = await e.embed(["a", "bb", "ccc"])
+        assert out == [[1.0, 0.0, 1.0], [2.0, 0.0, 1.0], [3.0, 0.0, 1.0]]
+        assert [s[0] for s in seen] == ["/v1/embeddings", "/v1/embeddings"]
+        assert seen[0][1] == "Bearer k"
+        assert [s[2]["input"] for s in seen] == [["a", "bb"], ["ccc"]]
+        assert all(s[2]["model"] == "m" for s in seen)
+        assert await e.embed([]) == []
+
+    async def test_dim_mismatch_raises(self):
+        from fastmcp_feedback.instrumentation.embeddings import _http_module
+
+        def handler(request):
+            return _http_module().Response(
+                200, json={"data": [{"index": 0, "embedding": [0.1, 0.2]}]}
+            )
+
+        e = OpenAIEmbedder(
+            "https://gw.example/v1", dim=3, transport=self.transport(handler)
+        )
+        with pytest.raises(ValueError, match="2-dim vector, expected 3"):
+            await e.embed(["a"])
+
+    async def test_http_error_raises(self):
+        from fastmcp_feedback.instrumentation.embeddings import _http_module
+
+        e = OpenAIEmbedder(
+            "https://gw.example/v1",
+            dim=3,
+            transport=self.transport(
+                lambda r: _http_module().Response(403, json={"error": "nope"})
+            ),
+        )
+        with pytest.raises(Exception, match="403"):
+            await e.embed(["a"])
+
+
+LIVE_URL = os.environ.get("FFB_TEST_EMBED_URL")
+
+
+@pytest.mark.skipif(not LIVE_URL, reason="FFB_TEST_EMBED_URL is not set")
+async def test_live_gateway():
+    """FFB_TEST_EMBED_URL=https://.../v1 FFB_TEST_EMBED_KEY=... [FFB_TEST_EMBED_MODEL]."""
+    e = OpenAIEmbedder(
+        LIVE_URL,
+        api_key=os.environ.get("FFB_TEST_EMBED_KEY"),
+        model=os.environ.get("FFB_TEST_EMBED_MODEL", "mxbai-embed-large"),
+        dim=int(os.environ.get("FFB_TEST_EMBED_DIM", "1024")),
+    )
+    a, b, c = await e.embed(
+        ["render timed out", "the render took too long and timed out", "dark mode"]
+    )
+    assert len(a) == e.dim
+    assert cosine_score(a, b) > cosine_score(a, c)

tests/unit/test_instrumentation.py

@@ -448,6 +448,7 @@ class TestDatabaseSink:
         assert build_metadata("ffb_", host) is host
         build_metadata("ffb_", host)
         assert sorted(host.tables) == [
+            "ffb_embeddings",
             "ffb_events",
             "ffb_feedback_call_links",
             "ffb_tool_calls",

uv.lock

@@ -601,7 +601,7 @@ wheels = [
 
 [[package]]
 name = "fastmcp-feedback"
-version = "2026.9.29.1"
+version = "2026.9.30"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },
@@ -638,6 +638,9 @@ dev = [
     { name = "ruff" },
     { name = "sqlalchemy", extra = ["asyncio"] },
 ]
+embeddings = [
+    { name = "pgvector" },
+]
 instrumentation = [
     { name = "aiosqlite" },
     { name = "sqlalchemy", extra = ["asyncio"] },
@@ -671,6 +674,7 @@ requires-dist = [
     { name = "fastmcp-feedback", extras = ["postgresql", "mysql", "dev"], marker = "extra == 'all'" },
     { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.7.0" },
     { name = "pgvector", marker = "extra == 'dev'", specifier = ">=0.5.0" },
+    { name = "pgvector", marker = "extra == 'embeddings'", specifier = ">=0.5.0" },
     { name = "pgvector", marker = "extra == 'test'", specifier = ">=0.5.0" },
     { name = "psycopg2-binary", marker = "extra == 'postgresql'", specifier = ">=2.9.0" },
     { name = "pydantic", specifier = ">=2.11.7" },
@@ -691,7 +695,7 @@ requires-dist = [
     { name = "sqlalchemy", extras = ["asyncio"], marker = "extra == 'instrumentation'", specifier = ">=2.0.43" },
     { name = "sqlalchemy", extras = ["asyncio"], marker = "extra == 'test'", specifier = ">=2.0.43" },
 ]
-provides-extras = ["postgresql", "instrumentation", "mysql", "dev", "test", "all"]
+provides-extras = ["postgresql", "instrumentation", "embeddings", "mysql", "dev", "test", "all"]
 
 [[package]]
 name = "fastmcp-slim"