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.

.env.example

@@ -0,0 +1,13 @@
+# Copy to .env (make does this for you on first run) and adjust as needed.
+COMPOSE_PROJECT_NAME=fastmcp-feedback-pkg
+
+FFB_PG_IMAGE=pgvector/pgvector:pg16
+FFB_PG_USER=ffb
+FFB_PG_PASSWORD=change-me
+FFB_PG_DB=ffb_test
+# Published on 127.0.0.1 only. 55432 is commonly taken by other local stacks.
+FFB_PG_PORT=55433
+
+# Read by the test suite. Unset, the Postgres test cases skip.
+# Each test gets its own database created on this server and dropped after.
+FFB_TEST_DATABASE_URL=postgresql+asyncpg://ffb:change-me@127.0.0.1:55433/ffb_test

.gitea/workflows/test.yml

@@ -36,11 +36,62 @@ jobs:
           uv pip show fastmcp | grep -i '^version'
 
       - name: Lint
-        run: uv run --no-sync ruff check src/ examples/ tests/_compat.py tests/unit/test_instrumentation.py
+        run: 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
 
       - name: Test
         run: uv run --no-sync pytest -q --no-cov
 
+  # The DatabaseSink tests again, on Postgres + pgvector instead of SQLite.
+  # Oldest and newest supported FastMCP; the SQLite legs above cover the rest.
+  # act_runner runs the job in a container, so the service is reached by name.
+  test-postgres:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+      matrix:
+        fastmcp: ['==2.12.2', '>=4,<5']
+
+    services:
+      ffb-postgres:
+        image: pgvector/pgvector:pg16
+        env:
+          POSTGRES_USER: ffb
+          POSTGRES_PASSWORD: ffb-ci
+          POSTGRES_DB: ffb_test
+        options: >-
+          --health-cmd "pg_isready -U ffb -d ffb_test"
+          --health-interval 2s
+          --health-timeout 3s
+          --health-retries 30
+
+    env:
+      FFB_TEST_DATABASE_URL: postgresql+asyncpg://ffb:ffb-ci@ffb-postgres:5432/ffb_test
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: astral-sh/setup-uv@v6
+        with:
+          python-version: '3.13'
+
+      - name: Install against fastmcp${{ matrix.fastmcp }}
+        run: |
+          uv venv
+          uv pip install -e ".[dev]" "fastmcp${{ matrix.fastmcp }}"
+          uv pip show fastmcp | grep -i '^version'
+
+      - name: Test with the Postgres cases enabled
+        run: make test-pg
+
+      - name: Postgres cases ran rather than skipped
+        run: |
+          out=$(uv run --no-sync pytest -q --no-cov -m postgres -rs)
+          echo "$out" | tail -3
+          if echo "$out" | grep -q 'SKIPPED'; then
+            echo "Postgres cases skipped; is the service reachable?"; exit 1
+          fi
+
   # Newest FastMCP with no ceiling. Allowed to fail: it exists to say
   # "a new major version is out and here is what breaks" before users do.
   canary:

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

@@ -97,6 +97,50 @@ uv run pytest -m unit
 uv run pytest -m integration
 ```
 
+### Running the tests against Postgres
+
+The `DatabaseSink` tests run on SQLite by default. Each one also has a
+`[postgres]` variant that runs against a real Postgres 16 with pgvector
+when `FFB_TEST_DATABASE_URL` is set, and skips otherwise. The repo ships a
+compose stack for it (Docker required):
+
+```bash
+make pg-up      # start ffb-postgres and wait until healthy (creates .env on first run)
+make test-pg    # whole suite, Postgres cases enabled
+make test       # whole suite, SQLite only
+make test-all   # both
+make pg-down    # stop it; the data volume is kept (pg-reset deletes it)
+```
+
+The server listens on `127.0.0.1` only, on `FFB_PG_PORT` from `.env`
+(55433 by default; change it if that port is taken). `make pg-psql` opens a
+shell on it. Every Postgres test gets its own freshly created database with
+the `vector` extension, dropped when the test ends, so tests never see each
+other's rows. To write a test that runs on both backends, take the `db_url`
+fixture from `tests/conftest.py`: `db_url("name")` returns an async URL for
+an empty database, the same one for the same name within a test.
+
+To point the suite at another server, set the variable yourself; the user
+needs permission to create and drop databases:
+
+```bash
+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

@@ -0,0 +1,63 @@
+# Dev/test helpers. The Postgres stack exists only so the test suite can run
+# against a real server; see CONTRIBUTING.md.
+
+COMPOSE ?= docker compose
+PYTEST ?= uv run --no-sync pytest
+PYTEST_ARGS ?= -q --no-cov
+
+# FFB_TEST_DATABASE_URL and friends come from .env when it exists. A value
+# already in the environment (CI points it at a service container) wins.
+ENV_TEST_DATABASE_URL := $(FFB_TEST_DATABASE_URL)
+-include .env
+ifneq ($(ENV_TEST_DATABASE_URL),)
+FFB_TEST_DATABASE_URL := $(ENV_TEST_DATABASE_URL)
+endif
+
+.PHONY: help pg-up pg-down pg-logs pg-psql pg-reset test test-pg test-all lint
+
+help:
+	@echo "pg-up     start Postgres + pgvector and wait until healthy"
+	@echo "pg-down   stop it (the data volume is kept)"
+	@echo "pg-reset  stop it and delete the data volume"
+	@echo "pg-logs   follow its logs"
+	@echo "pg-psql   psql into the test database"
+	@echo "test      test suite on SQLite only (Postgres cases skip)"
+	@echo "test-pg   test suite with the Postgres cases enabled (needs pg-up)"
+	@echo "test-all  both"
+	@echo "lint      ruff"
+
+# First run: copy the example and give it a random password.
+.env:
+	@pw=$$(openssl rand -hex 16); sed "s/change-me/$$pw/g" .env.example > .env
+	@echo "wrote .env from .env.example"
+
+pg-up: .env
+	$(COMPOSE) up -d --wait ffb-postgres
+	$(COMPOSE) logs --tail=20 ffb-postgres
+
+pg-down:
+	$(COMPOSE) down
+
+pg-reset:
+	$(COMPOSE) down -v
+
+pg-logs:
+	$(COMPOSE) logs -f ffb-postgres
+
+pg-psql:
+	$(COMPOSE) exec ffb-postgres psql -U "$(FFB_PG_USER)" -d "$(FFB_PG_DB)"
+
+test:
+	env -u FFB_TEST_DATABASE_URL $(PYTEST) $(PYTEST_ARGS)
+
+test-pg: .env
+	@test -n "$(FFB_TEST_DATABASE_URL)" || { echo "FFB_TEST_DATABASE_URL is not set (environment or .env)"; exit 1; }
+	@echo "$(PYTEST) $(PYTEST_ARGS)  (Postgres cases enabled)"
+	@FFB_TEST_DATABASE_URL="$(FFB_TEST_DATABASE_URL)" $(PYTEST) $(PYTEST_ARGS)
+
+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_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()`

docker-compose.yml

@@ -0,0 +1,31 @@
+# Postgres + pgvector for running the test suite against a real server.
+# Test-only: nothing here serves HTTP, so there is no caddy network.
+# The port is published on 127.0.0.1 only, for pytest running on the host.
+
+services:
+  ffb-postgres:
+    image: ${FFB_PG_IMAGE:-pgvector/pgvector:pg16}
+    restart: unless-stopped
+    environment:
+      POSTGRES_USER: ${FFB_PG_USER}
+      POSTGRES_PASSWORD: ${FFB_PG_PASSWORD}
+      POSTGRES_DB: ${FFB_PG_DB}
+    ports:
+      - "127.0.0.1:${FFB_PG_PORT:-55433}:5432"
+    volumes:
+      - ffb-postgres-data:/var/lib/postgresql/data
+      - ./docker/postgres-init:/docker-entrypoint-initdb.d:ro
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U \"$${POSTGRES_USER}\" -d \"$${POSTGRES_DB}\""]
+      interval: 2s
+      timeout: 3s
+      retries: 30
+      start_period: 5s
+    networks:
+      - internal
+
+volumes:
+  ffb-postgres-data:
+
+networks:
+  internal:

docker/postgres-init/01-vector.sql

@@ -0,0 +1,6 @@
+-- Runs once, on first start with an empty volume.
+-- The test fixtures create a fresh database per test from template1, so the
+-- extension goes there too; every database created afterwards inherits it.
+CREATE EXTENSION IF NOT EXISTS vector;
+\connect template1
+CREATE EXTENSION IF NOT EXISTS vector;

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,10 +52,14 @@ 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",
     "aiosqlite>=0.22.1",
+    "asyncpg>=0.31.0",
+    "pgvector>=0.5.0",
     "pytest>=8.2.0",
     "pytest-asyncio>=0.23.0",
     "pytest-cov>=5.0.0",
@@ -67,6 +71,8 @@ dev = [
 test = [
     "sqlalchemy[asyncio]>=2.0.43",
     "aiosqlite>=0.22.1",
+    "asyncpg>=0.31.0",
+    "pgvector>=0.5.0",
     "pytest>=8.2.0",
     "pytest-asyncio>=0.23.0",
     "pytest-cov>=5.0.0",
@@ -119,6 +125,7 @@ markers = [
     "integration: Integration tests for component interaction",
     "slow: Slow running tests",
     "database: Tests requiring database setup",
+    "postgres: Runs against the Postgres named by FFB_TEST_DATABASE_URL; skipped when unset",
 ]
 filterwarnings = [
     "ignore::DeprecationWarning",

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/_databases.py

@@ -0,0 +1,114 @@
+"""Fresh databases for DatabaseSink tests: SQLite always, Postgres when asked.
+
+Set FFB_TEST_DATABASE_URL to a ``postgresql+asyncpg://`` URL (``make test-pg``
+does this from .env) and every test using the ``db_url`` fixture also runs
+against that server. Each test gets its own uniquely named databases, created
+on first use and dropped at teardown, so no two tests share tables.
+"""
+
+import asyncio
+import os
+import re
+import threading
+import uuid
+from collections.abc import Coroutine
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.engine import URL, make_url
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import NullPool
+
+ENV_VAR = "FFB_TEST_DATABASE_URL"
+SKIP_REASON = f"{ENV_VAR} is not set; run `make pg-up` then `make test-pg`"
+
+
+def postgres_url() -> URL | None:
+    raw = os.environ.get(ENV_VAR, "").strip()
+    if not raw:
+        return None
+    url = make_url(raw)
+    if not url.drivername.startswith("postgresql"):
+        raise ValueError(f"{ENV_VAR} must be a postgresql URL, got {url.drivername}")
+    return url
+
+
+def run_blocking(coro: Coroutine[Any, Any, Any]) -> Any:
+    """Run ``coro`` to completion on its own loop in a worker thread.
+
+    The sink factories in the tests are plain functions called from inside a
+    running event loop, so the admin work cannot await on that loop.
+    """
+    box: dict[str, Any] = {}
+
+    def target() -> None:
+        try:
+            box["value"] = asyncio.run(coro)
+        except BaseException as exc:  # re-raised in the caller's thread
+            box["error"] = exc
+
+    thread = threading.Thread(target=target)
+    thread.start()
+    thread.join()
+    if "error" in box:
+        raise box["error"]
+    return box.get("value")
+
+
+class PostgresDatabases:
+    """Create databases on one server for a single test, then drop them all."""
+
+    def __init__(self, url: URL):
+        self.url = url
+        self.run_id = uuid.uuid4().hex[:10]
+        self.by_label: dict[str, str] = {}
+
+    def _admin(self, database: str | None = None):
+        url = self.url if database is None else self.url.set(database=database)
+        return create_async_engine(
+            url, isolation_level="AUTOCOMMIT", poolclass=NullPool
+        )
+
+    def url_for(self, label: str) -> str:
+        """URL of this test's database for ``label``, created on first request."""
+        if label not in self.by_label:
+            safe = re.sub(r"[^a-z0-9_]", "_", label.lower())
+            name = f"ffb_t_{self.run_id}_{safe}"[:63]
+            run_blocking(self._create(name))
+            self.by_label[label] = name
+        return self.url.set(database=self.by_label[label]).render_as_string(
+            hide_password=False
+        )
+
+    async def _create(self, name: str) -> None:
+        engine = self._admin()
+        try:
+            async with engine.connect() as conn:
+                await conn.execute(text(f'CREATE DATABASE "{name}"'))
+        finally:
+            await engine.dispose()
+        # The compose init script puts vector in template1 already; CI's
+        # service container has no init script, so make sure here.
+        engine = self._admin(name)
+        try:
+            async with engine.connect() as conn:
+                await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+        finally:
+            await engine.dispose()
+
+    def drop_all(self) -> None:
+        if self.by_label:
+            run_blocking(self._drop(list(self.by_label.values())))
+            self.by_label.clear()
+
+    async def _drop(self, names: list[str]) -> None:
+        engine = self._admin()
+        try:
+            async with engine.connect() as conn:
+                for name in names:
+                    # FORCE: a sink a test forgot to close must not keep it alive.
+                    await conn.execute(
+                        text(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')
+                    )
+        finally:
+            await engine.dispose()

tests/conftest.py

@@ -16,6 +16,8 @@ from fastmcp_feedback.database import FeedbackDatabase
 from fastmcp_feedback.insights import FeedbackInsights  
 from fastmcp_feedback.models import Base, Feedback, FeedbackType, FeedbackStatus
 
+from ._databases import SKIP_REASON, PostgresDatabases, postgres_url
+
 
 @pytest_asyncio.fixture
 async def event_loop():
@@ -177,6 +179,37 @@ def mock_database(mock_session):
     return db
 
 
+# Database backends for DatabaseSink tests
+
+@pytest.fixture(
+    params=["sqlite", pytest.param("postgres", marks=pytest.mark.postgres)]
+)
+def db_backend(request) -> str:
+    """Which database a ``db_url`` test runs on. Postgres needs FFB_TEST_DATABASE_URL."""
+    if request.param == "postgres" and postgres_url() is None:
+        pytest.skip(SKIP_REASON)
+    return request.param
+
+
+@pytest.fixture
+def db_url(db_backend, tmp_path):
+    """Factory: ``db_url(name)`` returns an async URL for an empty database.
+
+    The same name within one test returns the same database, so a test can
+    reopen it; different names and different tests never share one. SQLite
+    databases are files under tmp_path; Postgres ones are created on the
+    server named by FFB_TEST_DATABASE_URL and dropped after the test.
+    """
+    if db_backend == "sqlite":
+        yield lambda name="calls": f"sqlite+aiosqlite:///{tmp_path}/{name}.db"
+        return
+    databases = PostgresDatabases(postgres_url())
+    try:
+        yield lambda name="calls": databases.url_for(name)
+    finally:
+        databases.drop_all()
+
+
 # Environment helpers
 
 @pytest.fixture(autouse=True)

tests/unit/test_correlation.py

@@ -37,7 +37,7 @@ SAME_CLIENT_RULE = RULE_USER_WINDOW if FASTMCP_4 else RULE_SESSION
 
 
 @pytest.fixture
-async def db(tmp_path):
+async def db(db_url):
     """Make DatabaseSinks whose engines are always disposed, even on failure.
 
     An undisposed aiosqlite engine keeps a worker thread alive and the test
@@ -46,7 +46,7 @@ async def db(tmp_path):
     made = []
 
     def factory(name="calls", cls=DatabaseSink):
-        sink = cls(f"sqlite+aiosqlite:///{tmp_path}/{name}.db", create_tables=True)
+        sink = cls(db_url(name), create_tables=True)
         made.append(sink)
         return sink
 

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_events.py

@@ -14,7 +14,7 @@ import pytest
 from fastmcp import Client, FastMCP
 from fastmcp.exceptions import ToolError
 from sqlalchemy import MetaData, func, select
-from sqlalchemy.exc import OperationalError
+from sqlalchemy.exc import DBAPIError
 
 from fastmcp_feedback.instrumentation import (
     REDACTED,
@@ -30,13 +30,13 @@ from fastmcp_feedback.instrumentation import (
 
 
 @pytest.fixture
-async def db(tmp_path):
+async def db(db_url):
     """DatabaseSinks whose engines are always disposed, or the process hangs."""
     made = []
 
     def factory(name="calls", **kwargs):
         kwargs.setdefault("create_tables", True)
-        sink = DatabaseSink(f"sqlite+aiosqlite:///{tmp_path}/{name}.db", **kwargs)
+        sink = DatabaseSink(db_url(name), **kwargs)
         made.append(sink)
         return sink
 
@@ -437,7 +437,7 @@ class TestSinks:
         assert len(await table_rows(store, store.events)) == 1
         assert await table_rows(store, store.table) == []
 
-    async def test_calls_stored_when_events_table_missing(self, tmp_path, db):
+    async def test_calls_stored_when_events_table_missing(self, db):
         store = db("old", create_tables=False)
         old = MetaData()
         build_metadata("ffb_", old)
@@ -451,7 +451,8 @@ class TestSinks:
             ok=True,
             mode="meta",
         )
-        with pytest.raises(OperationalError, match="no such table"):
+        # SQLite raises OperationalError, Postgres ProgrammingError.
+        with pytest.raises(DBAPIError, match="no such table|does not exist"):
             await store.write([rec, event()])
         assert len(await table_rows(store, store.table)) == 1
 

tests/unit/test_instrumentation.py

@@ -373,9 +373,8 @@ class TestSinks:
 
 
 class TestDatabaseSink:
-    async def test_end_to_end_with_created_tables(self, tmp_path):
-        url = f"sqlite+aiosqlite:///{tmp_path}/calls.db"
-        db = DatabaseSink(url, create_tables=True)
+    async def test_end_to_end_with_created_tables(self, db_url):
+        db = DatabaseSink(db_url(), create_tables=True)
         app, _, mw = make_app(sinks=[db], mode="full", server_version="9.9")
         await call(app, "create_token", {"label": "ci"})
         with pytest.raises(ToolError):
@@ -397,8 +396,8 @@ class TestDatabaseSink:
         assert rows[0]["server_version"] == "9.9"
         assert "hunter2" not in json.dumps([dict(r) for r in rows], default=str)
 
-    async def test_prefix_and_host_owned_engine(self, tmp_path):
-        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
+    async def test_prefix_and_host_owned_engine(self, db_url):
+        engine = create_async_engine(db_url("host"))
         db = DatabaseSink(engine, prefix="qa_", create_tables=True)
         app, _, mw = make_app(sinks=[db])
         await call(app, "echo", {"text": "hi"})
@@ -414,8 +413,8 @@ class TestDatabaseSink:
             "qa_tool_calls",
         ]
 
-    async def test_no_tables_created_by_default(self, tmp_path):
-        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
+    async def test_no_tables_created_by_default(self, db_url):
+        engine = create_async_engine(db_url("host"))
         app, _, mw = make_app(sinks=[DatabaseSink(engine)])
         result = await call(app, "echo", {"text": "hi"})
         await mw.flush()
@@ -429,9 +428,9 @@ class TestDatabaseSink:
         assert tables == []
         assert mw.dispatcher.sink_errors == 1
 
-    async def test_absent_payloads_are_sql_null(self, tmp_path):
+    async def test_absent_payloads_are_sql_null(self, db_url):
         """Meta mode stores no payloads; they must be SQL NULL, not JSON null."""
-        db = DatabaseSink(f"sqlite+aiosqlite:///{tmp_path}/n.db", create_tables=True)
+        db = DatabaseSink(db_url("n"), create_tables=True)
         app, _, mw = make_app(sinks=[db])
         await call(app, "echo", {"text": "hi"})
         await mw.flush()
@@ -449,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",

tests/unit/test_retention_sampling.py

@@ -21,14 +21,12 @@ from fastmcp_feedback.instrumentation import (
 
 
 @pytest.fixture
-async def db(tmp_path):
+async def db(db_url):
     """DatabaseSinks whose engines are always disposed, or the process hangs."""
     made = []
 
     def factory(name="calls", **kwargs):
-        sink = DatabaseSink(
-            f"sqlite+aiosqlite:///{tmp_path}/{name}.db", create_tables=True, **kwargs
-        )
+        sink = DatabaseSink(db_url(name), create_tables=True, **kwargs)
         made.append(sink)
         return sink
 

tests/unit/test_soft_errors.py

@@ -39,14 +39,12 @@ except TypeError:  # added in FastMCP 3.4
 
 
 @pytest.fixture
-async def db(tmp_path):
+async def db(db_url):
     """DatabaseSinks whose engines are always disposed, or the process hangs."""
     made = []
 
     def factory(name="calls"):
-        sink = DatabaseSink(
-            f"sqlite+aiosqlite:///{tmp_path}/{name}.db", create_tables=True
-        )
+        sink = DatabaseSink(db_url(name), create_tables=True)
         made.append(sink)
         return sink
 

tests/unit/test_test_databases.py

@@ -0,0 +1,70 @@
+"""Checks on the db_url fixture itself: isolation between tests, and pgvector."""
+
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import func, select, text
+
+from fastmcp_feedback.instrumentation import DatabaseSink, ToolCallRecord
+
+SHARED_ID = "00000000-0000-0000-0000-00000000f00d"
+
+
+@pytest.fixture
+async def sink(db_url):
+    made = DatabaseSink(db_url(), create_tables=True)
+    yield made
+    await made.engine.dispose()
+
+
+async def write_fixed_row(sink):
+    """Insert a row with a fixed primary key and return the table's row count.
+
+    Two tests that did this against one shared database would collide on the
+    key, so each passing proves it had a database of its own.
+    """
+    rec = ToolCallRecord(
+        id=SHARED_ID,
+        tool="t",
+        started_at=datetime.now(UTC),
+        duration_ms=1.0,
+        ok=True,
+        mode="meta",
+    )
+    await sink.write([rec])
+    async with sink.engine.connect() as conn:
+        return (
+            await conn.execute(select(func.count()).select_from(sink.table))
+        ).scalar()
+
+
+class TestIsolation:
+    async def test_first_writer(self, sink):
+        assert await write_fixed_row(sink) == 1
+
+    async def test_second_writer(self, sink):
+        assert await write_fixed_row(sink) == 1
+
+    def test_names_map_to_databases(self, db_url):
+        assert db_url("a") == db_url("a")
+        assert db_url("a") != db_url("b")
+
+
+async def test_vector_extension_available(db_url, db_backend):
+    if db_backend != "postgres":
+        pytest.skip("pgvector is Postgres only")
+    sink = DatabaseSink(db_url())
+    try:
+        async with sink.engine.connect() as conn:
+            version = (
+                await conn.execute(
+                    text("SELECT extversion FROM pg_extension WHERE extname = 'vector'")
+                )
+            ).scalar()
+            distance = (
+                await conn.execute(text("SELECT '[1,2,3]'::vector <-> '[1,2,4]'"))
+            ).scalar()
+    finally:
+        await sink.engine.dispose()
+    assert version is not None
+    assert distance == 1.0

uv.lock

@@ -51,6 +51,54 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/38/0e/27be9fdef66e72d64c0cdc3cc2823101b80585f8119b5c112c2e8f5f7dab/anyio-4.12.1-py3-none-any.whl", hash = "sha256:d405828884fc140aa80a3c667b8beed277f1dfedec42ba031bd6ac3db606ab6c", size = 113592, upload-time = "2026-01-06T11:45:19.497Z" },
 ]
 
+[[package]]
+name = "asyncpg"
+version = "0.31.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/fe/cc/d18065ce2380d80b1bcce927c24a2642efd38918e33fd724bc4bca904877/asyncpg-0.31.0.tar.gz", hash = "sha256:c989386c83940bfbd787180f2b1519415e2d3d6277a70d9d0f0145ac73500735", size = 993667, upload-time = "2025-11-24T23:27:00.812Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/08/17/cc02bc49bc350623d050fa139e34ea512cd6e020562f2a7312a7bcae4bc9/asyncpg-0.31.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:eee690960e8ab85063ba93af2ce128c0f52fd655fdff9fdb1a28df01329f031d", size = 643159, upload-time = "2025-11-24T23:25:36.443Z" },
+    { url = "https://files.pythonhosted.org/packages/a4/62/4ded7d400a7b651adf06f49ea8f73100cca07c6df012119594d1e3447aa6/asyncpg-0.31.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:2657204552b75f8288de08ca60faf4a99a65deef3a71d1467454123205a88fab", size = 638157, upload-time = "2025-11-24T23:25:37.89Z" },
+    { url = "https://files.pythonhosted.org/packages/d6/5b/4179538a9a72166a0bf60ad783b1ef16efb7960e4d7b9afe9f77a5551680/asyncpg-0.31.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a429e842a3a4b4ea240ea52d7fe3f82d5149853249306f7ff166cb9948faa46c", size = 2918051, upload-time = "2025-11-24T23:25:39.461Z" },
+    { url = "https://files.pythonhosted.org/packages/e6/35/c27719ae0536c5b6e61e4701391ffe435ef59539e9360959240d6e47c8c8/asyncpg-0.31.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c0807be46c32c963ae40d329b3a686356e417f674c976c07fa49f1b30303f109", size = 2972640, upload-time = "2025-11-24T23:25:41.512Z" },
+    { url = "https://files.pythonhosted.org/packages/43/f4/01ebb9207f29e645a64699b9ce0eefeff8e7a33494e1d29bb53736f7766b/asyncpg-0.31.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:e5d5098f63beeae93512ee513d4c0c53dc12e9aa2b7a1af5a81cddf93fe4e4da", size = 2851050, upload-time = "2025-11-24T23:25:43.153Z" },
+    { url = "https://files.pythonhosted.org/packages/3e/f4/03ff1426acc87be0f4e8d40fa2bff5c3952bef0080062af9efc2212e3be8/asyncpg-0.31.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:37fc6c00a814e18eef51833545d1891cac9aa69140598bb076b4cd29b3e010b9", size = 2962574, upload-time = "2025-11-24T23:25:44.942Z" },
+    { url = "https://files.pythonhosted.org/packages/c7/39/cc788dfca3d4060f9d93e67be396ceec458dfc429e26139059e58c2c244d/asyncpg-0.31.0-cp311-cp311-win32.whl", hash = "sha256:5a4af56edf82a701aece93190cc4e094d2df7d33f6e915c222fb09efbb5afc24", size = 521076, upload-time = "2025-11-24T23:25:46.486Z" },
+    { url = "https://files.pythonhosted.org/packages/28/fc/735af5384c029eb7f1ca60ccb8fa95521dbdaeef788edf4cecfc604c3cab/asyncpg-0.31.0-cp311-cp311-win_amd64.whl", hash = "sha256:480c4befbdf079c14c9ca43c8c5e1fe8b6296c96f1f927158d4f1e750aacc047", size = 584980, upload-time = "2025-11-24T23:25:47.938Z" },
+    { url = "https://files.pythonhosted.org/packages/2a/a6/59d0a146e61d20e18db7396583242e32e0f120693b67a8de43f1557033e2/asyncpg-0.31.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:b44c31e1efc1c15188ef183f287c728e2046abb1d26af4d20858215d50d91fad", size = 662042, upload-time = "2025-11-24T23:25:49.578Z" },
+    { url = "https://files.pythonhosted.org/packages/36/01/ffaa189dcb63a2471720615e60185c3f6327716fdc0fc04334436fbb7c65/asyncpg-0.31.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:0c89ccf741c067614c9b5fc7f1fc6f3b61ab05ae4aaa966e6fd6b93097c7d20d", size = 638504, upload-time = "2025-11-24T23:25:51.501Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/62/3f699ba45d8bd24c5d65392190d19656d74ff0185f42e19d0bbd973bb371/asyncpg-0.31.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:12b3b2e39dc5470abd5e98c8d3373e4b1d1234d9fbdedf538798b2c13c64460a", size = 3426241, upload-time = "2025-11-24T23:25:53.278Z" },
+    { url = "https://files.pythonhosted.org/packages/8c/d1/a867c2150f9c6e7af6462637f613ba67f78a314b00db220cd26ff559d532/asyncpg-0.31.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:aad7a33913fb8bcb5454313377cc330fbb19a0cd5faa7272407d8a0c4257b671", size = 3520321, upload-time = "2025-11-24T23:25:54.982Z" },
+    { url = "https://files.pythonhosted.org/packages/7a/1a/cce4c3f246805ecd285a3591222a2611141f1669d002163abef999b60f98/asyncpg-0.31.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:3df118d94f46d85b2e434fd62c84cb66d5834d5a890725fe625f498e72e4d5ec", size = 3316685, upload-time = "2025-11-24T23:25:57.43Z" },
+    { url = "https://files.pythonhosted.org/packages/40/ae/0fc961179e78cc579e138fad6eb580448ecae64908f95b8cb8ee2f241f67/asyncpg-0.31.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:bd5b6efff3c17c3202d4b37189969acf8927438a238c6257f66be3c426beba20", size = 3471858, upload-time = "2025-11-24T23:25:59.636Z" },
+    { url = "https://files.pythonhosted.org/packages/52/b2/b20e09670be031afa4cbfabd645caece7f85ec62d69c312239de568e058e/asyncpg-0.31.0-cp312-cp312-win32.whl", hash = "sha256:027eaa61361ec735926566f995d959ade4796f6a49d3bde17e5134b9964f9ba8", size = 527852, upload-time = "2025-11-24T23:26:01.084Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/f0/f2ed1de154e15b107dc692262395b3c17fc34eafe2a78fc2115931561730/asyncpg-0.31.0-cp312-cp312-win_amd64.whl", hash = "sha256:72d6bdcbc93d608a1158f17932de2321f68b1a967a13e014998db87a72ed3186", size = 597175, upload-time = "2025-11-24T23:26:02.564Z" },
+    { url = "https://files.pythonhosted.org/packages/95/11/97b5c2af72a5d0b9bc3fa30cd4b9ce22284a9a943a150fdc768763caf035/asyncpg-0.31.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:c204fab1b91e08b0f47e90a75d1b3c62174dab21f670ad6c5d0f243a228f015b", size = 661111, upload-time = "2025-11-24T23:26:04.467Z" },
+    { url = "https://files.pythonhosted.org/packages/1b/71/157d611c791a5e2d0423f09f027bd499935f0906e0c2a416ce712ba51ef3/asyncpg-0.31.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:54a64f91839ba59008eccf7aad2e93d6e3de688d796f35803235ea1c4898ae1e", size = 636928, upload-time = "2025-11-24T23:26:05.944Z" },
+    { url = "https://files.pythonhosted.org/packages/2e/fc/9e3486fb2bbe69d4a867c0b76d68542650a7ff1574ca40e84c3111bb0c6e/asyncpg-0.31.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c0e0822b1038dc7253b337b0f3f676cadc4ac31b126c5d42691c39691962e403", size = 3424067, upload-time = "2025-11-24T23:26:07.957Z" },
+    { url = "https://files.pythonhosted.org/packages/12/c6/8c9d076f73f07f995013c791e018a1cd5f31823c2a3187fc8581706aa00f/asyncpg-0.31.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:bef056aa502ee34204c161c72ca1f3c274917596877f825968368b2c33f585f4", size = 3518156, upload-time = "2025-11-24T23:26:09.591Z" },
+    { url = "https://files.pythonhosted.org/packages/ae/3b/60683a0baf50fbc546499cfb53132cb6835b92b529a05f6a81471ab60d0c/asyncpg-0.31.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:0bfbcc5b7ffcd9b75ab1558f00db2ae07db9c80637ad1b2469c43df79d7a5ae2", size = 3319636, upload-time = "2025-11-24T23:26:11.168Z" },
+    { url = "https://files.pythonhosted.org/packages/50/dc/8487df0f69bd398a61e1792b3cba0e47477f214eff085ba0efa7eac9ce87/asyncpg-0.31.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:22bc525ebbdc24d1261ecbf6f504998244d4e3be1721784b5f64664d61fbe602", size = 3472079, upload-time = "2025-11-24T23:26:13.164Z" },
+    { url = "https://files.pythonhosted.org/packages/13/a1/c5bbeeb8531c05c89135cb8b28575ac2fac618bcb60119ee9696c3faf71c/asyncpg-0.31.0-cp313-cp313-win32.whl", hash = "sha256:f890de5e1e4f7e14023619399a471ce4b71f5418cd67a51853b9910fdfa73696", size = 527606, upload-time = "2025-11-24T23:26:14.78Z" },
+    { url = "https://files.pythonhosted.org/packages/91/66/b25ccb84a246b470eb943b0107c07edcae51804912b824054b3413995a10/asyncpg-0.31.0-cp313-cp313-win_amd64.whl", hash = "sha256:dc5f2fa9916f292e5c5c8b2ac2813763bcd7f58e130055b4ad8a0531314201ab", size = 596569, upload-time = "2025-11-24T23:26:16.189Z" },
+    { url = "https://files.pythonhosted.org/packages/3c/36/e9450d62e84a13aea6580c83a47a437f26c7ca6fa0f0fd40b6670793ea30/asyncpg-0.31.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:f6b56b91bb0ffc328c4e3ed113136cddd9deefdf5f79ab448598b9772831df44", size = 660867, upload-time = "2025-11-24T23:26:17.631Z" },
+    { url = "https://files.pythonhosted.org/packages/82/4b/1d0a2b33b3102d210439338e1beea616a6122267c0df459ff0265cd5807a/asyncpg-0.31.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:334dec28cf20d7f5bb9e45b39546ddf247f8042a690bff9b9573d00086e69cb5", size = 638349, upload-time = "2025-11-24T23:26:19.689Z" },
+    { url = "https://files.pythonhosted.org/packages/41/aa/e7f7ac9a7974f08eff9183e392b2d62516f90412686532d27e196c0f0eeb/asyncpg-0.31.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:98cc158c53f46de7bb677fd20c417e264fc02b36d901cc2a43bd6cb0dc6dbfd2", size = 3410428, upload-time = "2025-11-24T23:26:21.275Z" },
+    { url = "https://files.pythonhosted.org/packages/6f/de/bf1b60de3dede5c2731e6788617a512bc0ebd9693eac297ee74086f101d7/asyncpg-0.31.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9322b563e2661a52e3cdbc93eed3be7748b289f792e0011cb2720d278b366ce2", size = 3471678, upload-time = "2025-11-24T23:26:23.627Z" },
+    { url = "https://files.pythonhosted.org/packages/46/78/fc3ade003e22d8bd53aaf8f75f4be48f0b460fa73738f0391b9c856a9147/asyncpg-0.31.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:19857a358fc811d82227449b7ca40afb46e75b33eb8897240c3839dd8b744218", size = 3313505, upload-time = "2025-11-24T23:26:25.235Z" },
+    { url = "https://files.pythonhosted.org/packages/bf/e9/73eb8a6789e927816f4705291be21f2225687bfa97321e40cd23055e903a/asyncpg-0.31.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ba5f8886e850882ff2c2ace5732300e99193823e8107e2c53ef01c1ebfa1e85d", size = 3434744, upload-time = "2025-11-24T23:26:26.944Z" },
+    { url = "https://files.pythonhosted.org/packages/08/4b/f10b880534413c65c5b5862f79b8e81553a8f364e5238832ad4c0af71b7f/asyncpg-0.31.0-cp314-cp314-win32.whl", hash = "sha256:cea3a0b2a14f95834cee29432e4ddc399b95700eb1d51bbc5bfee8f31fa07b2b", size = 532251, upload-time = "2025-11-24T23:26:28.404Z" },
+    { url = "https://files.pythonhosted.org/packages/d3/2d/7aa40750b7a19efa5d66e67fc06008ca0f27ba1bd082e457ad82f59aba49/asyncpg-0.31.0-cp314-cp314-win_amd64.whl", hash = "sha256:04d19392716af6b029411a0264d92093b6e5e8285ae97a39957b9a9c14ea72be", size = 604901, upload-time = "2025-11-24T23:26:30.34Z" },
+    { url = "https://files.pythonhosted.org/packages/ce/fe/b9dfe349b83b9dee28cc42360d2c86b2cdce4cb551a2c2d27e156bcac84d/asyncpg-0.31.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:bdb957706da132e982cc6856bb2f7b740603472b54c3ebc77fe60ea3e57e1bd2", size = 702280, upload-time = "2025-11-24T23:26:32Z" },
+    { url = "https://files.pythonhosted.org/packages/6a/81/e6be6e37e560bd91e6c23ea8a6138a04fd057b08cf63d3c5055c98e81c1d/asyncpg-0.31.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:6d11b198111a72f47154fa03b85799f9be63701e068b43f84ac25da0bda9cb31", size = 682931, upload-time = "2025-11-24T23:26:33.572Z" },
+    { url = "https://files.pythonhosted.org/packages/a6/45/6009040da85a1648dd5bc75b3b0a062081c483e75a1a29041ae63a0bf0dc/asyncpg-0.31.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:18c83b03bc0d1b23e6230f5bf8d4f217dc9bc08644ce0502a9d91dc9e634a9c7", size = 3581608, upload-time = "2025-11-24T23:26:35.638Z" },
+    { url = "https://files.pythonhosted.org/packages/7e/06/2e3d4d7608b0b2b3adbee0d0bd6a2d29ca0fc4d8a78f8277df04e2d1fd7b/asyncpg-0.31.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:e009abc333464ff18b8f6fd146addffd9aaf63e79aa3bb40ab7a4c332d0c5e9e", size = 3498738, upload-time = "2025-11-24T23:26:37.275Z" },
+    { url = "https://files.pythonhosted.org/packages/7d/aa/7d75ede780033141c51d83577ea23236ba7d3a23593929b32b49db8ed36e/asyncpg-0.31.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:3b1fbcb0e396a5ca435a8826a87e5c2c2cc0c8c68eb6fadf82168056b0e53a8c", size = 3401026, upload-time = "2025-11-24T23:26:39.423Z" },
+    { url = "https://files.pythonhosted.org/packages/ba/7a/15e37d45e7f7c94facc1e9148c0e455e8f33c08f0b8a0b1deb2c5171771b/asyncpg-0.31.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:8df714dba348efcc162d2adf02d213e5fab1bd9f557e1305633e851a61814a7a", size = 3429426, upload-time = "2025-11-24T23:26:41.032Z" },
+    { url = "https://files.pythonhosted.org/packages/13/d5/71437c5f6ae5f307828710efbe62163974e71237d5d46ebd2869ea052d10/asyncpg-0.31.0-cp314-cp314t-win32.whl", hash = "sha256:1b41f1afb1033f2b44f3234993b15096ddc9cd71b21a42dbd87fc6a57b43d65d", size = 614495, upload-time = "2025-11-24T23:26:42.659Z" },
+    { url = "https://files.pythonhosted.org/packages/3c/d7/8fb3044eaef08a310acfe23dae9a8e2e07d305edc29a53497e52bc76eca7/asyncpg-0.31.0-cp314-cp314t-win_amd64.whl", hash = "sha256:bd4107bb7cdd0e9e65fae66a62afd3a249663b844fa34d479f6d5b3bef9c04c3", size = 706062, upload-time = "2025-11-24T23:26:44.086Z" },
+]
+
 [[package]]
 name = "attrs"
 version = "25.4.0"
@@ -553,7 +601,7 @@ wheels = [
 
 [[package]]
 name = "fastmcp-feedback"
-version = "2026.9.29.1"
+version = "2026.9.30"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },
@@ -564,7 +612,9 @@ dependencies = [
 [package.optional-dependencies]
 all = [
     { name = "aiosqlite" },
+    { name = "asyncpg" },
     { name = "mypy" },
+    { name = "pgvector" },
     { name = "psycopg2-binary" },
     { name = "pymysql" },
     { name = "pytest" },
@@ -577,7 +627,9 @@ all = [
 ]
 dev = [
     { name = "aiosqlite" },
+    { name = "asyncpg" },
     { name = "mypy" },
+    { name = "pgvector" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-cov" },
@@ -586,6 +638,9 @@ dev = [
     { name = "ruff" },
     { name = "sqlalchemy", extra = ["asyncio"] },
 ]
+embeddings = [
+    { name = "pgvector" },
+]
 instrumentation = [
     { name = "aiosqlite" },
     { name = "sqlalchemy", extra = ["asyncio"] },
@@ -598,6 +653,8 @@ postgresql = [
 ]
 test = [
     { name = "aiosqlite" },
+    { name = "asyncpg" },
+    { name = "pgvector" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-cov" },
@@ -611,9 +668,14 @@ requires-dist = [
     { name = "aiosqlite", marker = "extra == 'dev'", specifier = ">=0.22.1" },
     { name = "aiosqlite", marker = "extra == 'instrumentation'", specifier = ">=0.22.1" },
     { name = "aiosqlite", marker = "extra == 'test'", specifier = ">=0.22.1" },
+    { name = "asyncpg", marker = "extra == 'dev'", specifier = ">=0.31.0" },
+    { name = "asyncpg", marker = "extra == 'test'", specifier = ">=0.31.0" },
     { name = "fastmcp", specifier = ">=2.12.2,<5" },
     { 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" },
     { name = "pymysql", marker = "extra == 'mysql'", specifier = ">=1.1.0" },
@@ -633,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"
@@ -1264,6 +1326,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/32/2b/121e912bd60eebd623f873fd090de0e84f322972ab25a7f9044c056804ed/pathspec-1.0.3-py3-none-any.whl", hash = "sha256:e80767021c1cc524aa3fb14bedda9c34406591343cc42797b386ce7b9354fb6c", size = 55021, upload-time = "2026-01-09T15:46:44.652Z" },
 ]
 
+[[package]]
+name = "pgvector"
+version = "0.5.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/a7/ec/6eb80aebc728200f95229219882994c1b0585b956ca47da5edb9d062627a/pgvector-0.5.0.tar.gz", hash = "sha256:07a9dcf735696879406983afc6eba9a787cef7c0cf6c367ca1a5779f036dee74", size = 35170, upload-time = "2026-07-06T18:27:27.767Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5c/e4/a5573f2c579ca9ad133293bfb624148ba0893674ca4a6eeec85ced9a6a09/pgvector-0.5.0-py3-none-any.whl", hash = "sha256:fedc9800894e6da2be51358d7b7c574bf34f247ca741a5a09513622135f5964f", size = 30958, upload-time = "2026-07-06T18:27:26.797Z" },
+]
+
 [[package]]
 name = "platformdirs"
 version = "4.5.1"