Merge fix/explicit-success-wins: an explicit success flag beats error keys

A result with ok or success exactly True skips the error-key check, so
{"ok": True, "error": "deprecated arg"} is ok and the string reads as a
warning. A failing status, a False flag and isError still flag it, and
truthy non-bools are not an explicit success.

.env.example

@@ -6,8 +6,8 @@ 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
+FFB_PG_PORT=55447
 
 # 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
+FFB_TEST_DATABASE_URL=postgresql+asyncpg://ffb:change-me@127.0.0.1:55447/ffb_test

CHANGELOG.md

@@ -5,6 +5,100 @@ 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.10.01.1] - 2026-10-01
+
+A result that says it succeeded is no longer flagged for its own error string.
+
+### Changed
+- Explicit success flag wins over error keys: when any of the classifier's
+  `flag_keys` (`ok`, `success`) is exactly `True`, the `error_keys` check is
+  skipped and the string is treated as a warning, so
+  `{"ok": True, "error": "deprecated arg"}` is recorded as `ok` again. Only
+  that check is skipped: a failing status (`{"success": True, "status":
+  "failed"}`), any flag key exactly `False` and `isError` still flag the
+  result. Truthy values that are not `True`, such as `1` or `"yes"`, are not
+  an explicit success. Results with no flag key are classified as in
+  2026.10.01.
+
+## [2026.10.01] - 2026-10-01
+
+Results that report failure only as an `error` string now count as soft
+errors, and the docs cover joining events to calls.
+
+### Changed
+- `soft_error_classifier(..., error_keys=("error",))`: a dict result whose
+  value at any of `error_keys` is a non-empty string (after stripping) is a
+  soft error, with message `<key>: <string>`, redacted and truncated as
+  before. Checked after the status, flag and `isError` checks, so results
+  those already flag keep their message and are counted once. None, empty or
+  blank strings, numbers, booleans, dicts and lists do not count, so
+  `{"error": None}` and `{"error": 0.02}` stay ok. `error_keys=()` turns it
+  off. `DEFAULT_RESULT_CLASSIFIER` has it on.
+- **Behavior change:** results such as `{"played": False, "error": "..."}`,
+  `{"error": "..."}` and `{"status": "unknown_target", "error": "..."}` were
+  recorded as `ok` and are now `soft_error`, so `NOT ok` and soft-error
+  counts can rise after upgrading. This includes the package's own
+  `list_feedback`, `get_feedback_statistics` and analytics tools, whose
+  failure results carry an `error` string and no flag key. Pass
+  `result_classifier=soft_error_classifier(error_keys=())` for the old
+  behavior.
+
+### Docs
+- `record_event` / `events_for`: a `key` used for joining must be unique for
+  at least the retention period and across restarts; a per-boot counter
+  collides. Use a UUID or a startup token plus a counter.
+- The join from a call to its events through an enricher-written `extra`
+  field, for SQLite (`json_extract(c.extra, '$.speech_id') = e.key`) and
+  Postgres (`c.extra->>'speech_id' = e.key`).
+- Schema notes: `ffb_tool_calls` has primary key `id` and no `call_id`
+  column; `ffb_events.call_id` holds a call's `id`. Ids are UUIDs, so order
+  events by `occurred_at`.
+- The soft-error section documents `error_keys`, its default and the
+  upgrade note.
+
+## [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

CONTRIBUTING.md

@@ -113,7 +113,7 @@ 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
+(55447 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`

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

@@ -224,6 +224,49 @@ The default classifier flags a dict result when `ok` or `success` is exactly
 `False`, when `status` or `state` is `error`, `failed` or `failure` (any case),
 or when the tool marked the result as an MCP error (`isError`) without raising.
 The message comes from the first of `error`, `message`, `detail` and `reason`.
+
+Failing those, a result whose `error` holds a non-empty string is a soft
+error too, with that string as the message: `{"played": False, "error":
+"audio output stalled"}` is recorded as `error: audio output stalled`. Only a
+string with something other than whitespace counts, so `{"error": None}`,
+`{"error": ""}` and a statistic such as `{"error": 0.02}` stay ok. A dict or
+list there (`{"error": {"code": 1}}`) is not flagged either; catch that shape
+with a flag key, a status or your own classifier. `error_keys` names the keys
+to check, and `error_keys=()` turns the check off.
+
+A result that says it succeeded wins over its own error string: when `ok` or
+`success` (the `flag_keys`) is exactly `True`, the `error` check is skipped and
+the string is read as a warning, so `{"ok": True, "error": "deprecated arg"}`
+is recorded as `ok`. Only that check is skipped. In order of precedence:
+
+1. A failing `status` or `state` flags the result, even next to `ok: True`,
+   since `{"success": True, "status": "failed"}` contradicts itself and the
+   status is the more specific of the two.
+2. Any flag key exactly `False` flags it, even if another is `True`.
+3. `isError` flags it.
+4. A flag key exactly `True` makes it ok. Truthy values that are not `True`,
+   such as `1` or `"yes"`, do not count.
+5. A non-empty string at an `error_keys` key flags it.
+
+```python
+from fastmcp_feedback.instrumentation import DEFAULT_RESULT_CLASSIFIER as classify
+
+assert classify("t", {"ok": True, "error": "deprecated arg"}) is None
+assert classify("t", {"ok": 1, "error": "x"}) == "error: x"
+assert classify("t", {"success": True, "status": "failed"}) == "status=failed"
+assert classify("t", {"played": False, "error": "stalled"}) == "error: stalled"
+```
+
+```python
+from fastmcp_feedback.instrumentation import instrument, soft_error_classifier
+
+# Tools that report {"failure_reason": "..."} instead
+instrument(app, result_classifier=soft_error_classifier(error_keys=("failure_reason",)))
+
+# As before 2026.10.01: an error string alone is not a soft error
+instrument(app, result_classifier=soft_error_classifier(error_keys=()))
+```
+
 Add your own status values, or pass any `(tool, result) -> str | None`:
 
 ```python
@@ -260,6 +303,19 @@ table made earlier by `create_tables=True` needs
 tables are not altered. Until the column exists, database writes fail and
 those records are dropped (logged).
 
+**Upgrading to 2026.10.01:** the `error` key check is on by default, so a
+result that carries a non-empty `error` string with no failing status or flag
+(such as `{"status": "unknown_target", "error": "no object named Cube"}`, or
+`{"error": "..."}` alone) is now a soft error where earlier releases recorded
+it as `ok`. Counts of `NOT ok` and `outcome = 'soft_error'` can rise after
+upgrading. Results that also have a failing status or flag keep their old
+message, such as `success=False: ...`, and are counted once. Pass
+`soft_error_classifier(error_keys=())` to keep the old behavior.
+
+**Upgrading to 2026.10.01.1:** a result with `ok` or `success` exactly `True`
+and an `error` string, which 2026.10.01 recorded as `soft_error`, is `ok` again.
+Nothing else changes.
+
 ### Identity and enrichment hooks
 
 ```python
@@ -443,7 +499,12 @@ background queue, the same drop-and-count policy when the queue is full.
 - `kind` is a dotted name (`job.submitted`, `llm.call`), a non-empty string of
   at most 128 characters.
 - `key` is what you join on, such as a job id. It is stored as a string, at
-  most 255 characters.
+  most 255 characters. A key used to join events to calls or to each other
+  must be unique for at least the retention period, **including across
+  restarts**. A per-boot counter (`speech-1`, `speech-2`, ...) starts over
+  on every restart, and the second boot's `speech-1` joins to the first
+  boot's events. Use a UUID, or a token chosen at startup plus a counter
+  (`speech-<boot token>-<n>`).
 - `attrs` is any JSON-like dict. It is redacted like call arguments, then made
   JSON-safe: `NaN` and infinities become the strings `"nan"`/`"inf"` (Postgres
   JSONB rejects them), datetimes become ISO strings, anything else unknown
@@ -468,6 +529,23 @@ for ev in await mw.events_for("job-wing"):
 steps = await mw.events_for("job-wing", kinds=["step.started", "step.finished"], limit=100)
 ```
 
+To join a call to its events in SQL, have the enricher write the key into
+`extra` and match on it. This is how mcspeak reads the phases of one `speak`
+call, with SQLite:
+
+```sql
+SELECT e.occurred_at, e.attrs FROM ffb_events e
+JOIN ffb_tool_calls c ON json_extract(c.extra, '$.speech_id') = e.key
+WHERE e.kind = 'speech.phase' AND c.id = ?
+ORDER BY e.occurred_at;
+```
+
+On Postgres the join condition is `c.extra->>'speech_id' = e.key`. Events
+recorded inside the call also carry its id, so `e.call_id = c.id` joins those
+without a key. `ffb_tool_calls` has no `call_id` column; its primary key is
+`id`, and `ffb_events.call_id` holds that value. Order events by
+`occurred_at`: ids are random UUIDs and do not sort in recording order.
+
 `events_for` reads the `DatabaseSink` and adds events still waiting in the
 queue, so an event is visible as soon as `record_event` returns. With more
 than `limit` (1000) matches it returns the newest `limit`. Without a
@@ -533,8 +611,10 @@ With the default `capture_llm_text=False` both are discarded, and so are
 Events go to a `ffb_events` table: `id`, `occurred_at`, `kind`, `key`,
 `call_id`, `session_id`, `user_sub`, `caller_kind`, `client_id`,
 `server_version`, `attrs` (JSON; JSONB on Postgres), indexed on
-`occurred_at`, `kind`, `key`, `call_id` and `user_sub`. It has no foreign key
-to `ffb_tool_calls`, since calls are pruned and may be written later.
+`occurred_at`, `kind`, `key`, `call_id` and `user_sub`. `id` is a UUID
+string in both tables, so sort by `occurred_at` (events) or `started_at`
+(calls), never by `id`. `call_id` refers to `ffb_tool_calls.id` but has no
+foreign key, since calls are pruned and may be written later.
 
 Sinks get calls and events in the same batches. `DatabaseSink` writes each to
 its table, calls first in their own transaction, so a database without
@@ -570,7 +650,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 +789,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()`

docker-compose.yml

@@ -11,7 +11,7 @@ services:
       POSTGRES_PASSWORD: ${FFB_PG_PASSWORD}
       POSTGRES_DB: ${FFB_PG_DB}
     ports:
-      - "127.0.0.1:${FFB_PG_PORT:-55433}:5432"
+      - "127.0.0.1:${FFB_PG_PORT:-55447}:5432"
     volumes:
       - ffb-postgres-data:/var/lib/postgresql/data
       - ./docker/postgres-init:/docker-entrypoint-initdb.d:ro

docker/pgvector-alpine/Dockerfile

@@ -0,0 +1,13 @@
+# Postgres on Alpine with pgvector built in. Keeps the musl C library, so an
+# existing data directory's collation and text indexes are unaffected.
+# with_llvm=no: the base image's Postgres was built with a specific clang for
+# JIT bitcode that Alpine may no longer ship; pgvector itself builds with gcc.
+ARG PG_MAJOR=16
+FROM postgres:${PG_MAJOR}-alpine
+ARG PGVECTOR_VERSION=v0.8.6
+RUN apk add --no-cache --virtual .build-deps git build-base \
+ && git clone --depth 1 --branch ${PGVECTOR_VERSION} https://github.com/pgvector/pgvector.git /tmp/pgvector \
+ && make -C /tmp/pgvector OPTFLAGS="" with_llvm=no \
+ && make -C /tmp/pgvector install with_llvm=no \
+ && rm -rf /tmp/pgvector \
+ && apk del .build-deps

docker/pgvector-alpine/README.md

@@ -0,0 +1,28 @@
+# Postgres on Alpine with pgvector
+
+For stacks that already run `postgres:NN-alpine` and want the `vector`
+extension for fastmcp-feedback embeddings.
+
+Why not switch to the `pgvector/pgvector` image: it's Debian (glibc), and the
+Alpine images use musl. The two C libraries sort text differently under
+locales such as `en_US.utf8`, so reusing an Alpine data directory under a
+Debian image changes the order of every text index. The indexes need a full
+`REINDEX` or they silently return wrong results. Building pgvector into the
+same Alpine image keeps the data directory's collation unchanged.
+
+```bash
+docker build --build-arg PG_MAJOR=16 -t postgres-pgvector:16-alpine .
+docker build --build-arg PG_MAJOR=17 -t postgres-pgvector:17-alpine .
+```
+
+Then point the stack's Postgres service at the new image (same major version
+as the existing data directory) and run `CREATE EXTENSION IF NOT EXISTS
+vector;` once, or let a migration do it.
+
+Tested 2026-09-30 with pgvector v0.8.6 on postgres:16-alpine and
+postgres:17-alpine: the extension loads, and an HNSW `vector_cosine_ops`
+index is used for `ORDER BY embedding <=> ...` queries.
+
+`with_llvm=no` skips the LLVM bitcode used for JIT inlining. The base images'
+Postgres was built with a clang version Alpine no longer ships, so the bitcode
+step fails; pgvector itself still builds with gcc and works normally.

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.10.01.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.10.01.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/classify.py

@@ -58,6 +58,7 @@ def soft_error_classifier(
     status_keys: Iterable[str] = ("status", "state"),
     flag_keys: Iterable[str] = ("ok", "success"),
     extra_statuses: Iterable[str] = (),
+    error_keys: Iterable[str] = ("error",),
 ) -> ResultClassifier:
     """Build a classifier for dict results that describe their own failure.
 
@@ -67,13 +68,31 @@ def soft_error_classifier(
     ``isError``. The message is the first of ``error``, ``message``,
     ``detail`` and ``reason`` present, e.g. ``status=failed: disk full``.
 
+    Failing those, it is a soft error when any of ``error_keys`` holds a
+    non-empty string, as in ``{"played": False, "error": "output stalled"}``,
+    with that string as the message (``error: output stalled``). None, empty
+    or blank strings, numbers, booleans, dicts and lists there do not count,
+    so ``{"error": None}`` and ``{"error": 0.02}`` stay ok. ``error_keys=()``
+    turns this off.
+
+    A result that says it succeeded, with any of ``flag_keys`` exactly True,
+    skips only that last check: ``{"ok": True, "error": "deprecated arg"}``
+    is ok, the string being a warning. A failing status, a False flag and
+    ``isError`` still flag it. Truthy non-bools such as ``1`` or ``"yes"``
+    are not an explicit success.
+
     >>> classify = soft_error_classifier(extra_statuses=("unknown_target",))
     >>> classify("render", {"status": "unknown_target", "error": "no such object"})
     'status=unknown_target: no such object'
+    >>> classify("speak", {"played": False, "error": "output stalled"})
+    'error: output stalled'
+    >>> print(classify("speak", {"ok": True, "error": "deprecated arg"}))
+    None
     """
     bad = frozenset(s.lower() for s in (*statuses, *extra_statuses))
     status_keys = tuple(status_keys)
     flag_keys = tuple(flag_keys)
+    error_keys = tuple(error_keys)
 
     def classify(tool: str, result: Any) -> str | None:
         if not isinstance(result, dict):
@@ -89,6 +108,12 @@ def soft_error_classifier(
             if result.get(key) is True:
                 detail = _message(result) or _content_text(result.get("content"))
                 return _describe("isError", detail)
+        if any(result.get(key) is True for key in flag_keys):
+            return None
+        for key in error_keys:
+            value = result.get(key)
+            if isinstance(value, str) and value.strip():
+                return _describe(key, value.strip())
         return None
 
     return classify

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

src/fastmcp_feedback/instrumentation/middleware.py

@@ -430,6 +430,9 @@ class InstrumentationMiddleware(Middleware):
         and session; ``call_id`` given here wins. ``attrs`` is redacted like
         call arguments and made JSON-safe. ``kind`` must be a non-empty string
         of at most 128 characters and ``key`` at most 255 once made a string.
+        A ``key`` used for joining must stay unique for the retention period
+        and across restarts, so not a per-boot counter: use a UUID or a
+        startup token plus a counter.
 
         Returns False when the event was not recorded: mode ``off``, an invalid
         ``kind``, ``key`` or ``attrs``, a full queue, or no event loop to hand
@@ -548,6 +551,7 @@ class InstrumentationMiddleware(Middleware):
         memory, so an event is visible as soon as ``record_event`` returns (on
         the loop thread). Without a DatabaseSink, memory only. ``occurred_at``
         is a UTC datetime. A database failure is logged and memory is used.
+        Events are ordered by ``occurred_at``; their ids are random UUIDs.
         """
         key = str(key)
         kinds = None if kinds is None else frozenset(kinds)

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)

tests/unit/test_retention_sampling.py

@@ -57,7 +57,7 @@ def make_app(sinks=None, **kwargs):
 
     @app.tool
     def render(status: str) -> dict:
-        return {"status": status, "error": f"render said {status}"}
+        return {"status": status, "detail": f"render said {status}"}
 
     @app.tool
     def crash() -> dict:

tests/unit/test_soft_errors.py

@@ -59,7 +59,7 @@ def make_app(sinks=None, **kwargs):
 
     @app.tool
     def render(status: str) -> dict:
-        return {"status": status, "error": f"render said {status}"}
+        return {"status": status, "detail": f"render said {status}"}
 
     @app.tool
     def respond(payload: dict) -> dict:
@@ -119,6 +119,19 @@ class TestClassifier:
                 {"is_error": True, "content": [{"type": "text", "text": "a"}]},
                 "isError: a",
             ),
+            # mcspeak's failures: an error string and no status or flag key.
+            (
+                {"played": False, "error": "audio output stalled after 30s"},
+                "error: audio output stalled after 30s",
+            ),
+            (
+                {"error": "Not listening: microphone stopped"},
+                "error: Not listening: microphone stopped",
+            ),
+            ({"error": "  padded  "}, "error: padded"),
+            ({"status": "unhealthy", "error": "db gone"}, "error: db gone"),
+            # Earlier checks still win, with their own message.
+            ({"success": False, "error": "x"}, "success=False: x"),
         ],
     )
     def test_flagged(self, result, expected):
@@ -135,6 +148,16 @@ class TestClassifier:
             {"result": False},
             {"state": 3},
             {"isError": False},
+            {"played": True, "error": None},
+            {"error": ""},
+            {"error": "   "},
+            {"error": 0.02},
+            {"error": 0},
+            {"error": False},
+            # A structured error is left to custom classifiers.
+            {"error": {"code": 1}},
+            {"error": ["a"]},
+            {"failure_reason": "x"},
             [{"status": "failed"}],
             "failed",
             None,
@@ -159,6 +182,88 @@ class TestClassifier:
         assert classify("t", {"passed": False}) == "passed=False"
         assert classify("t", {"status": "failed", "ok": False}) is None
 
+    def test_unknown_status_with_error_string(self):
+        """blender-mcp's shape: flagged by default, labelled by status when known."""
+        result = {"status": "unknown_target", "error": "no object named Cube"}
+        assert DEFAULT_RESULT_CLASSIFIER("t", result) == "error: no object named Cube"
+        classify = soft_error_classifier(extra_statuses=("unknown_target",))
+        assert classify("t", result) == "status=unknown_target: no object named Cube"
+
+    def test_custom_error_keys(self):
+        classify = soft_error_classifier(error_keys=("failure_reason",))
+        assert classify("t", {"failure_reason": "no gpu"}) == "failure_reason: no gpu"
+        assert classify("t", {"error": "ignored"}) is None
+        assert classify("t", {"failure_reason": ""}) is None
+
+    def test_empty_error_keys_disable(self):
+        classify = soft_error_classifier(error_keys=())
+        assert classify("t", {"played": False, "error": "stalled"}) is None
+        assert classify("t", {"ok": False, "error": "x"}) == "ok=False: x"
+
+
+class TestExplicitSuccess:
+    """A flag key that is exactly True turns an error string into a warning."""
+
+    @pytest.mark.parametrize(
+        "result",
+        [
+            {"ok": True, "error": "deprecated arg"},
+            {"success": True, "error": "x"},
+            {"ok": True, "status": "unknown_target", "error": "fell back"},
+        ],
+    )
+    def test_true_flag_with_error_string_is_ok(self, result):
+        assert DEFAULT_RESULT_CLASSIFIER("t", result) is None
+
+    @pytest.mark.parametrize(
+        "result, expected",
+        [
+            ({"ok": False, "error": "x"}, "ok=False: x"),
+            # Not exactly True, so not an explicit success.
+            ({"ok": 1, "error": "x"}, "error: x"),
+            ({"ok": "yes", "error": "x"}, "error: x"),
+            ({"success": 1.0, "error": "x"}, "error: x"),
+            # No flag key: mcspeak's shape, as in 2026.10.01.
+            (
+                {"played": False, "error": "audio output stalled"},
+                "error: audio output stalled",
+            ),
+            # A False flag beats a True one.
+            ({"ok": True, "success": False, "error": "x"}, "success=False: x"),
+            # A failing status is more specific than the flag, so it still flags.
+            ({"success": True, "status": "failed"}, "status=failed"),
+            ({"ok": True, "state": "error", "error": "x"}, "state=error: x"),
+            ({"ok": True, "isError": True, "content": ["boom"]}, "isError: boom"),
+        ],
+    )
+    def test_still_flagged(self, result, expected):
+        assert DEFAULT_RESULT_CLASSIFIER("t", result) == expected
+
+    def test_custom_flag_keys(self):
+        classify = soft_error_classifier(flag_keys=("passed",))
+        assert classify("t", {"passed": True, "error": "slow"}) is None
+        assert classify("t", {"passed": False, "error": "x"}) == "passed=False: x"
+        # ok is not a flag key here, so it does not suppress the error check.
+        assert classify("t", {"ok": True, "error": "x"}) == "error: x"
+
+    def test_no_flag_keys(self):
+        classify = soft_error_classifier(flag_keys=())
+        assert classify("t", {"ok": True, "error": "x"}) == "error: x"
+
+    async def test_recorded_as_ok(self):
+        record = await record_for(
+            "respond", {"payload": {"ok": True, "error": "deprecated arg"}}
+        )
+        assert (record.outcome, record.ok, record.error_message) == ("ok", True, None)
+
+    async def test_truthy_flag_recorded_as_soft_error(self):
+        record = await record_for("respond", {"payload": {"ok": 1, "error": "x"}})
+        assert (record.outcome, record.ok, record.error_message) == (
+            "soft_error",
+            False,
+            "error: x",
+        )
+
 
 class TestMiddleware:
     async def test_status_failed_is_soft_error(self):
@@ -186,6 +291,33 @@ class TestMiddleware:
         assert REDACTED in record.error_message
         assert len(record.error_message) == 60
 
+    async def test_error_string_is_soft_error(self):
+        record = await record_for(
+            "respond",
+            {"payload": {"played": False, "error": f"stalled, key {BMCP_TOKEN}"}},
+        )
+        assert (record.outcome, record.ok, record.error_type) == (
+            "soft_error",
+            False,
+            "SoftError",
+        )
+        assert record.error_message == f"error: stalled, key {REDACTED}"
+
+    @pytest.mark.parametrize("value", [None, "", 0.02])
+    async def test_empty_or_numeric_error_stays_ok(self, value):
+        record = await record_for(
+            "respond", {"payload": {"played": True, "error": value}}
+        )
+        assert (record.outcome, record.ok, record.error_message) == ("ok", True, None)
+
+    async def test_error_keys_disabled(self):
+        record = await record_for(
+            "respond",
+            {"payload": {"error": "stalled"}},
+            result_classifier=soft_error_classifier(error_keys=()),
+        )
+        assert (record.outcome, record.ok) == ("ok", True)
+
     async def test_extra_statuses(self):
         record = await record_for(
             "render",
@@ -393,3 +525,27 @@ class TestStorageAndCorrelation:
         assert record.error_message == (
             f"success=False: database down, password={REDACTED}"
         )
+
+    async def test_list_feedback_failure_is_soft_error(self, monkeypatch):
+        """list_feedback reports failure as {"feedback": [], "error": ...}."""
+        app = FastMCP("fb")
+        sink = MemorySink()
+        mw = instrument(app, [sink])
+        add_feedback_tools(app, database_url="sqlite:///:memory:", instrumentation=mw)
+
+        def broken(*args, **kwargs):
+            raise RuntimeError("database down")
+
+        from fastmcp_feedback.database import FeedbackDatabase
+
+        monkeypatch.setattr(FeedbackDatabase, "session", broken)
+        async with Client(app) as client:
+            result = await client.call_tool("list_feedback", {})
+        await mw.flush()
+        assert result.data["error"] == "database down"
+        [record] = sink.records
+        assert (record.tool, record.outcome, record.error_message) == (
+            "list_feedback",
+            "soft_error",
+            "error: database down",
+        )

uv.lock

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