Merge feature/backfill: backfill embeddings, flatten list and dict text

CHANGELOG.md

@@ -5,6 +5,49 @@ 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.1] - 2026-09-30
+
+Embed what was stored before embedding was switched on, and embed list and
+dict attrs as text rather than as Python reprs.
+
+### Added
+- `await backfill_embeddings(embedding_sink, *, sources=None, since=None,
+  until=None, limit=None, batch_size=500, dry_run=False)`. Reads the
+  `DatabaseSink`'s tables for calls with outcome `error` or `soft_error`
+  (and, for rows from before the `outcome` column, outcome NULL with an
+  `error_message`), `feedback.submitted` events, `llm.call` events with
+  prompt or completion, and other events with any of the sink's
+  `event_text_keys`. Each record goes through the live sink's extraction,
+  redactor and `max_chars`, so text, `text_hash` and `source_id` match and
+  later records dedupe against backfilled rows. Idempotent: a second run
+  embeds nothing. Keyset-paged by timestamp and id, `batch_size` rows per
+  query; embeds in chunks of the sink's `batch_size`. `limit` embeds at most
+  that many texts, oldest first. Returns counts per source type plus
+  `skipped` and `errors`; embed failures are counted, never raised. Raises
+  only for bad arguments or a missing table. `sources` defaults to the
+  sink's.
+- `await backfill_feedback(embedding_sink, items, *, batch_size=500,
+  dry_run=False)` for feedback in the host's own tables, as
+  `(ref, title, description)` tuples; the same text a `feedback.submitted`
+  event with that ref gives.
+- `python -m fastmcp_feedback.instrumentation.backfill --database-url ...
+  --embed-base-url ... [--model --dim --prefix --sources --since --until
+  --limit --batch-size --max-chars --event-text-keys --create-tables
+  --dry-run]`. The API key comes from `FFB_EMBED_API_KEY` or the variable
+  named by `--api-key-env`. Prints counts as JSON; exits 1 when any text
+  failed, 2 on a setup error.
+
+### Changed
+- **Embedded text of list, tuple and dict values.** Lists and tuples are
+  embedded one cleaned, non-empty item per line instead of as their repr
+  (`"['a', 'b']"`); dicts as `key: value` lines, sorted by key because
+  PostgreSQL JSONB does not keep key order and a backfill must rebuild the
+  same text. Nested values are flattened the same way, to a depth of 32 and
+  about 64k characters before `max_chars` applies. This applies to event
+  text keys, feedback title and description, and LLM prompt and completion.
+  Existing rows embedded from such values keep their old text; a backfill
+  run replaces them, since their text changed.
+
 ## [2026.09.30] - 2026-09-30
 
 Embed feedback, tool errors, event text and (opt-in) LLM text, and search

Makefile

@@ -60,4 +60,4 @@ 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
+		tests/unit/test_embeddings.py tests/unit/test_backfill.py

README.md

@@ -570,7 +570,9 @@ you can search by meaning. It embeds four kinds of text:
 | `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.
+failed LLM call's `error` is still embedded. A list or tuple value is
+embedded one item per line, a dict as `key: value` lines sorted by key (at
+any depth), rather than as its Python repr.
 
 It needs a `DatabaseSink` to store into and an embedder. `OpenAIEmbedder`
 talks to any OpenAI-compatible `/v1/embeddings` endpoint (OpenAI, Ollama,
@@ -707,6 +709,71 @@ Wrap it in `op.execute(...)` in the migration's `upgrade()`, and
 renders the column as `fastmcp_feedback.instrumentation.storage.EmbeddingVector(dim=1024)`,
 which ties the migration to this package.
 
+#### Backfilling existing data
+
+The sink embeds records as they arrive. Calls, events and feedback stored
+before it was running can be embedded afterwards with `backfill_embeddings`,
+which reads them back from the `DatabaseSink`'s tables:
+
+```python
+from fastmcp_feedback.instrumentation import backfill_embeddings
+
+counts = await backfill_embeddings(mw.embedding_sink)
+# {"feedback": 4, "call_error": 90, "event": 12, "llm": 0, "skipped": 0, "errors": 0}
+```
+
+It picks up the same four kinds of text as the live sink: calls with outcome
+`error` or `soft_error`, `feedback.submitted` events, `llm.call` events with
+a prompt or completion, and other events with any of the sink's
+`event_text_keys`. Calls recorded before the `outcome` column existed
+(2026.09.27.4) have it NULL; those with an `error_message` count as errors.
+
+Each text goes through the sink's own extraction, redactor and `max_chars`,
+so it comes out with the same text, `text_hash` and `source_id` the live sink
+would give that record. That makes it idempotent: a second run embeds nothing
+and reports the rows as `skipped`, and records arriving later dedupe against
+what was backfilled. When the middleware has a custom `Redactor`, build the
+sink with it (`EmbeddingSink(..., redactor=mw.redactor)`), or use
+`mw.embedding_sink`, which already has it.
+
+`sources=` narrows it (default: the sink's sources), `since=` and `until=`
+select rows by timestamp (naive means UTC), and `limit=` embeds at most that
+many texts, oldest first; running it again carries on where it stopped. Rows
+are read `batch_size` (500) at a time and embedded in chunks of the sink's
+`batch_size`, so a large table never sits in memory. A failed chunk is
+logged and counted in `errors`, never raised; only a missing table or bad
+arguments raise. `dry_run=True` counts what would be embedded without calling
+the embedder.
+
+Feedback kept in your own tables rather than as events goes through
+`backfill_feedback`, which makes the same text a `feedback.submitted` event
+with that ref would:
+
+```python
+from fastmcp_feedback.instrumentation import backfill_feedback
+
+rows = [("bug-7Q2X", "Render hangs", "Stuck at 99%"), ("bug-8K1D", "Crash", None)]
+await backfill_feedback(mw.embedding_sink, rows)  # (ref, title, description)
+```
+
+From a shell, the same over a database URL. The API key comes from
+`FFB_EMBED_API_KEY` (or the variable `--api-key-env` names), never a flag:
+
+```bash
+export FFB_EMBED_API_KEY=...
+python -m fastmcp_feedback.instrumentation.backfill \
+    --database-url postgresql+asyncpg://user:pass@db/app \
+    --embed-base-url https://api.example.com/v1 \
+    --model mxbai-embed-large --dim 1024 --prefix ffb_ \
+    --since 2026-09-01 --sources call_error,feedback --dry-run
+```
+
+It prints the counts as JSON and exits 1 if any text failed to embed, 2 on
+a setup error. `--create-tables` creates `ffb_embeddings` (and on PostgreSQL
+the vector extension) when it is missing; without it a missing table is an
+error. `--help` lists the rest (`--until`, `--limit`, `--batch-size`,
+`--max-chars`, `--event-text-keys`).
+
 ### Middleware order and shutdown
 
 FastMCP runs the first middleware added as the outermost. Call `instrument()`

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "fastmcp-feedback"
-version = "2026.09.30"  # YYYYMMDD.patch versioning - update for each release
+version = "2026.09.30.1"  # YYYYMMDD.patch versioning - update for each release
 description = "Production-ready feedback collection system for FastMCP servers"
 readme = "README.md"
 license = "MIT"

src/fastmcp_feedback/__init__.py

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

src/fastmcp_feedback/instrumentation/__init__.py

@@ -10,8 +10,10 @@ 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")
+await backfill_embeddings(mw.embedding_sink)  # rows stored before embedding
 """
 
+from .backfill import backfill_embeddings, backfill_feedback
 from .classify import (
     DEFAULT_RESULT_CLASSIFIER,
     ResultClassifier,
@@ -72,6 +74,8 @@ __all__ = [
     "ResultClassifier",
     "Sink",
     "ToolCallRecord",
+    "backfill_embeddings",
+    "backfill_feedback",
     "build_metadata",
     "extract_texts",
     "instrument",

src/fastmcp_feedback/instrumentation/backfill/__init__.py

@@ -0,0 +1,338 @@
+"""Embed records that were stored before an EmbeddingSink was running.
+
+``backfill_embeddings`` reads tool-call errors and events back out of the
+DatabaseSink's tables, rebuilds the records, and sends them through the same
+extraction, redaction and truncation as ``EmbeddingSink.write``. The texts,
+hashes and source ids therefore match what the live sink would produce, so
+records arriving later dedupe against backfilled rows, and a second run
+embeds nothing new.
+
+Command line: ``python -m fastmcp_feedback.instrumentation.backfill --help``.
+"""
+
+import asyncio
+import logging
+from collections.abc import AsyncIterator, Iterable
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import and_, inspect, or_, select, tuple_
+
+from ..embeddings import EmbeddingSink, extract_texts
+from ..events import EventRecord
+from ..record import ToolCallRecord
+from ..sinks import Record
+from ..storage import EMBEDDING_SOURCES
+
+logger = logging.getLogger(__name__)
+
+FEEDBACK_KIND = "feedback.submitted"
+LLM_KIND = "llm.call"
+
+_CALL_COLUMNS = (
+    "id",
+    "tool",
+    "started_at",
+    "duration_ms",
+    "mode",
+    "ok",
+    "error_type",
+    "error_message",
+)
+
+
+def _utc(value: datetime | None) -> datetime | None:
+    if value is None or value.tzinfo is None:
+        return value if value is None else value.replace(tzinfo=UTC)
+    return value.astimezone(UTC)
+
+
+class _Run:
+    """Counts and limits for one backfill call."""
+
+    def __init__(
+        self,
+        sink: EmbeddingSink,
+        sources: Iterable[str],
+        limit: int | None,
+        dry_run: bool,
+    ):
+        self.sink = sink
+        self.sources = tuple(sources)
+        self.limit = limit
+        self.dry_run = dry_run
+        self.counts = dict.fromkeys(self.sources, 0)
+        self.skipped = 0
+        self.errors = 0
+        self.sent = 0
+        self.have_table = True
+
+    @property
+    def done(self) -> bool:
+        return self.limit is not None and self.sent >= self.limit
+
+    def result(self) -> dict[str, int]:
+        return {**self.counts, "skipped": self.skipped, "errors": self.errors}
+
+    async def feed(self, records: list[Record]) -> None:
+        sink = self.sink
+        items = [
+            sink._prepare(i)
+            for i in extract_texts(
+                records, sources=self.sources, event_text_keys=sink.event_text_keys
+            )
+        ]
+        # The last text for a source wins, as in the live sink.
+        latest = list({(i.source_type, i.source_id): i for i in items}.values())
+        if self.have_table:
+            todo, stored = await sink._unembedded(latest)
+        else:
+            todo, stored = latest, set()
+        self.skipped += len(latest) - len(todo)
+        if self.limit is not None:
+            todo = todo[: max(0, self.limit - self.sent)]
+        self.sent += len(todo)
+        if self.dry_run:
+            for item in todo:
+                self.counts[item.source_type] += 1
+            return
+        for start in range(0, len(todo), sink.batch_size):
+            chunk = todo[start : start + sink.batch_size]
+            try:
+                vectors = await asyncio.wait_for(
+                    sink.embedder.embed([i.text for i in chunk]), sink.timeout
+                )
+                await sink._store(chunk, vectors, stored)
+            except Exception:
+                self.errors += len(chunk)
+                logger.warning(
+                    "Backfill failed to embed %d texts with %s",
+                    len(chunk),
+                    sink.embedder.model,
+                    exc_info=True,
+                )
+                continue
+            for item in chunk:
+                self.counts[item.source_type] += 1
+
+
+async def _has_table(engine: Any, name: str) -> bool:
+    async with engine.connect() as conn:
+        return await conn.run_sync(lambda sync: inspect(sync).has_table(name))
+
+
+async def _columns(engine: Any, name: str) -> set[str]:
+    async with engine.connect() as conn:
+        return await conn.run_sync(
+            lambda sync: {c["name"] for c in inspect(sync).get_columns(name)}
+        )
+
+
+async def _pages(
+    engine: Any,
+    table: Any,
+    columns: list[Any],
+    ts: Any,
+    where: Any,
+    since: datetime | None,
+    until: datetime | None,
+    page_size: int,
+) -> AsyncIterator[list[dict[str, Any]]]:
+    """Rows matching ``where``, oldest first, ``page_size`` at a time.
+
+    Keyset paging on (timestamp, id), so rows written during the run neither
+    shift the pages nor get read twice.
+    """
+    conditions = [where]
+    if since is not None:
+        conditions.append(ts >= since)
+    if until is not None:
+        conditions.append(ts < until)
+    last: tuple[Any, Any] | None = None
+    while True:
+        query = select(*columns).where(*conditions)
+        if last is not None:
+            query = query.where(tuple_(ts, table.c.id) > tuple_(*last))
+        query = query.order_by(ts, table.c.id).limit(page_size)
+        async with engine.connect() as conn:
+            rows = [dict(r) for r in (await conn.execute(query)).mappings()]
+        if not rows:
+            return
+        yield rows
+        if len(rows) < page_size:
+            return
+        last = (rows[-1][ts.name], rows[-1]["id"])
+
+
+async def _backfill_calls(run: _Run, since, until, page_size) -> None:
+    db = run.sink.database_sink
+    t = db.table
+    have = await _columns(db.engine, t.name)
+    if "outcome" in have:
+        # Rows from before the outcome column was added have it NULL and
+        # were errors exactly when they carry an error message.
+        where = or_(
+            t.c.outcome.in_(("error", "soft_error")),
+            and_(t.c.outcome.is_(None), t.c.error_message.is_not(None)),
+        )
+        columns = [t.c[c] for c in (*_CALL_COLUMNS, "outcome")]
+    else:
+        where = t.c.error_message.is_not(None)
+        columns = [t.c[c] for c in _CALL_COLUMNS]
+    async for rows in _pages(
+        db.engine, t, columns, t.c.started_at, where, since, until, page_size
+    ):
+        records: list[Record] = []
+        for row in rows:
+            row["outcome"] = row.get("outcome") or "error"
+            records.append(ToolCallRecord(**row))
+        await run.feed(records)
+        if run.done:
+            return
+
+
+async def _backfill_events(run: _Run, since, until, page_size) -> None:
+    db = run.sink.database_sink
+    e = db.events
+    kinds = []
+    if "feedback" in run.sources:
+        kinds.append(e.c.kind == FEEDBACK_KIND)
+    if "llm" in run.sources:
+        kinds.append(e.c.kind == LLM_KIND)
+    if "event" in run.sources:
+        # Includes llm.call: one without prompt or completion is a plain event.
+        kinds.append(e.c.kind != FEEDBACK_KIND)
+    columns = [e.c.id, e.c.kind, e.c.key, e.c.occurred_at, e.c.attrs]
+    async for rows in _pages(
+        db.engine, e, columns, e.c.occurred_at, or_(*kinds), since, until, page_size
+    ):
+        records: list[Record] = [
+            EventRecord(**{**row, "attrs": row["attrs"] or {}}) for row in rows
+        ]
+        await run.feed(records)
+        if run.done:
+            return
+
+
+async def _start(
+    sink: EmbeddingSink,
+    sources: Iterable[str] | None,
+    limit: int | None,
+    batch_size: int,
+    dry_run: bool,
+) -> _Run:
+    chosen = tuple(sink.sources if sources is None else sources)
+    unknown = set(chosen) - set(EMBEDDING_SOURCES)
+    if unknown:
+        raise ValueError(
+            f"unknown sources {sorted(unknown)}; choose from {EMBEDDING_SOURCES}"
+        )
+    if batch_size < 1:
+        raise ValueError(f"batch_size must be at least 1, got {batch_size!r}")
+    if limit is not None and limit < 0:
+        raise ValueError(f"limit must not be negative, got {limit!r}")
+    run = _Run(sink, sorted(chosen, key=EMBEDDING_SOURCES.index), limit, dry_run)
+    db = sink.database_sink
+    if not dry_run:
+        await db.ensure_embeddings()
+    run.have_table = await _has_table(db.engine, db.embeddings.name)
+    if not run.have_table and not dry_run:
+        raise RuntimeError(
+            f"{db.embeddings.name} does not exist; create it with a migration "
+            "or pass create_tables=True to the DatabaseSink"
+        )
+    return run
+
+
+async def backfill_embeddings(
+    embedding_sink: EmbeddingSink,
+    *,
+    sources: Iterable[str] | None = None,
+    since: datetime | None = None,
+    until: datetime | None = None,
+    limit: int | None = None,
+    batch_size: int = 500,
+    dry_run: bool = False,
+) -> dict[str, int]:
+    """Embed stored tool-call errors and events that have no embedding yet.
+
+    Reads the tables of ``embedding_sink.database_sink``:
+
+    - ``call_error``: calls with outcome ``error`` or ``soft_error``, and
+      calls with outcome NULL and an error message (rows written before the
+      outcome column existed).
+    - ``feedback``: ``feedback.submitted`` events.
+    - ``llm``: ``llm.call`` events with a prompt or completion.
+    - ``event``: other events with any of the sink's ``event_text_keys``.
+
+    Each text goes through the sink's extraction, redactor and ``max_chars``,
+    so it matches what the live sink makes of the same record. Texts already
+    stored with the same hash are skipped; a changed text replaces its row.
+
+    Args:
+        sources: which of the above; defaults to the sink's ``sources``.
+        since, until: only rows timestamped in [since, until). Naive
+            datetimes are taken as UTC.
+        limit: embed at most this many texts, oldest first. A later run
+            carries on, since what was embedded is then skipped.
+        batch_size: rows read per query. Texts are embedded in chunks of
+            the sink's ``batch_size``, each limited to its ``timeout``.
+        dry_run: count what would be embedded without calling the embedder
+            or writing anything.
+
+    Returns counts of texts embedded per source type, plus ``skipped``
+    (already embedded) and ``errors`` (texts whose embed or store failed;
+    logged, never raised). Raises ValueError for bad arguments and
+    RuntimeError when a table it needs does not exist.
+    """
+    run = await _start(embedding_sink, sources, limit, batch_size, dry_run)
+    db = embedding_sink.database_sink
+    since, until = _utc(since), _utc(until)
+    if "call_error" in run.sources:
+        if not await _has_table(db.engine, db.table.name):
+            raise RuntimeError(f"{db.table.name} does not exist")
+        if not run.done:
+            await _backfill_calls(run, since, until, batch_size)
+    if set(run.sources) & {"feedback", "event", "llm"}:
+        if not await _has_table(db.engine, db.events.name):
+            raise RuntimeError(f"{db.events.name} does not exist")
+        if not run.done:
+            await _backfill_events(run, since, until, batch_size)
+    return run.result()
+
+
+async def backfill_feedback(
+    embedding_sink: EmbeddingSink,
+    items: Iterable[tuple[str, str, str | None]],
+    *,
+    batch_size: int = 500,
+    dry_run: bool = False,
+) -> dict[str, int]:
+    """Embed feedback kept in the host's own tables, as (ref, title, description).
+
+    The text and source id are those of a ``feedback.submitted`` event with
+    key ``ref`` and those attrs, so feedback recorded later as an event
+    dedupes against these rows. Returns ``{"feedback", "skipped", "errors"}``
+    like ``backfill_embeddings``.
+    """
+    run = await _start(embedding_sink, ("feedback",), None, batch_size, dry_run)
+    now = datetime.now(UTC)
+    page: list[Record] = []
+    for ref, title, description in items:
+        page.append(
+            EventRecord(
+                kind=FEEDBACK_KIND,
+                occurred_at=now,
+                key=str(ref),
+                attrs={"title": title, "description": description},
+            )
+        )
+        if len(page) >= batch_size:
+            await run.feed(page)
+            page = []
+    if page:
+        await run.feed(page)
+    return run.result()
+
+
+__all__ = ["backfill_embeddings", "backfill_feedback"]

src/fastmcp_feedback/instrumentation/backfill/__main__.py

@@ -0,0 +1,5 @@
+import sys
+
+from .cli import main
+
+sys.exit(main())

src/fastmcp_feedback/instrumentation/backfill/cli.py

@@ -0,0 +1,143 @@
+"""python -m fastmcp_feedback.instrumentation.backfill
+
+Prints the counts as JSON. Exits 1 when any text failed to embed, 2 on a
+setup error (bad arguments, missing table, unreachable database).
+"""
+
+import argparse
+import asyncio
+import json
+import logging
+import os
+import sys
+from datetime import datetime
+
+from ..embeddings import DEFAULT_EVENT_TEXT_KEYS, EmbeddingSink, OpenAIEmbedder
+from ..storage import DEFAULT_TABLE_PREFIX, EMBEDDING_SOURCES, DatabaseSink
+from . import backfill_embeddings
+
+API_KEY_ENV = "FFB_EMBED_API_KEY"
+
+
+def _sources(value: str) -> tuple[str, ...]:
+    chosen = tuple(s.strip() for s in value.split(",") if s.strip())
+    unknown = set(chosen) - set(EMBEDDING_SOURCES)
+    if not chosen or unknown:
+        raise argparse.ArgumentTypeError(
+            f"choose from {','.join(EMBEDDING_SOURCES)}, comma separated"
+        )
+    return chosen
+
+
+def _when(value: str) -> datetime:
+    try:
+        return datetime.fromisoformat(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(f"not an ISO 8601 time: {value!r}") from exc
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="python -m fastmcp_feedback.instrumentation.backfill",
+        description=(
+            "Embed stored tool-call errors and events that have no embedding "
+            f"yet. The embedder's API key is read from ${API_KEY_ENV} (or the "
+            "variable named by --api-key-env)."
+        ),
+    )
+    parser.add_argument(
+        "--database-url",
+        required=True,
+        help="async SQLAlchemy URL, e.g. postgresql+asyncpg://... or sqlite+aiosqlite:///calls.db",
+    )
+    parser.add_argument(
+        "--embed-base-url",
+        help="OpenAI-compatible base URL up to the version, e.g. https://api.example.com/v1 "
+        "(required unless --dry-run)",
+    )
+    parser.add_argument("--api-key-env", default=API_KEY_ENV, metavar="NAME")
+    parser.add_argument("--model", default="mxbai-embed-large")
+    parser.add_argument("--dim", type=int, default=1024)
+    parser.add_argument("--prefix", default=DEFAULT_TABLE_PREFIX)
+    parser.add_argument(
+        "--sources",
+        type=_sources,
+        default=EMBEDDING_SOURCES,
+        help=f"comma separated, default {','.join(EMBEDDING_SOURCES)}",
+    )
+    parser.add_argument(
+        "--event-text-keys",
+        default=",".join(DEFAULT_EVENT_TEXT_KEYS),
+        help="attrs making up an event text, comma separated",
+    )
+    parser.add_argument("--since", type=_when, help="ISO 8601; naive means UTC")
+    parser.add_argument("--until", type=_when, help="ISO 8601, exclusive")
+    parser.add_argument("--limit", type=int, help="embed at most this many texts")
+    parser.add_argument("--batch-size", type=int, default=500, help="rows per query")
+    parser.add_argument("--max-chars", type=int, default=4000)
+    parser.add_argument(
+        "--create-tables",
+        action="store_true",
+        help="create the embeddings table (and vector extension) if missing",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="count what would be embedded; call no embedder, write nothing",
+    )
+    parser.add_argument("-v", "--verbose", action="store_true")
+    return parser
+
+
+async def _run(args: argparse.Namespace) -> dict[str, int]:
+    embedder = OpenAIEmbedder(
+        args.embed_base_url or "http://embedder.invalid/v1",
+        api_key=os.environ.get(args.api_key_env) or None,
+        model=args.model,
+        dim=args.dim,
+    )
+    db = DatabaseSink(
+        args.database_url,
+        prefix=args.prefix,
+        create_tables=args.create_tables and not args.dry_run,
+        embedding_dim=args.dim,
+    )
+    try:
+        sink = EmbeddingSink(
+            embedder,
+            db,
+            sources=args.sources,
+            event_text_keys=[k for k in args.event_text_keys.split(",") if k],
+            max_chars=args.max_chars,
+        )
+        return await backfill_embeddings(
+            sink,
+            since=args.since,
+            until=args.until,
+            limit=args.limit,
+            batch_size=args.batch_size,
+            dry_run=args.dry_run,
+        )
+    finally:
+        await db.aclose()
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    if not args.dry_run and not args.embed_base_url:
+        parser.error("--embed-base-url is required unless --dry-run")
+    # force: importing fastmcp_feedback already configured the root logger
+    # at INFO, which would print every embedding request.
+    logging.basicConfig(
+        level=logging.INFO if args.verbose else logging.WARNING,
+        format="%(levelname)s %(name)s: %(message)s",
+        force=True,
+    )
+    try:
+        counts = asyncio.run(_run(args))
+    except Exception as exc:
+        print(f"backfill failed: {type(exc).__name__}: {exc}", file=sys.stderr)
+        return 2
+    print(json.dumps({"dry_run": args.dry_run, **counts}))
+    return 1 if counts["errors"] else 0

src/fastmcp_feedback/instrumentation/embeddings.py

@@ -143,10 +143,44 @@ class EmbeddingText:
         return hashlib.sha256(self.text.encode()).hexdigest()
 
 
-def _clean(value: Any) -> str | None:
+# Flattening stops adding items past this many characters; max_chars cuts
+# the result again later. Keeps a huge list from being joined in full.
+_FLAT_CHARS = 65536
+_FLAT_DEPTH = 32
+
+
+def _clean(value: Any, depth: int = 0) -> str | None:
+    """``value`` as stripped text, or None when there is none.
+
+    Lists and tuples become their items' texts, one per line. Dicts become
+    ``key: value`` lines sorted by key: PostgreSQL JSONB does not keep key
+    order, and the text has to come out the same when a backfill rebuilds it
+    from a stored row. Other values go through ``str()``.
+    """
     if value is None:
         return None
-    text = value if isinstance(value, str) else str(value)
+    if isinstance(value, str):
+        text = value
+    elif isinstance(value, list | tuple | dict):
+        if depth >= _FLAT_DEPTH:
+            return None
+        if isinstance(value, dict):
+            pairs = sorted(value.items(), key=lambda kv: str(kv[0]))
+            parts = ((f"{k}: ", _clean(v, depth + 1)) for k, v in pairs)
+        else:
+            parts = (("", _clean(v, depth + 1)) for v in value)
+        lines: list[str] = []
+        size = 0
+        for label, part in parts:
+            if not part:
+                continue
+            lines.append(label + part)
+            size += len(lines[-1]) + 1
+            if size > _FLAT_CHARS:
+                break
+        text = "\n".join(lines)
+    else:
+        text = str(value)
     text = text.strip()
     return text or None
 
@@ -168,6 +202,9 @@ def extract_texts(
     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.
+
+    List and tuple values become one line per item, dicts ``key: value``
+    lines sorted by key, at any depth.
     """
     wanted = frozenset(sources)
     out: list[EmbeddingText] = []
@@ -379,18 +416,35 @@ class EmbeddingSink:
     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()
+        todo, stored = await self._unembedded(list(latest.values()))
+        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
+        )
+        await self._store(todo, vectors, stored)
+        self.embedded += len(todo)
+
+    async def _unembedded(
+        self, items: list[EmbeddingText]
+    ) -> tuple[list[EmbeddingText], set[tuple[str, str]]]:
+        """The items whose stored row is missing or holds another text, and
+        the keys that have a row (to be replaced). Items must be unique by
+        (source_type, source_id)."""
+        if not items:
+            return [], set()
         t = self.table
-        model = self.embedder.model
-        async with db.engine.connect() as conn:
+        keys = [(i.source_type, i.source_id) for i in items]
+        async with self.database_sink.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)),
+                        t.c.model == self.embedder.model,
+                        tuple_(t.c.source_type, t.c.source_id).in_(keys),
                     )
                 )
             ).all()
@@ -398,12 +452,15 @@ class EmbeddingSink:
         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
-        )
+        return todo, set(stored)
+
+    async def _store(
+        self,
+        todo: list[EmbeddingText],
+        vectors: list[list[float]],
+        replace: set[tuple[str, str]],
+    ) -> None:
+        db = self.database_sink
         if len(vectors) != len(todo):
             raise ValueError(
                 f"embedder returned {len(vectors)} vectors for {len(todo)}"
@@ -420,7 +477,7 @@ class EmbeddingSink:
                 "id": str(uuid.uuid4()),
                 "source_type": i.source_type,
                 "source_id": i.source_id,
-                "model": model,
+                "model": self.embedder.model,
                 "dim": len(v),
                 "text_hash": i.text_hash,
                 "text": i.text,
@@ -429,8 +486,7 @@ class EmbeddingSink:
             }
             for i, v in zip(todo, vectors, strict=True)
         ]
-        await self._upsert(values, replace=set(stored))
-        self.embedded += len(values)
+        await self._upsert(values, replace=replace)
 
     async def _upsert(self, values: list[dict[str, Any]], replace: set) -> None:
         t = self.table

tests/unit/test_backfill.py

@@ -0,0 +1,565 @@
+"""Tests for backfill_embeddings, backfill_feedback, its CLI, and list/dict
+text flattening.
+
+DB tests take the ``db_url`` fixture, so they run on SQLite and, with
+FFB_TEST_DATABASE_URL set, on Postgres with pgvector.
+"""
+
+import json
+import os
+import subprocess
+import sys
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy import func, select, update
+
+from fastmcp_feedback.instrumentation import (
+    REDACTED,
+    DatabaseSink,
+    EmbeddingSink,
+    EventRecord,
+    Redactor,
+    backfill_embeddings,
+    backfill_feedback,
+    extract_texts,
+)
+from fastmcp_feedback.instrumentation.backfill.cli import main as cli_main
+
+from .test_embeddings import DIM, SECRET, FakeEmbedder, call_record, event, rows
+
+T0 = datetime(2026, 9, 1, 12, 0, tzinfo=UTC)
+
+
+@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 at(minutes: float) -> datetime:
+    return T0 + timedelta(minutes=minutes)
+
+
+def error_call(minute, tool="render", message="render timed out", **kw):
+    record = call_record(tool=tool, message=message, **kw)
+    record.started_at = at(minute)
+    return record
+
+
+def ev(minute, kind, key=None, **attrs):
+    record = event(kind, key, **attrs)
+    record.occurred_at = at(minute)
+    return record
+
+
+def mixed_records():
+    """One of everything backfill should pick up, and some it should not."""
+    return [
+        error_call(1, message="render timed out after 30s"),
+        error_call(2, tool="find", outcome="soft_error", message="no object Cube"),
+        call_record(outcome="ok"),
+        ev(
+            3, "feedback.submitted", "bug-1", title="Render hangs", description="at 99%"
+        ),
+        ev(
+            4, "llm.call", model="m", prompt="summarize the log", completion="it failed"
+        ),
+        ev(5, "llm.call", model="m", ok=False, error="rate limited"),
+        ev(6, "llm.call", model="m", ok=True),
+        ev(7, "job.failed", "job-1", error_excerpt=["line one", "line two"]),
+        ev(8, "job.progress", "job-1", percent=50),
+    ]
+
+
+def keyed(stored):
+    return {(r["source_type"], r["source_id"]): r for r in stored}
+
+
+async def count_rows(sink):
+    t = sink.embeddings
+    async with sink.engine.connect() as conn:
+        return (await conn.execute(select(func.count()).select_from(t))).scalar_one()
+
+
+class TestFlattening:
+    def test_list_and_tuple_join_lines(self):
+        records = [
+            event("job.failed", error_excerpt=["  a  ", "", None, "b"]),
+            event("job.failed", error=("x", 1, 2.5)),
+        ]
+        texts = [t.text for t in extract_texts(records)]
+        assert texts == ["a\nb", "x\n1\n2.5"]
+
+    def test_dict_becomes_sorted_key_lines(self):
+        records = [event("job.failed", error={"zeta": "last", "alpha": 1, "mid": None})]
+        [item] = extract_texts(records)
+        assert item.text == "alpha: 1\nzeta: last"
+
+    def test_nested(self):
+        records = [
+            event(
+                "job.failed",
+                error_excerpt=[["a", "b"], {"k": ["c", "d"]}, ("e",)],
+            )
+        ]
+        [item] = extract_texts(records)
+        assert item.text == "a\nb\nk: c\nd\ne"
+        assert "[" not in item.text and "{" not in item.text
+
+    def test_flattening_applies_to_feedback_and_llm(self):
+        records = [
+            event("feedback.submitted", "f1", title=["Two", "parts"], description=None),
+            event("llm.call", prompt=["p1", "p2"], completion={"text": "done"}),
+        ]
+        by_type = {t.source_type: t.text for t in extract_texts(records)}
+        assert by_type["feedback"] == "Two\nparts"
+        assert by_type["llm"] == "prompt: p1\np2\ncompletion: text: done"
+
+    def test_empty_containers_are_no_text(self):
+        records = [event("job.failed", error_excerpt=[], error={}, message=[[""]])]
+        assert extract_texts(records) == []
+
+    def test_bounded(self):
+        deep: list = ["bottom"]
+        for _ in range(100):
+            deep = [deep]
+        huge = ["x" * 1000] * 10_000
+        records = [event("job.failed", error=deep), event("job.failed", message=huge)]
+        texts = extract_texts(records)
+        assert len(texts) == 1  # the deep one is cut off before any text
+        assert len(texts[0].text) < 70_000
+
+    def test_other_values_use_str(self):
+        [item] = extract_texts([event("job.failed", error=42)])
+        assert item.text == "42"
+
+
+class TestBackfill:
+    async def test_everything_once(self, db):
+        store = db()
+        await store.write(mixed_records())
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        counts = await backfill_embeddings(emb)
+        assert counts == {
+            "feedback": 1,
+            "call_error": 2,
+            "event": 2,
+            "llm": 1,
+            "skipped": 0,
+            "errors": 0,
+        }
+        stored = keyed(await rows(store))
+        texts = {k[0]: [] for k in stored}
+        for (source_type, _), row in stored.items():
+            texts[source_type].append(row["text"])
+        assert sorted(texts["call_error"]) == [
+            "find: TimeoutError: no object Cube",
+            "render: TimeoutError: render timed out after 30s",
+        ]
+        assert texts["feedback"] == ["Render hangs\nat 99%"]
+        assert texts["llm"] == ["prompt: summarize the log\ncompletion: it failed"]
+        assert sorted(texts["event"]) == ["line one\nline two", "rate limited"]
+        assert ("feedback", "bug-1") in stored
+
+    async def test_pre_outcome_error_rows(self, db):
+        store = db()
+        old = error_call(1, message="boom from before outcome existed")
+        old_ok = call_record(outcome="ok")
+        await store.write([old, old_ok])
+        t = store.table
+        async with store.engine.begin() as conn:
+            await conn.execute(update(t).values(outcome=None))
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        counts = await backfill_embeddings(emb)
+        assert counts["call_error"] == 1
+        [row] = await rows(store)
+        assert row["source_id"] == old.id
+        assert row["text"] == "render: TimeoutError: boom from before outcome existed"
+
+    async def test_same_text_hash_and_id_as_live_sink(self, db):
+        records = mixed_records() + [
+            ev(9, "job.failed", error={"zeta": 1, "alpha": {"b": 2, "a": [1, 2]}}),
+        ]
+        backfilled, live = db("backfilled"), db("live")
+        await backfilled.write(records)
+        await backfill_embeddings(EmbeddingSink(FakeEmbedder(), backfilled))
+        live_emb = EmbeddingSink(FakeEmbedder(), live)
+        await live_emb.write(records)
+        await live_emb.flush()
+        assert live_emb.errors == 0
+
+        def shape(stored):
+            return sorted(
+                (r["source_type"], r["source_id"], r["text"], r["text_hash"])
+                for r in stored
+            )
+
+        assert shape(await rows(backfilled)) == shape(await rows(live))
+        assert len(await rows(live)) == 7
+
+    async def test_live_records_dedupe_against_backfill(self, db):
+        records = mixed_records() + [
+            ev(9, "job.failed", error={"zeta": 1, "alpha": {"b": 2, "a": [1, 2]}}),
+        ]
+        store = db()
+        await store.write(records)
+        await backfill_embeddings(EmbeddingSink(FakeEmbedder(), store))
+        before = await count_rows(store)
+        embedder = FakeEmbedder()
+        live = EmbeddingSink(embedder, store)
+        # The records as they come back from the database, as a live sink
+        # in another process would have seen them before storing.
+        await live.write(records)
+        await live.flush()
+        assert embedder.calls == []
+        assert live.embedded == 0 and live.skipped == 7
+        assert await count_rows(store) == before
+
+    async def test_second_run_embeds_nothing(self, db):
+        store = db()
+        await store.write(mixed_records())
+        embedder = FakeEmbedder()
+        emb = EmbeddingSink(embedder, store)
+        await backfill_embeddings(emb)
+        calls = len(embedder.calls)
+        again = await backfill_embeddings(emb)
+        assert again == {
+            "feedback": 0,
+            "call_error": 0,
+            "event": 0,
+            "llm": 0,
+            "skipped": 6,
+            "errors": 0,
+        }
+        assert len(embedder.calls) == calls
+
+    async def test_changed_text_replaces(self, db):
+        store = db()
+        record = error_call(1, message="first")
+        await store.write([record])
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        await backfill_embeddings(emb)
+        async with store.engine.begin() as conn:
+            await conn.execute(
+                update(store.table).values(error_message="second"),
+            )
+        counts = await backfill_embeddings(emb)
+        assert counts["call_error"] == 1
+        [row] = await rows(store)
+        assert row["text"] == "render: TimeoutError: second"
+
+    async def test_sources(self, db):
+        store = db()
+        await store.write(mixed_records())
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        counts = await backfill_embeddings(emb, sources=["feedback", "llm"])
+        assert counts == {"feedback": 1, "llm": 1, "skipped": 0, "errors": 0}
+        assert {r["source_type"] for r in await rows(store)} == {"feedback", "llm"}
+
+    async def test_defaults_to_sink_sources(self, db):
+        store = db()
+        await store.write(mixed_records())
+        emb = EmbeddingSink(FakeEmbedder(), store, sources=["call_error"])
+        counts = await backfill_embeddings(emb)
+        assert counts == {"call_error": 2, "skipped": 0, "errors": 0}
+
+    async def test_since_until(self, db):
+        store = db()
+        await store.write([error_call(m, message=f"error {m}") for m in range(10)])
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        counts = await backfill_embeddings(emb, since=at(3), until=at(6))
+        assert counts["call_error"] == 3
+        texts = sorted(r["text"] for r in await rows(store))
+        assert texts == [f"render: TimeoutError: error {m}" for m in (3, 4, 5)]
+        # A naive datetime is UTC.
+        naive = (T0 + timedelta(minutes=8)).replace(tzinfo=None)
+        counts = await backfill_embeddings(emb, since=naive)
+        assert counts["call_error"] == 2
+
+    async def test_limit_takes_oldest_and_resumes(self, db):
+        store = db()
+        await store.write([error_call(m, message=f"error {m}") for m in range(10)])
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        counts = await backfill_embeddings(emb, limit=4, batch_size=3)
+        assert counts["call_error"] == 4
+        texts = sorted(r["text"] for r in await rows(store))
+        assert texts == [f"render: TimeoutError: error {m}" for m in range(4)]
+        counts = await backfill_embeddings(emb, limit=4, batch_size=3)
+        assert counts["call_error"] == 4 and counts["skipped"] == 4
+        counts = await backfill_embeddings(emb, limit=0)
+        assert counts["call_error"] == 0
+
+    async def test_pages_through_many_rows(self, db):
+        store = db()
+        # Identical timestamps too, so paging must break ties on id.
+        records = [error_call(m // 5, message=f"error {m}") for m in range(53)]
+        records += [ev(m // 5, "job.failed", error=f"job error {m}") for m in range(41)]
+        await store.write(records)
+        embedder = FakeEmbedder()
+        emb = EmbeddingSink(embedder, store, batch_size=8)
+        counts = await backfill_embeddings(emb, batch_size=7)
+        assert counts["call_error"] == 53 and counts["event"] == 41
+        assert await count_rows(store) == 94
+        assert len(embedder.seen) == 94  # nothing read twice
+        assert max(len(c) for c in embedder.calls) <= 8
+
+    async def test_embedder_errors_are_counted(self, db):
+        store = db()
+        await store.write([error_call(m, message=f"error {m}") for m in range(5)])
+        emb = EmbeddingSink(FakeEmbedder(fail=True), store, batch_size=2)
+        counts = await backfill_embeddings(emb)
+        assert counts["errors"] == 5 and counts["call_error"] == 0
+        assert await count_rows(store) == 0
+        emb.embedder.fail = False
+        counts = await backfill_embeddings(emb)
+        assert counts["call_error"] == 5 and counts["errors"] == 0
+
+    async def test_wrong_vector_count_is_counted(self, db):
+        class Short(FakeEmbedder):
+            async def embed(self, texts):
+                return (await super().embed(texts))[:-1]
+
+        store = db()
+        await store.write([error_call(1), error_call(2, message="other")])
+        counts = await backfill_embeddings(EmbeddingSink(Short(), store))
+        assert counts["errors"] == 2
+
+    async def test_redaction(self, db):
+        store = db()
+        leaked = error_call(1, message=f"auth failed with {SECRET}")
+        custom = ev(2, "job.failed", error="ticket ACME-1234 failed")
+        await store.write([leaked, custom])
+        embedder = FakeEmbedder()
+        emb = EmbeddingSink(
+            embedder, store, redactor=Redactor(extra_patterns=[r"ACME-\d+"])
+        )
+        await backfill_embeddings(emb)
+        texts = sorted(r["text"] for r in await rows(store))
+        assert SECRET not in "".join(embedder.seen + texts)
+        assert "ACME-1234" not in "".join(embedder.seen + texts)
+        assert f"render: TimeoutError: auth failed with {REDACTED}" in texts
+        assert f"ticket {REDACTED} failed" in texts
+
+    async def test_truncation(self, db):
+        store = db()
+        await store.write([error_call(1, message="x" * 500)])
+        emb = EmbeddingSink(FakeEmbedder(), store, max_chars=50)
+        await backfill_embeddings(emb)
+        [row] = await rows(store)
+        assert len(row["text"]) == 50
+
+    async def test_dry_run_writes_nothing(self, db):
+        store = db()
+        await store.write(mixed_records())
+        embedder = FakeEmbedder()
+        emb = EmbeddingSink(embedder, store)
+        counts = await backfill_embeddings(emb, dry_run=True)
+        assert counts["call_error"] == 2 and counts["feedback"] == 1
+        assert embedder.calls == []
+        assert await rows(store) == []
+
+    async def test_missing_tables_raise(self, db):
+        store = db(create_tables=False)
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        with pytest.raises(RuntimeError, match="does not exist"):
+            await backfill_embeddings(emb)
+        with pytest.raises(RuntimeError, match="does not exist"):
+            await backfill_embeddings(emb, dry_run=True)
+
+    async def test_bad_arguments(self, db):
+        emb = EmbeddingSink(FakeEmbedder(), db())
+        with pytest.raises(ValueError, match="unknown sources"):
+            await backfill_embeddings(emb, sources=["nope"])
+        with pytest.raises(ValueError):
+            await backfill_embeddings(emb, batch_size=0)
+        with pytest.raises(ValueError):
+            await backfill_embeddings(emb, limit=-1)
+
+
+class TestBackfillFeedback:
+    async def test_matches_feedback_event(self, db):
+        own, events = db("own"), db("events")
+        items = [
+            ("bug-1", " Render hangs ", "Stuck at 99%"),
+            ("bug-2", "No description", None),
+            ("bug-3", "", ""),
+        ]
+        counts = await backfill_feedback(EmbeddingSink(FakeEmbedder(), own), items)
+        assert counts == {"feedback": 2, "skipped": 0, "errors": 0}
+        live = EmbeddingSink(FakeEmbedder(), events)
+        await live.write(
+            [
+                EventRecord(
+                    kind="feedback.submitted",
+                    occurred_at=datetime.now(UTC),
+                    key=ref,
+                    attrs={"title": title, "description": desc},
+                )
+                for ref, title, desc in items
+            ]
+        )
+        await live.flush()
+
+        def shape(stored):
+            return sorted((r["source_id"], r["text"], r["text_hash"]) for r in stored)
+
+        assert shape(await rows(own)) == shape(await rows(events))
+        assert shape(await rows(own))[0][1] == "Render hangs\nStuck at 99%"
+
+    async def test_idempotent_and_dedupes_with_events(self, db):
+        store = db()
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        await backfill_feedback(emb, [("bug-1", "Render hangs", "at 99%")])
+        await store.write(
+            [
+                ev(
+                    1,
+                    "feedback.submitted",
+                    "bug-1",
+                    title="Render hangs",
+                    description="at 99%",
+                )
+            ]
+        )
+        counts = await backfill_embeddings(emb)
+        assert counts["feedback"] == 0 and counts["skipped"] == 1
+        counts = await backfill_feedback(
+            emb, iter([("bug-1", "Render hangs", "at 99%")]), batch_size=1
+        )
+        assert counts == {"feedback": 0, "skipped": 1, "errors": 0}
+
+    async def test_redacted(self, db):
+        store = db()
+        emb = EmbeddingSink(FakeEmbedder(), store)
+        await backfill_feedback(emb, [("bug-1", "Leak", f"my key is {SECRET}")])
+        [row] = await rows(store)
+        assert SECRET not in row["text"] and REDACTED in row["text"]
+
+
+class TestCli:
+    async def seeded(self, db):
+        store = db()
+        await store.write(mixed_records())
+        await store.ensure_embeddings()
+        return store
+
+    async def test_dry_run_main(self, db, capsys):
+        store = await self.seeded(db)
+        url = store.engine.url.render_as_string(hide_password=False)
+        code = await _in_thread(
+            cli_main,
+            [
+                "--database-url",
+                url,
+                "--dim",
+                str(DIM),
+                "--dry-run",
+                "--sources",
+                "call_error,feedback",
+            ],
+        )
+        assert code == 0
+        out = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
+        assert out == {
+            "dry_run": True,
+            "feedback": 1,
+            "call_error": 2,
+            "skipped": 0,
+            "errors": 0,
+        }
+        assert await rows(store) == []
+
+    async def test_dry_run_subprocess(self, db):
+        store = await self.seeded(db)
+        url = store.engine.url.render_as_string(hide_password=False)
+        env = {k: v for k, v in os.environ.items() if k != "FFB_EMBED_API_KEY"}
+        proc = await _in_thread(
+            subprocess.run,
+            [
+                sys.executable,
+                "-m",
+                "fastmcp_feedback.instrumentation.backfill",
+                "--database-url",
+                url,
+                "--dim",
+                str(DIM),
+                "--dry-run",
+            ],
+            capture_output=True,
+            text=True,
+            env=env,
+            timeout=60,
+        )
+        assert proc.returncode == 0, proc.stderr
+        out = json.loads(proc.stdout.strip().splitlines()[-1])
+        assert out["call_error"] == 2 and out["llm"] == 1 and out["event"] == 2
+        assert "RuntimeWarning" not in proc.stderr
+
+    async def test_needs_embed_url_without_dry_run(self, capsys):
+        with pytest.raises(SystemExit) as exc:
+            cli_main(["--database-url", "sqlite+aiosqlite:///x.db"])
+        assert exc.value.code == 2
+        assert "--embed-base-url" in capsys.readouterr().err
+
+    async def test_no_api_key_flag(self, capsys):
+        with pytest.raises(SystemExit):
+            cli_main(["--database-url", "sqlite+aiosqlite:///x.db", "--api-key", "k"])
+
+    async def test_setup_error_exit_code(self, tmp_path, capsys):
+        url = f"sqlite+aiosqlite:///{tmp_path / 'empty.db'}"
+        code = await _in_thread(cli_main, ["--database-url", url, "--dry-run"])
+        assert code == 2
+        assert "does not exist" in capsys.readouterr().err
+
+    async def test_api_key_from_env(self, db, monkeypatch):
+        seen = {}
+
+        class Recorder(FakeEmbedder):
+            def __init__(self, base_url, api_key=None, model="m", dim=DIM, **kw):
+                super().__init__(dim=dim, model=model)
+                seen["api_key"] = api_key
+                seen["base_url"] = base_url
+
+        store = await self.seeded(db)
+        url = store.engine.url.render_as_string(hide_password=False)
+        monkeypatch.setattr(
+            "fastmcp_feedback.instrumentation.backfill.cli.OpenAIEmbedder", Recorder
+        )
+        monkeypatch.setenv("MY_KEY", "secret-value")
+        code = await _in_thread(
+            cli_main,
+            [
+                "--database-url",
+                url,
+                "--dim",
+                str(DIM),
+                "--embed-base-url",
+                "https://embed.example/v1",
+                "--api-key-env",
+                "MY_KEY",
+            ],
+        )
+        assert code == 0
+        assert seen == {
+            "api_key": "secret-value",
+            "base_url": "https://embed.example/v1",
+        }
+        assert await count_rows(store) == 6
+
+
+async def _in_thread(fn, *args, **kwargs):
+    # The CLI runs its own event loop, which cannot start on the test's.
+    import asyncio
+
+    return await asyncio.to_thread(fn, *args, **kwargs)

uv.lock

@@ -601,7 +601,7 @@ wheels = [
 
 [[package]]
 name = "fastmcp-feedback"
-version = "2026.9.30"
+version = "2026.9.30.1"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },