Run the DatabaseSink tests against Postgres + pgvector as well as SQLite

The sink ships with a JSONB variant and is documented for asyncpg, but
the suite only ever exercised SQLite. The embeddings work will need
pgvector, so this adds a local Postgres stack and a way to run the
database tests on it, here and in CI.

- docker-compose.yml: ffb-postgres (pgvector/pgvector:pg16) on a private
  network, published on 127.0.0.1 only, with an init script that adds
  the vector extension to the test database and template1
- Makefile: pg-up/pg-down/pg-reset/pg-logs/pg-psql, test, test-pg,
  test-all, lint; .env is generated from .env.example with a random
  password, and FFB_TEST_DATABASE_URL from the environment wins over it
- tests: a db_url fixture parametrized over sqlite and postgres. The
  postgres leg skips unless FFB_TEST_DATABASE_URL is set, and creates a
  uniquely named database per test (with vector) that is dropped after.
  The DatabaseSink factories in the events, retention, correlation,
  soft-error and instrumentation tests now use it, so their 38 database
  tests run on both. The missing-events-table test accepts either
  backend's error. New tests check isolation and that vector works.
- asyncpg and pgvector added to the dev and test extras
- CI: a test-postgres job on FastMCP 2.12.2 and 4.x with a pgvector
  service container, failing if the Postgres cases skip
- CONTRIBUTING: how to run the suite against Postgres

.env.example

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

.gitea/workflows/test.yml

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

CHANGELOG.md

@@ -5,6 +5,140 @@ 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.29] - 2026-09-29
+
+Record things that are not tool calls: background job transitions, LLM
+requests a tool makes, anything with a dotted name.
+
+### Added
+- `mw.record_event(kind, *, key=None, attrs=None, user_sub=None,
+  call_id=None) -> bool`. Synchronous; never raises, never blocks. `attrs` is
+  redacted like call arguments and made JSON-safe (`NaN`/infinities become
+  strings, since Postgres JSONB rejects them). Returns False when the event
+  was not recorded: mode `off`, an empty or over-128-character `kind`, a `key`
+  over 255 characters, non-mapping `attrs`, a full queue, or no event loop.
+  Events share the call records' queue and drop-and-count policy.
+- **Linking to the running call.** The middleware now picks a call's record id
+  before the tool runs and exposes it through a ContextVar, so an event
+  recorded inside a tool gets that call's id as `call_id` (and its session),
+  matching the stored `ffb_tool_calls` row. Outside a call, `call_id` is None;
+  a task a tool started that records after the tool returned counts as
+  outside.
+- **Thread safety.** `record_event` works from any thread: off the server's
+  loop, the event is handed over with `call_soon_threadsafe`. The loop is
+  known from the first tool call (the dispatcher now binds to it as each call
+  starts), so sync tools running in FastMCP's worker threads link too. Before
+  any call or record, a thread with no loop of its own gets False and a
+  counted drop.
+- `mw.record_llm_call(*, model, provider=None, duration_ms=None,
+  input_tokens=None, output_tokens=None, ok=None, error=None, key=None,
+  attrs=None) -> bool` records kind `llm.call` with plain attribute names:
+  `model`, `provider`, `duration_ms`, `input_tokens`, `output_tokens`, `ok`
+  (defaults to `error is None`), `error` (redacted, truncated to
+  `max_error_chars`) and `error_type` for exceptions. The README maps them to
+  OTel GenAI names.
+- `await mw.events_for(key, *, kinds=None, limit=1000)`: a key's events, oldest
+  first (the newest `limit`), from the `DatabaseSink` plus events still in the
+  queue, deduplicated by id, `occurred_at` in UTC. Without a `DatabaseSink` it
+  answers from memory (`recent_events=2000` on the middleware).
+- `EventRecord`, exported from `fastmcp_feedback.instrumentation`.
+  `MemorySink` has `.calls` and `.events`.
+- Exclusion and sampling still apply to tool calls only. A sampled tool's call
+  that recorded an event is always kept (no `sample_rate`), so the event's
+  `call_id` has a row.
+
+### Changed
+- Sinks receive mixed batches of `ToolCallRecord` and `EventRecord` once
+  events are recorded. `record.record_type` is `tool_call` or `event`. A
+  `CallbackSink` callback that assumes every record is a call needs a check.
+- `JsonLinesSink` lines carry `"record": "tool_call"` or `"record": "event"`
+  (a new field; existing fields unchanged).
+- `DatabaseSink.write` stores calls and events in separate transactions,
+  calls first, so a database without the events table still stores calls.
+- `DatabaseSink.prune()` also deletes events whose `occurred_at` is before
+  the cutoff, in `prune_batch` batches, and **returns the total rows deleted
+  from both tables**; each table's count is logged. It skips events on a
+  database that has no events table.
+
+### Migration
+- New table `{prefix}events` (`ffb_events` by default): `id`, `occurred_at`,
+  `kind`, `key`, `call_id`, `session_id`, `user_sub`, `caller_kind`,
+  `client_id`, `server_version`, `attrs` (JSON/JSONB), indexed on
+  `occurred_at`, `kind`, `key`, `call_id` and `user_sub`. Hosts that manage the
+  schema with Alembic need a migration creating it (`build_metadata()`
+  includes it). `create_tables=True` creates it on first use. Until it exists,
+  event writes fail and are dropped (logged); calls are unaffected. No change
+  to existing tables.
+
+## [2026.09.28] - 2026-09-28
+
+Keep the tool-call table small: leave out or sample busy tools, and prune old
+rows.
+
+### Added
+- `InstrumentationMiddleware(..., exclude_tools=...)`: tools that are never
+  recorded and never kept for feedback links. The calls themselves are
+  untouched.
+- `InstrumentationMiddleware(..., sample_rates={"tool": 0.01})`: record that
+  fraction of a tool's `ok` calls. **Failures are always recorded**: calls with
+  outcome `error` or `soft_error` are kept whatever the rate. Rates outside
+  (0, 1] raise `ValueError`. Calls sampled out skip the hooks and redaction and
+  are not kept for feedback links. `random=` takes a `() -> float` for tests.
+- Records have a `sample_rate`: the configured rate on sampled-in `ok` rows,
+  None on everything recorded unconditionally (including failures of sampled
+  tools). `SUM(1.0 / COALESCE(sample_rate, 1))` estimates the true call count.
+- `DatabaseSink(..., retention=timedelta(days=30), prune_interval=...,
+  prune_batch=5000)`. With `retention` set, the background writer prunes after
+  an insert at most once per `prune_interval` (one hour by default). A failing
+  prune is logged and never stops the insert.
+- `DatabaseSink.prune(older_than=None) -> int` deletes older rows in batches
+  of `prune_batch`, one transaction each, on SQLite and Postgres. **Calls
+  linked to feedback are never pruned**, and links are never deleted.
+
+### Migration
+- New nullable `sample_rate` column (`Float`) on `{prefix}tool_calls`. Hosts
+  that manage the schema with Alembic need a migration adding it.
+  `create_tables=True` does not alter a table that already exists, so add it by
+  hand there (`ALTER TABLE ffb_tool_calls ADD COLUMN sample_rate FLOAT`). Until
+  the column exists, `DatabaseSink` writes fail and those records are dropped
+  (logged). Hosts still on 2026.09.27.3 or earlier also need the `outcome`
+  column from 2026.09.27.4.
+
+## [2026.09.27.4] - 2026-09-27
+
+Count errors that tools return as data instead of raising.
+
+### Added
+- Records have an `outcome`: `ok`, `soft_error` (the tool returned a result
+  flagged as a failure) or `error` (the tool raised). Soft errors get
+  `error_type="SoftError"` and an `error_message` built from the result, such
+  as `status=failed: disk full`, redacted and truncated to `max_error_chars`.
+- `soft_error_classifier(statuses=..., status_keys=..., flag_keys=...,
+  extra_statuses=...)` builds a classifier that flags dict results where `ok`
+  or `success` is `False`, where `status` or `state` is `error`, `failed` or
+  `failure`, or that the tool marked `isError` without raising.
+  `DEFAULT_RESULT_CLASSIFIER` is `soft_error_classifier()`.
+- `InstrumentationMiddleware(..., result_classifier=...)` takes any
+  `(tool, result) -> str | None`; `None` turns classification off. A
+  classifier that raises is logged and the call is recorded as `ok`.
+- `feedback_context()` rows include `outcome`.
+
+### Changed
+- `ok` is now `outcome == "ok"`, so it is also false for soft errors. Existing
+  `WHERE NOT ok` queries and error rates will rise; filter on
+  `outcome = 'error'` for exceptions only. `CallLink.ok` follows.
+- The package's own feedback tools answer `{"success": False, ...}` on failure,
+  so those calls are now recorded as soft errors.
+
+### Migration
+- New nullable, indexed `outcome` column (`String(16)`) on
+  `{prefix}tool_calls`. Hosts that manage the schema with Alembic need a
+  migration adding it. `create_tables=True` does not alter a table that
+  already exists, so add the column by hand there
+  (`ALTER TABLE ffb_tool_calls ADD COLUMN outcome VARCHAR(16)` plus an index).
+  Until the column exists, `DatabaseSink` writes fail and those records are
+  dropped (logged). Rows from earlier versions keep NULL.
+
 ## [2026.09.27.3] - 2026-09-27
 
 Link a feedback item to the tool calls that preceded it.

CONTRIBUTING.md

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

Makefile

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

README.md

@@ -128,7 +128,7 @@ app = FastMCP("My Server")
 instrument(app)  # one JSON line per tool call on stderr
 ```
 
-Each record has the tool name, start time, duration, success or error type,
+Each record has the tool name, start time, duration, outcome and error type,
 session and request ids, argument and result sizes, and whatever your hooks
 add. The recording happens in middleware, so it covers every tool on the server,
 not just the feedback ones.
@@ -185,8 +185,9 @@ instrument(app, [
 ### Using your own database and migrations
 
 Pass the `AsyncEngine` you already have. Records go to a `ffb_tool_calls`
-table (the prefix is configurable), and nothing is created at startup unless you
-ask. To manage the table with Alembic, add it to your `env.py`:
+table (events to `ffb_events`; the prefix is configurable), and nothing is
+created at startup unless you ask. To manage the tables with Alembic, add them
+to your `env.py`:
 
 ```python
 from fastmcp_feedback.instrumentation import build_metadata
@@ -201,6 +202,62 @@ sink = DatabaseSink(engine, prefix="ffb_")  # your engine is never disposed
 instrument(app, [sink])
 ```
 
+### Errors returned as data
+
+Many tools report failure in their result instead of raising, for example
+`{"status": "failed", "error": "disk full"}` or `{"success": False, ...}`. A
+result classifier catches these, so error rates count them. Each record has an
+`outcome`:
+
+| `outcome` | `ok` | Meaning |
+|-----------|------|---------|
+| `ok` | true | The call succeeded |
+| `soft_error` | false | The tool returned a result the classifier flagged |
+| `error` | false | The tool raised |
+
+Soft errors get `error_type="SoftError"` and an `error_message` such as
+`status=failed: disk full`, redacted and truncated like exception text.
+
+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`.
+Add your own status values, or pass any `(tool, result) -> str | None`:
+
+```python
+from fastmcp_feedback.instrumentation import instrument, soft_error_classifier
+
+# blender-mcp answers {"status": "unknown_target", ...} for a missing object
+instrument(app, result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)))
+```
+
+```python
+from fastmcp_feedback.instrumentation import instrument
+
+def classify(tool, result):
+    """None means ok; a string marks a soft error and is its message."""
+    if tool == "search" and isinstance(result, dict) and result.get("hits") == []:
+        return "no hits"
+    return None
+
+instrument(app, result_classifier=classify)
+```
+
+`result_classifier=None` turns classification off. The classifier only sees
+calls that did not raise, gets the same payload the record stores (structured
+content when present), and if it raises, the error is logged and the call is
+recorded as `ok`. It never changes what the client receives.
+
+**Upgrading:** soft errors set `ok` to false, so existing `WHERE NOT ok` counts
+include them from this release on. Filter on `outcome = 'error'` for exceptions
+only. The `outcome` column is new in 2026.09.27.4; it is nullable, and rows
+written by earlier versions have NULL there. Hosts that manage the schema with
+Alembic need a migration adding it (with its index) to `ffb_tool_calls`, and a
+table made earlier by `create_tables=True` needs
+`ALTER TABLE ffb_tool_calls ADD COLUMN outcome VARCHAR(16)`, since existing
+tables are not altered. Until the column exists, database writes fail and
+those records are dropped (logged).
+
 ### Identity and enrichment hooks
 
 ```python
@@ -242,7 +299,7 @@ Read the linked calls back, in order, with their outcomes:
 
 ```python
 for call in await mw.feedback_context(feedback_id):
-    print(call["position"], call["tool"], call["ok"], call["error_type"], call["rule"])
+    print(call["position"], call["tool"], call["outcome"], call["error_type"], call["rule"])
 ```
 
 If you have your own feedback tools, use the same pieces directly. Feedback ids
@@ -280,6 +337,203 @@ Code and FastMCP 4's own `Client`, do not open sessions over HTTP. Their calls
 link by user only, so configure an `identity_resolver` for HTTP servers. Stdio
 servers link by session with every client.
 
+### Busy tools: exclusion and sampling
+
+A polling tool can make up most of the rows. Leave it out entirely, or keep a
+fraction of its successful calls:
+
+```python
+from fastmcp_feedback.instrumentation import instrument
+
+instrument(
+    app,
+    exclude_tools={"heartbeat"},                     # never recorded
+    sample_rates={"pending_dispatches": 0.01},       # 1% of ok calls
+)
+```
+
+Excluded tools are not recorded at all and are left out of feedback links.
+Sampled tools keep a random fraction of their `ok` calls, but **failures are
+always recorded**: a call whose outcome is `error` or `soft_error` is kept
+whatever the rate. Calls sampled out cost no hook or redaction work and are
+left out of feedback links too. Rates must be in (0, 1]; anything else raises
+`ValueError`. Neither setting changes what the client receives.
+
+Sampled-in `ok` rows carry the rate in a `sample_rate` column; everything
+recorded unconditionally, including failures of sampled tools, has NULL there.
+So each row stands for `1 / coalesce(sample_rate, 1)` calls, and this estimates
+the true number of calls per tool:
+
+```sql
+SELECT tool, SUM(1.0 / COALESCE(sample_rate, 1)) AS est_calls
+FROM ffb_tool_calls GROUP BY tool;
+```
+
+### Retention
+
+`ffb_tool_calls` grows with every call. Give the sink a retention period and it
+deletes older rows as it goes:
+
+```python
+from datetime import timedelta
+from fastmcp_feedback.instrumentation import DatabaseSink, instrument
+
+sink = DatabaseSink(engine, prefix="ffb_", retention=timedelta(days=30))
+mw = instrument(app, [sink])
+```
+
+Pruning runs in the background writer after an insert, at most once per
+`prune_interval` (an hour by default; the first write after startup prunes).
+It deletes `prune_batch` rows (5000) per transaction, so it never holds a long
+lock, and a failing prune is logged without affecting the insert. **Calls linked
+to feedback are never pruned**, and neither are the links, so a report keeps its
+evidence. Without `retention`, nothing is deleted.
+
+To prune on your own schedule, call it directly; it returns the rows deleted:
+
+```python
+from datetime import UTC, datetime, timedelta
+
+deleted = await sink.prune()                                          # now - retention
+deleted = await sink.prune(older_than=datetime.now(UTC) - timedelta(days=7))
+```
+
+Automatic pruning runs inside a sink write, so it is bounded by the
+dispatcher's `sink_timeout` (10 s). Batches finished before a timeout stay
+deleted and the rest goes on the next run, but for a large backlog it is
+quicker to run `await sink.prune()` once yourself.
+
+**Upgrading:** `sample_rate` is a new nullable `FLOAT` column in 2026.09.28.
+Hosts that manage the schema with Alembic need a migration adding it, and a
+table made earlier by `create_tables=True` needs
+`ALTER TABLE ffb_tool_calls ADD COLUMN sample_rate FLOAT` (`DOUBLE PRECISION`
+on Postgres also works), since existing tables are not altered. Until the
+column exists, database writes fail and those records are dropped (logged).
+Retention needs no schema change.
+
+### Events: things that aren't tool calls
+
+Some things worth recording happen outside a tool call, or inside one without
+being one: a background job changing state minutes after the tool that
+started it returned, or an LLM request a tool makes on the caller's behalf.
+Record them as events:
+
+```python
+from fastmcp_feedback.instrumentation import DatabaseSink, instrument
+
+mw = instrument(app, [DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True)])
+
+@app.tool
+async def submit(case: str) -> dict:
+    job_id = f"job-{case}"
+    mw.record_event("job.submitted", key=job_id, attrs={"case": case, "nproc": 8})
+    return {"job_id": job_id}
+
+# Later, from a watcher task or thread, with no tool call running:
+mw.record_event("job.finished", key="job-wing", attrs={"duration_s": 3512, "converged": True})
+```
+
+`record_event(kind, *, key=None, attrs=None, user_sub=None, call_id=None)`
+returns True when the event was queued and False when it was not. It never
+raises and never blocks, and it takes the same path as call records: the same
+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.
+- `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
+  becomes `str(value)`.
+- **Inside a tool call**, the event gets that call's id as `call_id` and its
+  session, so it joins to the call's row in `ffb_tool_calls`. Outside a call,
+  `call_id` is None. A task started by a tool that records after the tool has
+  returned counts as outside. Pass `call_id=` to set it yourself.
+- **From any thread.** Called off the server's event loop (a watcher thread, a
+  sync tool running in a worker thread), the event is handed to the loop with
+  `call_soon_threadsafe`. The loop is known once the server has handled a tool
+  call or recorded something on it; before that, a call from a thread with no
+  event loop of its own returns False and counts a drop.
+- In mode `off` it does nothing and returns False.
+
+Read a key's events back, oldest first, with `occurred_at` as UTC:
+
+```python
+for ev in await mw.events_for("job-wing"):
+    print(ev["occurred_at"], ev["kind"], ev["call_id"], ev["attrs"])
+
+steps = await mw.events_for("job-wing", kinds=["step.started", "step.finished"], limit=100)
+```
+
+`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
+`DatabaseSink` it answers from the last `recent_events` (2000) events in
+memory; a failing database read is logged and memory is used.
+
+#### LLM calls
+
+`record_llm_call` records an event of kind `llm.call`. Call it from inside the
+tool that made the request and it links to that call:
+
+```python
+mw.record_llm_call(
+    model="claude-opus-4",
+    provider="anthropic",
+    duration_ms=812.5,
+    input_tokens=1200,
+    output_tokens=90,
+    attrs={"cache_read_input_tokens": 1000},
+)
+```
+
+Its attributes use plain names, which are easier to query from SQL than
+dotted ones (`attrs->>'model'` on Postgres, `json_extract(attrs, '$.model')`
+on SQLite):
+
+| Attribute | OTel GenAI equivalent | Notes |
+|-----------|-----------------------|-------|
+| `model` | `gen_ai.request.model` | Always set |
+| `provider` | `gen_ai.provider.name` | When given |
+| `duration_ms` | `gen_ai.client.operation.duration` (seconds) | When given |
+| `input_tokens` | `gen_ai.usage.input_tokens` | When given |
+| `output_tokens` | `gen_ai.usage.output_tokens` | When given |
+| `ok` | | Defaults to `error is None` |
+| `error` | | Redacted, truncated to `max_error_chars` |
+| `error_type` | `error.type` | When `error` is an exception |
+
+Anything in `attrs` is kept alongside, but these names win. `key=` works as
+for `record_event`, for example a chat turn id.
+
+#### Storage and sinks
+
+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.
+
+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
+`ffb_events` still stores calls. `JsonLinesSink` adds `"record": "tool_call"`
+or `"record": "event"` to every line. A `CallbackSink` callback receives both
+kinds (`record.record_type` tells them apart), and `MemorySink` has `.calls`
+and `.events` next to `.records`.
+
+Retention covers events too: `prune()` deletes events whose `occurred_at` is
+before the cutoff, in the same batches, and returns the rows deleted from both
+tables (each table's count is logged). Exclusion and sampling apply to tool
+calls only; events are always recorded. A call of a sampled tool that recorded
+an event is always kept, with no `sample_rate`, so the event's `call_id` has a
+row to join to. Events recorded inside an excluded tool have no `call_id`.
+
+**Upgrading:** `ffb_events` is a new table in this release. Hosts that manage
+the schema with Alembic need a migration creating it (`build_metadata()`
+includes it). `create_tables=True` creates it on first use, since it creates
+missing tables. Until the table exists, event writes fail and those events
+are dropped (logged); tool calls and pruning are unaffected.
+
 ### Middleware order and shutdown
 
 FastMCP runs the first middleware added as the outermost. Call `instrument()`
@@ -390,10 +644,17 @@ Full documentation: https://fastmcp-feedback.l.supported.systems
 
 ## Examples
 
-See the [fastmcp-feedback-showcase](https://git.supported.systems/fastmcp-feedback/fastmcp-example) repository for examples of every feature:
+The [`examples/`](https://git.supported.systems/fastmcp-feedback/fastmcp-feedback/src/branch/main/examples)
+directory has runnable servers, each with a walkthrough:
+
+- `simple_integration.py`: the one-line integration
+- `mcp_server.py`: a production-style server configured from environment variables
+- `demo_server.py`: every feature, including server composition and sample data
 
 ```bash
-uvx fastmcp-feedback-showcase  # Run full demo
+git clone https://git.supported.systems/fastmcp-feedback/fastmcp-feedback.git
+cd fastmcp-feedback
+uv run python examples/simple_integration.py
 ```
 
 ## Contributing
@@ -441,5 +702,4 @@ MIT License - see [LICENSE](LICENSE) for details.
 - **Documentation**: https://fastmcp-feedback.l.supported.systems
 - **Repository**: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback
 - **Issues**: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback/issues
-- **Showcase**: https://git.supported.systems/fastmcp-feedback/fastmcp-example
 - **FastMCP**: https://gofastmcp.com

docker-compose.yml

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

docker/postgres-init/01-vector.sql

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

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "fastmcp-feedback"
-version = "2026.09.27.3"  # YYYYMMDD.patch versioning - update for each release
+version = "2026.09.29"  # YYYYMMDD.patch versioning - update for each release
 description = "Production-ready feedback collection system for FastMCP servers"
 readme = "README.md"
 license = "MIT"
@@ -56,6 +56,8 @@ mysql = ["pymysql>=1.1.0"]
 dev = [
     "sqlalchemy[asyncio]>=2.0.43",
     "aiosqlite>=0.22.1",
+    "asyncpg>=0.31.0",
+    "pgvector>=0.5.0",
     "pytest>=8.2.0",
     "pytest-asyncio>=0.23.0",
     "pytest-cov>=5.0.0",
@@ -67,6 +69,8 @@ dev = [
 test = [
     "sqlalchemy[asyncio]>=2.0.43",
     "aiosqlite>=0.22.1",
+    "asyncpg>=0.31.0",
+    "pgvector>=0.5.0",
     "pytest>=8.2.0",
     "pytest-asyncio>=0.23.0",
     "pytest-cov>=5.0.0",
@@ -119,6 +123,7 @@ markers = [
     "integration: Integration tests for component interaction",
     "slow: Slow running tests",
     "database: Tests requiring database setup",
+    "postgres: Runs against the Postgres named by FFB_TEST_DATABASE_URL; skipped when unset",
 ]
 filterwarnings = [
     "ignore::DeprecationWarning",

src/fastmcp_feedback/__init__.py

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

src/fastmcp_feedback/instrumentation/__init__.py

@@ -4,9 +4,17 @@ from fastmcp_feedback.instrumentation import instrument, DatabaseSink
 
 mw = instrument(app)  # every tool call as JSON lines on stderr
 mw = instrument(app, [DatabaseSink(engine)], identity_resolver=who)
+mw = instrument(app, result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)))
+mw.record_event("job.finished", key=job_id, attrs={"duration_s": 3600})
 """
 
+from .classify import (
+    DEFAULT_RESULT_CLASSIFIER,
+    ResultClassifier,
+    soft_error_classifier,
+)
 from .dispatch import RecordDispatcher
+from .events import EventRecord
 from .middleware import MODE_ENV_VAR, MODES, InstrumentationMiddleware, instrument
 from .record import ToolCallRecord
 from .redaction import (
@@ -21,9 +29,11 @@ from .storage import DEFAULT_TABLE_PREFIX, DatabaseSink, build_metadata
 __all__ = [
     "CallbackSink",
     "DatabaseSink",
+    "DEFAULT_RESULT_CLASSIFIER",
     "DEFAULT_SECRET_KEYS",
     "DEFAULT_SECRET_PATTERNS",
     "DEFAULT_TABLE_PREFIX",
+    "EventRecord",
     "InstrumentationMiddleware",
     "JsonLinesSink",
     "MemorySink",
@@ -32,8 +42,10 @@ __all__ = [
     "RecordDispatcher",
     "REDACTED",
     "Redactor",
+    "ResultClassifier",
     "Sink",
     "ToolCallRecord",
     "build_metadata",
     "instrument",
+    "soft_error_classifier",
 ]

src/fastmcp_feedback/instrumentation/classify.py

@@ -0,0 +1,97 @@
+"""Spotting tools that report failure as data instead of raising.
+
+A result classifier is ``(tool, result) -> str | None``. ``result`` is the
+recorded payload: the structured content when there is one, otherwise the
+content items. None means the call succeeded; a string marks it as a soft
+error and becomes the record's ``error_message`` (after redaction).
+
+A result the tool marked as an MCP error (``isError``) without raising reaches
+the classifier as a dict with ``"isError": True``, alongside the payload's own
+keys or with the payload under ``"content"``.
+"""
+
+from collections.abc import Callable, Iterable
+from typing import Any
+
+ResultClassifier = Callable[[str, Any], str | None]
+
+MESSAGE_KEYS = ("error", "message", "detail", "reason")
+ERROR_FLAG_KEYS = ("isError", "is_error")
+
+
+def _message(result: dict, depth: int = 0) -> str | None:
+    for key in MESSAGE_KEYS:
+        value = result.get(key)
+        if value is None or value == "":
+            continue
+        if isinstance(value, dict):
+            # {"error": {"code": 404, "message": "..."}}
+            if depth == 0 and (inner := _message(value, 1)):
+                return inner
+            continue
+        if isinstance(value, list | tuple):
+            continue
+        return str(value)
+    return None
+
+
+def _content_text(content: Any) -> str | None:
+    if isinstance(content, str):
+        return content
+    if not isinstance(content, list | tuple):
+        return None
+    texts = [
+        item if isinstance(item, str) else item.get("text")
+        for item in content
+        if isinstance(item, str | dict)
+    ]
+    return " ".join(t for t in texts if t) or None
+
+
+def _describe(label: str, detail: str | None) -> str:
+    return f"{label}: {detail}" if detail else label
+
+
+def soft_error_classifier(
+    *,
+    statuses: Iterable[str] = ("error", "failed", "failure"),
+    status_keys: Iterable[str] = ("status", "state"),
+    flag_keys: Iterable[str] = ("ok", "success"),
+    extra_statuses: Iterable[str] = (),
+) -> ResultClassifier:
+    """Build a classifier for dict results that describe their own failure.
+
+    A dict result is a soft error when any of ``flag_keys`` is exactly False,
+    when any of ``status_keys`` holds one of ``statuses`` or
+    ``extra_statuses`` (case-insensitive), or when the tool marked it
+    ``isError``. The message is the first of ``error``, ``message``,
+    ``detail`` and ``reason`` present, e.g. ``status=failed: disk full``.
+
+    >>> classify = soft_error_classifier(extra_statuses=("unknown_target",))
+    >>> classify("render", {"status": "unknown_target", "error": "no such object"})
+    'status=unknown_target: no such object'
+    """
+    bad = frozenset(s.lower() for s in (*statuses, *extra_statuses))
+    status_keys = tuple(status_keys)
+    flag_keys = tuple(flag_keys)
+
+    def classify(tool: str, result: Any) -> str | None:
+        if not isinstance(result, dict):
+            return None
+        for key in status_keys:
+            value = result.get(key)
+            if isinstance(value, str) and value.lower() in bad:
+                return _describe(f"{key}={value}", _message(result))
+        for key in flag_keys:
+            if key in result and result[key] is False:
+                return _describe(f"{key}=False", _message(result))
+        for key in ERROR_FLAG_KEYS:
+            if result.get(key) is True:
+                detail = _message(result) or _content_text(result.get("content"))
+                return _describe("isError", detail)
+        return None
+
+    return classify
+
+
+DEFAULT_RESULT_CLASSIFIER = soft_error_classifier()

src/fastmcp_feedback/instrumentation/dispatch.py

@@ -3,10 +3,10 @@
 import asyncio
 import contextlib
 import logging
-from collections.abc import Sequence
+import threading
+from collections.abc import Callable, Sequence
 
-from .record import ToolCallRecord
-from .sinks import Sink
+from .sinks import Record, Sink
 
 logger = logging.getLogger(__name__)
 
@@ -33,11 +33,13 @@ class RecordDispatcher:
         self.sink_timeout = sink_timeout
         self.dropped = 0
         self.sink_errors = 0
-        self._queue: asyncio.Queue[ToolCallRecord] | None = None
+        self._queue: asyncio.Queue[Record] | None = None
         self._task: asyncio.Task | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
+        # submit_threadsafe can drop from any thread.
+        self._drop_lock = threading.Lock()
 
-    def submit(self, record: ToolCallRecord) -> bool:
+    def submit(self, record: Record) -> bool:
         """Queue a record for delivery. Returns False if it was dropped."""
         try:
             loop = asyncio.get_running_loop()
@@ -45,12 +47,7 @@ class RecordDispatcher:
             self._drop("no running event loop")
             return False
 
-        # A queue belongs to one loop; tests and some hosts run several.
-        if self._queue is None or self._loop is not loop:
-            self._loop = loop
-            self._queue = asyncio.Queue(maxsize=self.max_queue)
-            self._task = loop.create_task(self._run(self._queue))
-
+        self._bind(loop)
         try:
             self._queue.put_nowait(record)
         except asyncio.QueueFull:
@@ -58,11 +55,79 @@ class RecordDispatcher:
             return False
         return True
 
+    def bind_loop(self) -> None:
+        """Adopt the running loop, so other threads can submit before any record.
+
+        Called by the middleware as each tool call starts. Never raises.
+        """
+        with contextlib.suppress(Exception):
+            self._bind(asyncio.get_running_loop())
+
+    def _bind(self, loop: asyncio.AbstractEventLoop) -> None:
+        # A queue belongs to one loop; tests and some hosts run several.
+        if self._queue is None or self._loop is not loop:
+            self._loop = loop
+            self._queue = asyncio.Queue(maxsize=self.max_queue)
+            self._task = loop.create_task(self._run(self._queue))
+
+    def submit_threadsafe(
+        self,
+        record: Record,
+        on_queued: Callable[[Record], None] | None = None,
+    ) -> bool:
+        """``submit`` that may be called from any thread. Never blocks or raises.
+
+        On the loop the dispatcher runs on, or on any loop when it has none
+        running, this is ``submit``. From another thread the record is handed
+        to the dispatcher's loop with ``call_soon_threadsafe``; True then means
+        it was handed over, and a queue that fills before it lands still drops
+        and counts it. With no running loop anywhere the record is dropped.
+        ``on_queued(record)`` runs on the loop thread once the record is queued.
+        """
+        try:
+            try:
+                running = asyncio.get_running_loop()
+            except RuntimeError:
+                running = None
+            home = self._loop
+            home_alive = home is not None and home.is_running()
+            if running is not None and (not home_alive or home is running):
+                return self._submit_and_notify(record, on_queued)
+            if not home_alive:
+                self._drop("no running event loop")
+                return False
+            queue = self._queue
+            if queue is not None and queue.full():
+                self._drop("queue full")
+                return False
+            home.call_soon_threadsafe(self._submit_and_notify, record, on_queued)
+            return True
+        except Exception:
+            # call_soon_threadsafe raises if the loop closed after the check.
+            self._drop("event loop unavailable")
+            return False
+
+    def _submit_and_notify(
+        self, record: Record, on_queued: Callable[[Record], None] | None
+    ) -> bool:
+        if not self.submit(record):
+            return False
+        if on_queued is not None:
+            try:
+                on_queued(record)
+            except Exception:
+                logger.warning("on_queued callback failed", exc_info=True)
+        return True
+
     def _drop(self, reason: str) -> None:
-        self.dropped += 1
-        if self.dropped == 1 or self.dropped % 1000 == 0:
+        with self._drop_lock:
+            self.dropped += 1
+            dropped = self.dropped
+        if dropped == 1 or dropped % 1000 == 0:
             logger.warning(
-                "Dropped tool-call record (%s); %d dropped so far", reason, self.dropped
+                "Dropped instrumentation record (%s); %d dropped so far",
+                reason,
+                dropped,
             )
 
     async def _run(self, queue: asyncio.Queue) -> None:
@@ -79,7 +144,7 @@ class RecordDispatcher:
                 for _ in batch:
                     queue.task_done()
 
-    async def _deliver(self, batch: list[ToolCallRecord]) -> None:
+    async def _deliver(self, batch: list[Record]) -> None:
         for sink in self.sinks:
             try:
                 await asyncio.wait_for(sink.write(batch), self.sink_timeout)

src/fastmcp_feedback/instrumentation/events.py

@@ -0,0 +1,89 @@
+"""Events: things worth recording that are not tool calls.
+
+A background job changing state, an LLM request made on a tool's behalf. Each
+event has a dotted ``kind``, an optional ``key`` to join on (a job id), and,
+when recorded while a tool call is running, that call's id.
+"""
+
+import math
+import uuid
+from dataclasses import asdict, dataclass, field
+from datetime import datetime
+from typing import Any, ClassVar
+
+# Column widths in {prefix}events. Longer values are refused rather than
+# truncated, since a cut key would silently stop joining.
+MAX_KIND_CHARS = 128
+MAX_KEY_CHARS = 255
+
+
+@dataclass
+class EventRecord:
+    """One event, as passed to sinks next to ``ToolCallRecord``.
+
+    ``call_id`` is the id of the tool call that was running when the event was
+    recorded, None outside a call. ``attrs`` is already redacted.
+    """
+
+    record_type: ClassVar[str] = "event"
+
+    kind: str
+    occurred_at: datetime
+    id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    key: str | None = None
+    call_id: str | None = None
+    session_id: str | None = None
+    user_sub: str | None = None
+    caller_kind: str | None = None
+    client_id: str | None = None
+    server_version: str | None = None
+    attrs: dict[str, Any] = field(default_factory=dict)
+
+    def to_row(self) -> dict[str, Any]:
+        """Column values for the events table, with ``attrs`` made JSON-safe."""
+        data = asdict(self)
+        data["attrs"] = jsonable(data["attrs"])
+        return data
+
+    def to_dict(self) -> dict[str, Any]:
+        """JSON-friendly form, with the timestamp as ISO 8601."""
+        data = self.to_row()
+        data["occurred_at"] = self.occurred_at.isoformat()
+        return data
+
+
+@dataclass
+class CallScope:
+    """The tool call in progress, visible to ``record_event`` through a ContextVar.
+
+    Tasks started inside a tool inherit a copy of the context and outlive the
+    call, so the scope is switched off when the call ends instead of relying
+    on the ContextVar reset alone.
+    """
+
+    id: str
+    session_id: str | None
+    active: bool = True
+    events: int = 0
+
+
+def jsonable(value: Any, depth: int = 0) -> Any:
+    """A copy of ``value`` that any JSON column accepts.
+
+    Non-finite floats (a diverged residual) become strings, since Postgres
+    JSONB rejects NaN and one bad row would fail the whole batch. Other
+    unknown types become ``str(value)``.
+    """
+    if depth > 32:
+        return "[TRUNCATED]"
+    if value is None or isinstance(value, bool | int | str):
+        return value
+    if isinstance(value, float):
+        return value if math.isfinite(value) else str(value)
+    if isinstance(value, dict):
+        return {str(k): jsonable(v, depth + 1) for k, v in value.items()}
+    if isinstance(value, list | tuple | set | frozenset):
+        return [jsonable(v, depth + 1) for v in value]
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return str(value)

src/fastmcp_feedback/instrumentation/middleware.py

@@ -1,20 +1,24 @@
-"""FastMCP middleware that records every tool call."""
+"""FastMCP middleware that records every tool call, and events you report."""
 
 import asyncio
+import contextlib
+import contextvars
 import inspect
 import json
 import logging
 import os
+import random as _random
 import time
 import uuid
 from collections import deque
-from collections.abc import Awaitable, Callable, Iterable, Sequence
+from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
 from fastmcp import FastMCP
 from fastmcp.server.middleware import Middleware, MiddlewareContext
 
+from .classify import DEFAULT_RESULT_CLASSIFIER, ResultClassifier
 from .correlation import (
     CallLink,
     Candidate,
@@ -23,6 +27,13 @@ from .correlation import (
     select_calls,
 )
 from .dispatch import RecordDispatcher
+from .events import (
+    MAX_KEY_CHARS,
+    MAX_KIND_CHARS,
+    CallScope,
+    EventRecord,
+    jsonable,
+)
 from .record import ToolCallRecord
 from .redaction import Redactor
 from .sinks import JsonLinesSink, Sink
@@ -82,6 +93,10 @@ def session_key(ctx: Any) -> str | None:
     return _attr(ctx, "session_id")
 
 
+def _tool_name(context: MiddlewareContext) -> str:
+    return getattr(context.message, "name", None) or "unknown"
+
+
 def _result_payload(result: Any) -> Any:
     structured = getattr(result, "structured_content", None)
     if structured is not None:
@@ -131,6 +146,25 @@ class InstrumentationMiddleware(Middleware):
             correlation, including ones still waiting to reach the sinks.
         correlation_timeout: seconds a database lookup during correlation may
             take before it is skipped.
+        result_classifier: ``(tool, result) -> str | None`` run on results
+            of calls that did not raise. A string marks the call as a soft
+            error (``outcome="soft_error"``, ``ok=False``) and becomes the
+            error message. Defaults to ``DEFAULT_RESULT_CLASSIFIER``; None
+            turns it off. If it raises, the call is recorded as ok.
+        exclude_tools: tools that are never recorded, nor kept for
+            correlation. The calls themselves are untouched.
+        sample_rates: tool name to the fraction of its ok calls recorded, in
+            (0, 1]. Failures (``error`` and ``soft_error``) are always
+            recorded. Sampled-in ok rows carry ``sample_rate``; calls
+            sampled out are not kept for correlation either.
+        random: ``() -> float`` in [0, 1) deciding sampling; a call is kept
+            when it returns less than the tool's rate. For tests.
+        recent_events: how many recent events to keep in memory for
+            ``events_for``, including ones still waiting to reach the sinks.
+
+    Exclusion and sampling apply to tool calls only; events are always
+    recorded. A sampled tool's call is kept whenever an event was recorded
+    during it, so the event's ``call_id`` always has a row.
     """
 
     def __init__(
@@ -150,6 +184,11 @@ class InstrumentationMiddleware(Middleware):
         max_error_chars: int = 500,
         recent_calls: int = 2000,
         correlation_timeout: float = 1.0,
+        result_classifier: ResultClassifier | None = DEFAULT_RESULT_CLASSIFIER,
+        exclude_tools: Iterable[str] = (),
+        sample_rates: Mapping[str, float] | None = None,
+        random: Callable[[], float] = _random.random,
+        recent_events: int = 2000,
     ):
         mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
         if mode not in MODES:
@@ -163,7 +202,23 @@ class InstrumentationMiddleware(Middleware):
         self.hook_timeout = hook_timeout
         self.max_error_chars = max_error_chars
         self.correlation_timeout = correlation_timeout
+        self.result_classifier = result_classifier
+        self.exclude_tools = frozenset(exclude_tools)
+        self.sample_rates: dict[str, float] = {}
+        for tool, rate in (sample_rates or {}).items():
+            rate = float(rate)
+            if not 0 < rate <= 1:  # also rejects NaN
+                raise ValueError(
+                    f"sample rate for {tool!r} must be in (0, 1], got {rate!r}"
+                )
+            self.sample_rates[tool] = rate
+        self.random = random
         self.recent: deque[ToolCallRecord] = deque(maxlen=recent_calls)
+        self.recent_events: deque[EventRecord] = deque(maxlen=recent_events)
+        # One per middleware, so events only link to calls this one records.
+        self._call_scope: contextvars.ContextVar[CallScope | None] = (
+            contextvars.ContextVar(f"ffb_call_scope_{id(self):x}", default=None)
+        )
         self.dispatcher = RecordDispatcher(
             sinks if sinks is not None else [JsonLinesSink()],
             max_queue=max_queue,
@@ -174,11 +229,25 @@ class InstrumentationMiddleware(Middleware):
     async def on_call_tool(self, context: MiddlewareContext, call_next):
         if self.mode == "off":
             return await call_next(context)
+        # Sync tools run in worker threads; their events need the loop known.
+        self.dispatcher.bind_loop()
+        if _tool_name(context) in self.exclude_tools:
+            return await call_next(context)
 
         started_at = datetime.now(UTC)
         start = time.perf_counter()
         result = None
         error: BaseException | None = None
+        scope = token = None
+        try:
+            # The record id exists before the tool runs, so events recorded
+            # inside it carry the id the stored call row will have.
+            scope = CallScope(
+                id=str(uuid.uuid4()), session_id=session_key(context.fastmcp_context)
+            )
+            token = self._call_scope.set(scope)
+        except Exception:
+            logger.warning("Failed to open call scope", exc_info=True)
         try:
             result = await call_next(context)
             return result
@@ -187,15 +256,50 @@ class InstrumentationMiddleware(Middleware):
             raise
         finally:
             duration_ms = (time.perf_counter() - start) * 1000
+            if scope is not None:
+                scope.active = False
+            if token is not None:
+                # ValueError when finishing in another context (a task group).
+                with contextlib.suppress(ValueError):
+                    self._call_scope.reset(token)
             try:
-                record = await self._build_record(
-                    context, started_at, duration_ms, result, error
+                await self._record(
+                    context, started_at, duration_ms, result, error, scope
                 )
-                self.recent.append(record)
-                self.dispatcher.submit(record)
             except Exception:
                 logger.warning("Failed to record tool call", exc_info=True)
 
+    async def _record(
+        self,
+        context: MiddlewareContext,
+        started_at: datetime,
+        duration_ms: float,
+        result: Any,
+        error: BaseException | None,
+        scope: CallScope | None = None,
+    ) -> None:
+        tool = _tool_name(context)
+        payload = _result_payload(result) if result is not None else None
+        soft = None
+        if error is None and self.result_classifier is not None and payload is not None:
+            soft = self._classify(tool, result, payload)
+        rate = self.sample_rates.get(tool)
+        # Failures are always kept, and their rows carry no sample_rate. So are
+        # calls that recorded events, which would otherwise point at nothing.
+        if rate is not None and (
+            error is not None or soft is not None or (scope and scope.events)
+        ):
+            rate = None
+        elif rate is not None and not self.random() < rate:
+            return
+        record = await self._build_record(
+            context, started_at, duration_ms, payload, error, soft, rate
+        )
+        if scope is not None:
+            record.id = scope.id
+        self.recent.append(record)
+        self.dispatcher.submit(record)
+
     async def _call_hook(self, hook: Callable | None, *args: Any) -> dict[str, Any]:
         if hook is None:
             return {}
@@ -208,19 +312,35 @@ class InstrumentationMiddleware(Middleware):
             logger.warning("Instrumentation hook %r failed", hook, exc_info=True)
             return {}
 
+    def _classify(self, tool: str, result: Any, payload: Any) -> str | None:
+        if getattr(result, "is_error", False) is True:
+            # FastMCP 3.4+ keeps isError on the ToolResult, not in the payload.
+            if isinstance(payload, dict):
+                payload = {**payload, "isError": True}
+            else:
+                payload = {"isError": True, "content": payload}
+        try:
+            message = self.result_classifier(tool, payload)
+        except Exception:
+            logger.warning(
+                "Result classifier %r failed", self.result_classifier, exc_info=True
+            )
+            return None
+        return None if message is None else str(message)
+
     async def _build_record(
         self,
         context: MiddlewareContext,
         started_at: datetime,
         duration_ms: float,
-        result: Any,
+        payload: Any,
         error: BaseException | None,
+        soft: str | None,
+        sample_rate: float | None,
     ) -> ToolCallRecord:
-        message = context.message
-        tool = getattr(message, "name", None) or "unknown"
-        args = getattr(message, "arguments", None) or {}
+        tool = _tool_name(context)
+        args = getattr(context.message, "arguments", None) or {}
         ctx = context.fastmcp_context
-        payload = _result_payload(result) if result is not None else None
 
         identity = await self._call_hook(self.identity_resolver, context)
         extra = await self._call_hook(self.enricher, tool, args, payload, context)
@@ -232,20 +352,29 @@ class InstrumentationMiddleware(Middleware):
         mode = "meta" if tool in self.meta_only_tools else self.mode
         full = mode == "full"
 
+        outcome = "ok"
         error_type = error_message = None
         if error is not None:
             # FastMCP wraps tool exceptions in ToolError; report the original.
             cause = error.__cause__ or error
+            outcome = "error"
             error_type = type(cause).__name__
             error_message = self.redactor.redact_text(str(cause))[
                 : self.max_error_chars
             ]
+        elif soft is not None:
+            outcome = "soft_error"
+            error_type = "SoftError"
+            error_message = (
+                self.redactor.redact_text(soft)[: self.max_error_chars] or None
+            )
 
         return ToolCallRecord(
             tool=tool,
             started_at=started_at,
             duration_ms=duration_ms,
-            ok=error is None,
+            ok=outcome == "ok",
+            outcome=outcome,
             mode=mode,
             error_type=error_type,
             error_message=error_message,
@@ -263,8 +392,147 @@ class InstrumentationMiddleware(Middleware):
             if full and payload is not None
             else None,
             extra=extra,
+            sample_rate=sample_rate,
         )
 
+    def record_event(
+        self,
+        kind: str,
+        *,
+        key: Any = None,
+        attrs: Mapping[str, Any] | None = None,
+        user_sub: str | None = None,
+        call_id: str | None = None,
+    ) -> bool:
+        """Record something that is not a tool call. Never raises or blocks.
+
+        Safe from any thread. Inside a tool call the event gets that call's id
+        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.
+
+        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
+        it to (called from a thread before anything was recorded on the
+        server's loop). From another thread, True means handed to the loop.
+        """
+        try:
+            if self.mode == "off":
+                return False
+            if not isinstance(kind, str) or not kind.strip():
+                logger.warning("record_event needs a non-empty str kind, got %r", kind)
+                return False
+            if len(kind) > MAX_KIND_CHARS:
+                logger.warning(
+                    "Event kind longer than %d chars: %r", MAX_KIND_CHARS, kind
+                )
+                return False
+            if key is not None:
+                key = str(key)
+                if len(key) > MAX_KEY_CHARS:
+                    logger.warning("Event key longer than %d chars", MAX_KEY_CHARS)
+                    return False
+            if attrs is not None and not isinstance(attrs, Mapping):
+                logger.warning("Event attrs must be a mapping, got %s", type(attrs))
+                return False
+            session_id = None
+            scope = self._call_scope.get()
+            if scope is not None and scope.active:
+                scope.events += 1
+                session_id = scope.session_id
+                if call_id is None:
+                    call_id = scope.id
+            record = EventRecord(
+                kind=kind,
+                occurred_at=datetime.now(UTC),
+                key=key,
+                call_id=None if call_id is None else str(call_id),
+                session_id=session_id,
+                user_sub=None if user_sub is None else str(user_sub),
+                server_version=self.server_version,
+                attrs=jsonable(self.redactor.redact(dict(attrs or {}))),
+            )
+            return self.dispatcher.submit_threadsafe(record, self.recent_events.append)
+        except Exception:
+            logger.warning("Failed to record event %r", kind, exc_info=True)
+            return False
+
+    def record_llm_call(
+        self,
+        *,
+        model: str,
+        provider: str | None = None,
+        duration_ms: float | None = None,
+        input_tokens: int | None = None,
+        output_tokens: int | None = None,
+        ok: bool | None = None,
+        error: str | BaseException | None = None,
+        key: Any = None,
+        attrs: Mapping[str, Any] | None = None,
+    ) -> bool:
+        """Record one LLM request as an ``llm.call`` event. Never raises.
+
+        Attributes: ``model``, ``provider``, ``duration_ms``, ``input_tokens``,
+        ``output_tokens`` (each only when given), ``ok``, and on failure
+        ``error`` (redacted, truncated to ``max_error_chars``) plus
+        ``error_type`` when ``error`` is an exception. ``ok`` defaults to
+        ``error is None``. Extra ``attrs`` are kept, but these names win.
+        """
+        try:
+            fields: dict[str, Any] = dict(attrs or {})
+            fields["model"] = None if model is None else str(model)
+            for name, value in (
+                ("provider", provider),
+                ("duration_ms", duration_ms),
+                ("input_tokens", input_tokens),
+                ("output_tokens", output_tokens),
+            ):
+                if value is not None:
+                    fields[name] = value
+            fields["ok"] = (error is None) if ok is None else bool(ok)
+            if error is not None:
+                if isinstance(error, BaseException):
+                    fields["error_type"] = type(error).__name__
+                text = self.redactor.redact_text(str(error))
+                fields["error"] = text[: self.max_error_chars]
+            return self.record_event("llm.call", key=key, attrs=fields)
+        except Exception:
+            logger.warning("Failed to record LLM call", exc_info=True)
+            return False
+
+    async def events_for(
+        self,
+        key: Any,
+        *,
+        kinds: Iterable[str] | None = None,
+        limit: int = 1000,
+    ) -> list[dict[str, Any]]:
+        """Events for ``key``, oldest first: the newest ``limit`` of them.
+
+        Reads the DatabaseSink when there is one and adds events still only in
+        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.
+        """
+        key = str(key)
+        kinds = None if kinds is None else frozenset(kinds)
+        found: dict[str, dict[str, Any]] = {}
+        sink = self.database_sink
+        if sink is not None:
+            try:
+                for row in await sink.events_for(key, kinds=kinds, limit=limit):
+                    found[row["id"]] = row
+            except Exception:
+                logger.warning("Failed to read events for %r", key, exc_info=True)
+        for record in list(self.recent_events):
+            if record.key == key and (kinds is None or record.kind in kinds):
+                found.setdefault(record.id, record.to_row())
+        rows = list(found.values())
+        for row in rows:
+            row["occurred_at"] = as_utc(row["occurred_at"])
+        rows.sort(key=lambda r: r["occurred_at"])
+        return rows[-limit:] if limit > 0 else []
+
     @property
     def database_sink(self) -> DatabaseSink | None:
         """The first DatabaseSink among the sinks, if any."""

src/fastmcp_feedback/instrumentation/record.py

@@ -3,7 +3,7 @@
 import uuid
 from dataclasses import asdict, dataclass, field
 from datetime import datetime
-from typing import Any
+from typing import Any, ClassVar
 
 
 @dataclass
@@ -12,13 +12,24 @@ class ToolCallRecord:
 
     ``args`` and ``result`` are only populated in ``full`` mode, and are
     already redacted when they are. Sizes are measured before redaction.
+
+    ``outcome`` is ``ok``, ``error`` (the tool raised) or ``soft_error`` (it
+    returned a result the classifier flagged). ``ok`` is ``outcome == "ok"``;
+    when ``outcome`` is not given it is derived from ``ok``.
+
+    ``sample_rate`` is set on ok calls of a sampled tool: each such row stands
+    for ``1 / sample_rate`` calls. It is None for everything recorded
+    unconditionally, including failures of sampled tools.
     """
 
+    record_type: ClassVar[str] = "tool_call"
+
     tool: str
     started_at: datetime
     duration_ms: float
     ok: bool
     mode: str
+    outcome: str | None = None
     id: str = field(default_factory=lambda: str(uuid.uuid4()))
     error_type: str | None = None
     error_message: str | None = None
@@ -34,6 +45,11 @@ class ToolCallRecord:
     args: Any = None
     result: Any = None
     extra: dict[str, Any] = field(default_factory=dict)
+    sample_rate: float | None = None
+
+    def __post_init__(self) -> None:
+        if self.outcome is None:
+            self.outcome = "ok" if self.ok else "error"
 
     def to_row(self) -> dict[str, Any]:
         """Column values for the tool_calls table."""

src/fastmcp_feedback/instrumentation/sinks.py

@@ -1,4 +1,4 @@
-"""Destinations for tool-call records."""
+"""Destinations for tool-call and event records."""
 
 import inspect
 import json
@@ -6,27 +6,38 @@ import sys
 from collections.abc import Callable, Sequence
 from typing import IO, Any, Protocol, runtime_checkable
 
+from .events import EventRecord
 from .record import ToolCallRecord
 
+Record = ToolCallRecord | EventRecord
+
 
 @runtime_checkable
 class Sink(Protocol):
-    """Anything with ``async write(records)``. ``async aclose()`` is optional."""
+    """Anything with ``async write(records)``. ``async aclose()`` is optional.
 
-    async def write(self, records: Sequence[ToolCallRecord]) -> None: ...
+    A batch can mix ``ToolCallRecord`` and ``EventRecord``; tell them apart
+    with ``isinstance`` or ``record.record_type`` (``tool_call``, ``event``).
+    """
+
+    async def write(self, records: Sequence[Record]) -> None: ...
 
 
 class JsonLinesSink:
-    """One JSON object per record, to stderr by default."""
+    """One JSON object per record, to stderr by default.
+
+    Each line has ``"record": "tool_call"`` or ``"record": "event"``.
+    """
 
     def __init__(self, stream: IO[str] | None = None):
         self._stream = stream
 
-    async def write(self, records: Sequence[ToolCallRecord]) -> None:
+    async def write(self, records: Sequence[Record]) -> None:
         # Resolved per write so a stderr swapped by the host or a test is honored.
         stream = self._stream or sys.stderr
         for record in records:
-            stream.write(json.dumps(record.to_dict(), default=str) + "\n")
+            line = {"record": record.record_type, **record.to_dict()}
+            stream.write(json.dumps(line, default=str) + "\n")
         stream.flush()
 
 
@@ -36,20 +47,32 @@ class CallbackSink:
     A sync callback runs on the event loop, so keep it quick.
     """
 
-    def __init__(self, callback: Callable[[Sequence[ToolCallRecord]], Any]):
+    def __init__(self, callback: Callable[[Sequence[Record]], Any]):
         self.callback = callback
 
-    async def write(self, records: Sequence[ToolCallRecord]) -> None:
+    async def write(self, records: Sequence[Record]) -> None:
         result = self.callback(records)
         if inspect.isawaitable(result):
             await result
 
 
 class MemorySink:
-    """Keep records in a list. Meant for tests."""
+    """Keep records in a list. Meant for tests.
+
+    ``records`` holds everything in arrival order; ``calls`` and ``events``
+    are the two kinds on their own.
+    """
 
     def __init__(self):
-        self.records: list[ToolCallRecord] = []
+        self.records: list[Record] = []
 
-    async def write(self, records: Sequence[ToolCallRecord]) -> None:
+    @property
+    def calls(self) -> list[ToolCallRecord]:
+        return [r for r in self.records if isinstance(r, ToolCallRecord)]
+
+    @property
+    def events(self) -> list[EventRecord]:
+        return [r for r in self.records if isinstance(r, EventRecord)]
+
+    async def write(self, records: Sequence[Record]) -> None:
         self.records.extend(records)

src/fastmcp_feedback/instrumentation/storage.py

@@ -1,4 +1,4 @@
-"""Async SQLAlchemy storage for tool-call records and feedback links.
+"""Async SQLAlchemy storage for tool-call records, events and feedback links.
 
 Hosts that manage their own schema include ``build_metadata(prefix)`` in their
 Alembic ``env.py`` (``target_metadata``) and pass their existing AsyncEngine to
@@ -6,8 +6,10 @@ Alembic ``env.py`` (``target_metadata``) and pass their existing AsyncEngine to
 """
 
 import asyncio
+import logging
+import time
 from collections.abc import Sequence
-from datetime import datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
 from sqlalchemy import (
@@ -23,13 +25,19 @@ from sqlalchemy import (
     Text,
     and_,
     delete,
+    exists,
+    inspect,
     or_,
     select,
 )
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 
+from .events import MAX_KEY_CHARS, MAX_KIND_CHARS, EventRecord
 from .record import ToolCallRecord
+from .sinks import Record
+
+logger = logging.getLogger(__name__)
 
 DEFAULT_TABLE_PREFIX = "ffb_"
 
@@ -40,12 +48,14 @@ _JSON = JSON(none_as_null=True).with_variant(JSONB(none_as_null=True), "postgres
 def build_metadata(
     prefix: str = DEFAULT_TABLE_PREFIX, metadata: MetaData | None = None
 ) -> MetaData:
-    """Return MetaData holding ``{prefix}tool_calls`` and ``{prefix}feedback_call_links``.
+    """Return MetaData holding ``{prefix}tool_calls``, ``{prefix}events`` and
+    ``{prefix}feedback_call_links``.
 
     Pass an existing MetaData to add the tables to it instead.
     """
     metadata = metadata if metadata is not None else MetaData()
     calls = f"{prefix}tool_calls"
+    events = f"{prefix}events"
     links = f"{prefix}feedback_call_links"
     if calls not in metadata.tables:
         Table(
@@ -57,6 +67,8 @@ def build_metadata(
             Column("tool", String(255), nullable=False, index=True),
             Column("mode", String(8), nullable=False),
             Column("ok", Boolean, nullable=False, index=True),
+            # ok, soft_error or error. NULL on rows from before 2026.09.27.4.
+            Column("outcome", String(16), index=True),
             Column("error_type", String(255)),
             Column("error_message", Text),
             Column("session_id", String(255), index=True),
@@ -71,6 +83,27 @@ def build_metadata(
             Column("args", _JSON),
             Column("result", _JSON),
             Column("extra", _JSON),
+            # Set on sampled-in ok rows; each stands for 1/sample_rate calls.
+            # NULL for rows recorded unconditionally and before 2026.09.28.
+            Column("sample_rate", Float),
+        )
+    if events not in metadata.tables:
+        # No foreign key to tool_calls for the same reasons as links: an event
+        # can be written before its call leaves the queue, and calls are pruned.
+        Table(
+            events,
+            metadata,
+            Column("id", String(36), primary_key=True),
+            Column("occurred_at", DateTime(timezone=True), nullable=False, index=True),
+            Column("kind", String(MAX_KIND_CHARS), nullable=False, index=True),
+            Column("key", String(MAX_KEY_CHARS), index=True),
+            Column("call_id", String(36), index=True),
+            Column("session_id", String(255)),
+            Column("user_sub", String(255), index=True),
+            Column("caller_kind", String(64)),
+            Column("client_id", String(255)),
+            Column("server_version", String(64)),
+            Column("attrs", _JSON),
         )
     if links not in metadata.tables:
         # No foreign key to tool_calls: a link can be written before its call
@@ -88,7 +121,7 @@ def build_metadata(
 
 
 class DatabaseSink:
-    """Write records to ``{prefix}tool_calls`` through an async engine.
+    """Write records to ``{prefix}tool_calls`` and ``{prefix}events``.
 
     Args:
         engine: an AsyncEngine the host already owns, or a URL such as
@@ -98,6 +131,13 @@ class DatabaseSink:
         prefix: table name prefix, so the tables cannot collide with the host's.
         create_tables: create the tables on first use. Off by default; hosts
             with Alembic should generate a migration from ``build_metadata``.
+        retention: delete tool-call and event rows older than this. When set, ``write``
+            runs ``prune()`` at most once per ``prune_interval``, after its
+            insert, in the background writer. None (the default) keeps rows
+            forever. Calls linked to feedback are never pruned.
+        prune_interval: minimum time between automatic prunes.
+        prune_batch: rows deleted per statement, so no single delete holds
+            locks for long.
     """
 
     def __init__(
@@ -106,17 +146,30 @@ class DatabaseSink:
         *,
         prefix: str = DEFAULT_TABLE_PREFIX,
         create_tables: bool = False,
+        retention: timedelta | None = None,
+        prune_interval: timedelta = timedelta(hours=1),
+        prune_batch: int = 5000,
     ):
+        if retention is not None and retention <= timedelta(0):
+            raise ValueError(f"retention must be positive, got {retention!r}")
+        if prune_batch < 1:
+            raise ValueError(f"prune_batch must be at least 1, got {prune_batch!r}")
         self._owns_engine = isinstance(engine, str)
         self.engine = create_async_engine(engine) if isinstance(engine, str) else engine
         self.metadata = build_metadata(prefix)
         self.table = self.metadata.tables[f"{prefix}tool_calls"]
+        self.events = self.metadata.tables[f"{prefix}events"]
         self.links = self.metadata.tables[f"{prefix}feedback_call_links"]
         self.create_tables = create_tables
         self._tables_ready = False
         # The background writer and link_feedback can both reach _ready() at
         # once; without the lock both run create_all and one fails.
         self._tables_lock = asyncio.Lock()
+        self.retention = retention
+        self.prune_interval = prune_interval
+        self.prune_batch = prune_batch
+        # time.monotonic() of the last automatic prune; None until the first.
+        self._last_prune: float | None = None
 
     async def ensure_tables(self) -> None:
         """Create the tables if missing. Only called when create_tables=True."""
@@ -130,12 +183,106 @@ class DatabaseSink:
                 if not self._tables_ready:
                     await self.ensure_tables()
 
-    async def write(self, records: Sequence[ToolCallRecord]) -> None:
-        if not records:
+    async def write(self, records: Sequence[Record]) -> None:
+        calls = [r.to_row() for r in records if isinstance(r, ToolCallRecord)]
+        events = [r.to_row() for r in records if isinstance(r, EventRecord)]
+        if not calls and not events:
             return
         await self._ready()
-        async with self.engine.begin() as conn:
-            await conn.execute(self.table.insert(), [r.to_row() for r in records])
+        # Separate transactions, calls first: a host that has not created the
+        # events table yet still gets its calls stored.
+        if calls:
+            async with self.engine.begin() as conn:
+                await conn.execute(self.table.insert(), calls)
+        if events:
+            async with self.engine.begin() as conn:
+                await conn.execute(self.events.insert(), events)
+        await self._auto_prune()
+
+    async def _auto_prune(self) -> None:
+        if self.retention is None:
+            return
+        now = time.monotonic()
+        if (
+            self._last_prune is not None
+            and now - self._last_prune < self.prune_interval.total_seconds()
+        ):
+            return
+        # Set before running, so a prune that fails is not retried every write.
+        self._last_prune = now
+        try:
+            await self.prune()
+        except Exception:
+            logger.warning("Pruning %s failed", self.table.name, exc_info=True)
+
+    async def prune(self, older_than: datetime | None = None) -> int:
+        """Delete tool-call and event rows from before ``older_than``.
+
+        Defaults to now minus ``retention``; with neither, deletes nothing.
+        Calls linked to feedback in ``{prefix}feedback_call_links`` are kept,
+        and links are never deleted. Events go by ``occurred_at``, whatever
+        call they belong to. Deletes ``prune_batch`` rows per transaction until
+        none are left. Returns the total rows deleted from both tables and
+        logs each table's count. A database without the events table (not yet
+        migrated) has its calls pruned and the events step skipped.
+        """
+        if older_than is None:
+            if self.retention is None:
+                return 0
+            older_than = datetime.now(UTC) - self.retention
+        await self._ready()
+        t, link = self.table, self.links
+        calls = await self._delete_batches(
+            t,
+            select(t.c.id).where(
+                t.c.started_at < older_than,
+                ~exists().where(link.c.call_id == t.c.id),
+            ),
+        )
+        events = 0
+        if await self._has_table(self.events.name):
+            e = self.events
+            events = await self._delete_batches(
+                e, select(e.c.id).where(e.c.occurred_at < older_than)
+            )
+        if calls or events:
+            logger.info(
+                "Pruned %d rows from %s and %d from %s",
+                calls,
+                t.name,
+                events,
+                self.events.name,
+            )
+        return calls + events
+
+    async def _delete_batches(self, table: Any, ids_query: Any) -> int:
+        batch = ids_query.limit(self.prune_batch)
+        total = 0
+        while True:
+            async with self.engine.begin() as conn:
+                ids = (await conn.execute(batch)).scalars().all()
+                if ids:
+                    await conn.execute(delete(table).where(table.c.id.in_(ids)))
+            total += len(ids)
+            if len(ids) < self.prune_batch:
+                return total
+
+    async def _has_table(self, name: str) -> bool:
+        async with self.engine.connect() as conn:
+            return await conn.run_sync(lambda sync: inspect(sync).has_table(name))
+
+    async def events_for(
+        self, key: str, *, kinds: Sequence[str] | None = None, limit: int = 1000
+    ) -> list[dict[str, Any]]:
+        """The newest ``limit`` events for ``key``, newest first."""
+        await self._ready()
+        e = self.events
+        query = select(e).where(e.c.key == key)
+        if kinds is not None:
+            query = query.where(e.c.kind.in_(list(kinds)))
+        query = query.order_by(e.c.occurred_at.desc()).limit(limit)
+        async with self.engine.connect() as conn:
+            return [dict(row) for row in (await conn.execute(query)).mappings()]
 
     async def recent_calls(
         self,
@@ -194,6 +341,7 @@ class DatabaseSink:
                 call.c.started_at,
                 call.c.duration_ms,
                 call.c.ok,
+                call.c.outcome,
                 call.c.error_type,
                 call.c.error_message,
                 call.c.session_id,

tests/_databases.py

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

tests/conftest.py

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

tests/unit/test_correlation.py

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

tests/unit/test_events.py

@@ -0,0 +1,582 @@
+"""Tests for events: record_event, record_llm_call, events_for and their storage.
+
+Real FastMCP servers through fastmcp.Client, as in test_instrumentation.py.
+"""
+
+import asyncio
+import contextlib
+import io
+import json
+import threading
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from fastmcp import Client, FastMCP
+from fastmcp.exceptions import ToolError
+from sqlalchemy import MetaData, func, select
+from sqlalchemy.exc import DBAPIError
+
+from fastmcp_feedback.instrumentation import (
+    REDACTED,
+    CallbackSink,
+    DatabaseSink,
+    EventRecord,
+    JsonLinesSink,
+    MemorySink,
+    ToolCallRecord,
+    build_metadata,
+    instrument,
+)
+
+
+@pytest.fixture
+async def db(db_url):
+    """DatabaseSinks whose engines are always disposed, or the process hangs."""
+    made = []
+
+    def factory(name="calls", **kwargs):
+        kwargs.setdefault("create_tables", True)
+        sink = DatabaseSink(db_url(name), **kwargs)
+        made.append(sink)
+        return sink
+
+    yield factory
+    for sink in made:
+        await sink.engine.dispose()
+
+
+def make_app(sinks=None, **kwargs):
+    """A server whose tools record events through the middleware returned."""
+    sink = MemorySink()
+    app = FastMCP("events")
+    holder = {}
+    spawned = []
+
+    @app.tool
+    async def submit(job_id: str) -> dict:
+        mw = holder["mw"]
+        mw.record_event("job.submitted", key=job_id, attrs={"case": "wing", "nproc": 4})
+        return {"job_id": job_id}
+
+    @app.tool
+    def submit_sync(job_id: str) -> dict:
+        holder["mw"].record_event("job.submitted", key=job_id)
+        return {"job_id": job_id}
+
+    @app.tool
+    async def spawn(job_id: str) -> dict:
+        """Start a watcher task that records after the call has returned."""
+        started = asyncio.Event()
+
+        async def watcher():
+            started.set()
+            await asyncio.sleep(0.05)
+            holder["mw"].record_event("job.started", key=job_id)
+
+        spawned.append(asyncio.get_running_loop().create_task(watcher()))
+        await started.wait()
+        return {"job_id": job_id}
+
+    @app.tool
+    async def chat(prompt: str) -> str:
+        holder["mw"].record_llm_call(
+            model="claude-opus-4",
+            provider="anthropic",
+            duration_ms=812.5,
+            input_tokens=1200,
+            output_tokens=90,
+            attrs={"cache_read_input_tokens": 1000},
+        )
+        return "ok"
+
+    @app.tool
+    async def chat_fails(prompt: str) -> str:
+        err = RuntimeError(
+            "upstream 401: api_key=sk-abcdefghijklmnopqrstuv " + "x" * 800
+        )
+        holder["mw"].record_llm_call(model="gpt-x", provider="openai", error=err)
+        raise err
+
+    @app.tool
+    def poll() -> dict:
+        return {"status": "idle"}
+
+    @app.tool
+    def poll_with_event() -> dict:
+        holder["mw"].record_event("poll.saw_job", key="j9")
+        return {"status": "idle"}
+
+    mw = instrument(app, sinks if sinks is not None else [sink], **kwargs)
+    holder["mw"] = mw
+    return app, sink, mw, spawned
+
+
+async def call(app, mw, *calls):
+    async with Client(app) as client:
+        for tool, args in calls:
+            with contextlib.suppress(ToolError):
+                await client.call_tool(tool, args)
+    await mw.flush()
+
+
+async def table_rows(sink, table):
+    async with sink.engine.connect() as conn:
+        return [dict(r) for r in (await conn.execute(select(table))).mappings()]
+
+
+def event(kind="job.step", key="j1", minutes_ago=0, **kwargs):
+    return EventRecord(
+        kind=kind,
+        occurred_at=datetime.now(UTC) - timedelta(minutes=minutes_ago),
+        key=key,
+        **kwargs,
+    )
+
+
+class TestLinking:
+    async def test_event_in_tool_links_to_stored_call(self, db):
+        store = db()
+        app, _, mw, _ = make_app([store])
+        await call(app, mw, ("submit", {"job_id": "j1"}))
+        [call_row] = await table_rows(store, store.table)
+        [ev] = await table_rows(store, store.events)
+        assert ev["kind"] == "job.submitted"
+        assert ev["call_id"] == call_row["id"]
+        assert ev["session_id"] == call_row["session_id"]
+        assert ev["call_id"] == mw.recent[-1].id
+
+    async def test_sync_tool_links_too(self):
+        app, sink, mw, _ = make_app()
+        await call(app, mw, ("submit_sync", {"job_id": "j2"}))
+        await asyncio.sleep(0.01)
+        await mw.flush()
+        [ev] = sink.events
+        [rec] = sink.calls
+        assert ev.call_id == rec.id
+
+    async def test_event_outside_call_has_no_call_id(self):
+        app, sink, mw, _ = make_app()
+        assert mw.record_event("job.finished", key="j1") is True
+        await mw.flush()
+        [ev] = sink.events
+        assert ev.call_id is None and ev.session_id is None
+
+    async def test_task_outliving_call_is_not_linked(self):
+        app, sink, mw, spawned = make_app()
+        await call(app, mw, ("spawn", {"job_id": "j3"}))
+        await asyncio.gather(*spawned)
+        await mw.flush()
+        [ev] = sink.events
+        assert ev.kind == "job.started" and ev.call_id is None
+
+    async def test_explicit_call_id_wins(self):
+        app, sink, mw, _ = make_app()
+        mw.record_event("x.y", call_id="given")
+        await mw.flush()
+        assert sink.events[0].call_id == "given"
+
+    async def test_sampled_out_call_kept_when_it_recorded_event(self):
+        app, sink, mw, _ = make_app(
+            sample_rates={"poll_with_event": 0.01, "poll": 0.01},
+            random=lambda: 0.99,
+        )
+        await call(app, mw, ("poll", {}), ("poll_with_event", {}))
+        [rec] = sink.calls
+        [ev] = sink.events
+        assert rec.tool == "poll_with_event" and rec.sample_rate is None
+        assert ev.call_id == rec.id
+
+    async def test_excluded_tool_events_unlinked(self):
+        app, sink, mw, _ = make_app(exclude_tools={"poll_with_event"})
+        await call(app, mw, ("poll_with_event", {}))
+        assert sink.calls == []
+        [ev] = sink.events
+        assert ev.call_id is None
+
+
+class TestRecordEvent:
+    async def test_key_attrs_and_server_version_stored(self, db):
+        store = db()
+        app, _, mw, _ = make_app([store], server_version="1.2.3")
+        mw.record_event(
+            "step.finished",
+            key=42,
+            attrs={"step": "snappyHexMesh", "cells": 1200000, "mesh_ok": True},
+            user_sub="u1",
+        )
+        await mw.flush()
+        [ev] = await table_rows(store, store.events)
+        assert ev["key"] == "42"
+        assert ev["attrs"] == {
+            "step": "snappyHexMesh",
+            "cells": 1200000,
+            "mesh_ok": True,
+        }
+        assert ev["user_sub"] == "u1" and ev["server_version"] == "1.2.3"
+
+    async def test_attrs_redacted(self):
+        app, sink, mw, _ = make_app()
+        mw.record_event(
+            "job.failed",
+            attrs={
+                "token": "abc123",
+                "nested": {"db_password": "hunter2"},
+                "error_excerpt": "curl -H 'Authorization: Bearer abcdefghijklmnop'",
+            },
+        )
+        await mw.flush()
+        attrs = sink.events[0].attrs
+        assert attrs["token"] == REDACTED
+        assert attrs["nested"]["db_password"] == REDACTED
+        assert "abcdefghijklmnop" not in attrs["error_excerpt"]
+
+    async def test_attrs_made_json_safe(self, db):
+        store = db()
+        app, _, mw, _ = make_app([store])
+        when = datetime(2026, 9, 29, tzinfo=UTC)
+        mw.record_event(
+            "solver.progress",
+            key="j1",
+            attrs={"residuals": {"p": float("nan"), "U": 1e-5}, "at": when, 3: (1, 2)},
+        )
+        await mw.flush()
+        [ev] = await table_rows(store, store.events)
+        assert ev["attrs"] == {
+            "residuals": {"p": "nan", "U": 1e-5},
+            "at": when.isoformat(),
+            "3": [1, 2],
+        }
+
+    @pytest.mark.parametrize(
+        "kind",
+        ["", "   ", None, 5, "k" * 129],
+        ids=["empty", "blank", "none", "int", "long"],
+    )
+    async def test_invalid_kind_rejected(self, kind):
+        app, sink, mw, _ = make_app()
+        assert mw.record_event(kind) is False
+        await mw.flush()
+        assert sink.events == [] and list(mw.recent_events) == []
+
+    async def test_invalid_key_and_attrs_rejected(self):
+        app, sink, mw, _ = make_app()
+        assert mw.record_event("a.b", key="k" * 256) is False
+        assert mw.record_event("a.b", attrs=["not", "a", "dict"]) is False
+        await mw.flush()
+        assert sink.events == []
+
+    async def test_mode_off_is_noop(self):
+        app, sink, mw, _ = make_app(mode="off")
+        assert mw.record_event("job.submitted", key="j1") is False
+        assert mw.record_llm_call(model="m") is False
+        await call(app, mw, ("submit", {"job_id": "j1"}))
+        assert sink.records == [] and list(mw.recent_events) == []
+        assert mw.dispatcher.dropped == 0
+
+    async def test_queue_full_returns_false_and_counts(self):
+        app, sink, mw, _ = make_app(max_queue=1)
+        assert mw.record_event("a.one") is True
+        assert mw.record_event("a.two") is False
+        assert mw.dispatcher.dropped == 1
+        await mw.flush()
+        assert [e.kind for e in sink.events] == ["a.one"]
+        assert [e.kind for e in mw.recent_events] == ["a.one"]
+
+    async def test_broken_sink_never_raises(self):
+        def boom(records):
+            raise RuntimeError("sink down")
+
+        app, _, mw, _ = make_app([CallbackSink(boom)])
+        assert mw.record_event("a.b", key="j") is True
+        await mw.flush()
+        assert mw.dispatcher.sink_errors == 1
+
+    async def test_broken_redactor_never_raises(self):
+        app, sink, mw, _ = make_app()
+
+        class Broken:
+            def redact(self, value):
+                raise RuntimeError("bad pattern")
+
+        mw.redactor = Broken()
+        assert mw.record_event("a.b", attrs={"x": 1}) is False
+
+
+class TestThreads:
+    async def test_record_from_thread_while_server_runs(self, db):
+        store = db()
+        app, _, mw, _ = make_app([store])
+        results = []
+        async with Client(app) as client:
+            await client.call_tool("submit", {"job_id": "j1"})
+
+            def watcher():
+                for kind in ("step.started", "job.finished"):
+                    results.append(
+                        mw.record_event(kind, key="j1", attrs={"case": "wing"})
+                    )
+
+            t = threading.Thread(target=watcher)
+            t.start()
+            await asyncio.to_thread(t.join)
+            await asyncio.sleep(0)
+        await mw.flush()
+        assert results == [True, True]
+        rows = await table_rows(store, store.events)
+        assert sorted(r["kind"] for r in rows) == [
+            "job.finished",
+            "job.submitted",
+            "step.started",
+        ]
+        threaded = [r for r in rows if r["kind"] != "job.submitted"]
+        assert all(r["call_id"] is None for r in threaded)
+        assert [e.kind for e in mw.recent_events] == [
+            "job.submitted",
+            "step.started",
+            "job.finished",
+        ]
+
+    async def test_thread_with_its_own_loop_hands_over(self):
+        app, sink, mw, _ = make_app()
+        mw.record_event("a.first")  # binds the dispatcher to this loop
+        out = []
+        t = threading.Thread(
+            target=lambda: out.append(asyncio.run(self._record_in_loop(mw)))
+        )
+        t.start()
+        await asyncio.to_thread(t.join)
+        await asyncio.sleep(0)
+        await mw.flush()
+        assert out == [True]
+        assert [e.kind for e in sink.events] == ["a.first", "a.second"]
+        assert mw.dispatcher._loop is asyncio.get_running_loop()
+
+    @staticmethod
+    async def _record_in_loop(mw):
+        return mw.record_event("a.second")
+
+    def test_no_loop_and_nothing_recorded_drops(self):
+        app, sink, mw, _ = make_app()
+        assert mw.record_event("job.started", key="j1") is False
+        assert mw.dispatcher.dropped == 1
+        assert list(mw.recent_events) == []
+
+    async def test_thread_before_anything_recorded_drops(self):
+        app, sink, mw, _ = make_app()
+        out = []
+        t = threading.Thread(target=lambda: out.append(mw.record_event("a.b")))
+        t.start()
+        await asyncio.to_thread(t.join)
+        assert out == [False] and mw.dispatcher.dropped == 1
+
+
+class TestLlmCall:
+    async def test_attrs_and_linking(self, db):
+        store = db()
+        app, _, mw, _ = make_app([store])
+        await call(app, mw, ("chat", {"prompt": "hi"}))
+        [call_row] = await table_rows(store, store.table)
+        [ev] = await table_rows(store, store.events)
+        assert ev["kind"] == "llm.call" and ev["call_id"] == call_row["id"]
+        assert ev["attrs"] == {
+            "cache_read_input_tokens": 1000,
+            "model": "claude-opus-4",
+            "provider": "anthropic",
+            "duration_ms": 812.5,
+            "input_tokens": 1200,
+            "output_tokens": 90,
+            "ok": True,
+        }
+
+    async def test_error_redacted_truncated_and_not_ok(self):
+        app, sink, mw, _ = make_app(max_error_chars=100)
+        await call(app, mw, ("chat_fails", {"prompt": "hi"}))
+        [ev] = sink.events
+        attrs = ev.attrs
+        assert attrs["ok"] is False and attrs["error_type"] == "RuntimeError"
+        assert "sk-abcdefghijklmnopqrstuv" not in attrs["error"]
+        assert len(attrs["error"]) == 100
+        assert ev.call_id == sink.calls[0].id and sink.calls[0].outcome == "error"
+
+    async def test_named_fields_win_and_ok_explicit(self):
+        app, sink, mw, _ = make_app()
+        mw.record_llm_call(
+            model="m", ok=False, key="turn-1", attrs={"model": "spoofed", "round": 2}
+        )
+        await mw.flush()
+        [ev] = sink.events
+        assert ev.key == "turn-1"
+        assert ev.attrs == {"model": "m", "round": 2, "ok": False}
+
+
+class TestSinks:
+    async def test_database_sink_mixed_batch(self, db):
+        store = db()
+        rec = ToolCallRecord(
+            tool="t",
+            started_at=datetime.now(UTC),
+            duration_ms=1.0,
+            ok=True,
+            mode="meta",
+        )
+        ev = event(call_id=rec.id, attrs={"a": 1})
+        await store.write([rec, ev])
+        assert [r["id"] for r in await table_rows(store, store.table)] == [rec.id]
+        [row] = await table_rows(store, store.events)
+        assert row["id"] == ev.id and row["call_id"] == rec.id
+
+    async def test_hand_built_event_attrs_made_json_safe(self, db):
+        store = db()
+        await store.write([event(attrs={"p": float("nan"), "q": float("-inf")})])
+        [row] = await table_rows(store, store.events)
+        assert row["attrs"] == {"p": "nan", "q": "-inf"}
+
+    async def test_database_sink_events_only_and_calls_only(self, db):
+        store = db()
+        await store.write([event()])
+        assert len(await table_rows(store, store.events)) == 1
+        assert await table_rows(store, store.table) == []
+
+    async def test_calls_stored_when_events_table_missing(self, db):
+        store = db("old", create_tables=False)
+        old = MetaData()
+        build_metadata("ffb_", old)
+        old.remove(old.tables["ffb_events"])
+        async with store.engine.begin() as conn:
+            await conn.run_sync(old.create_all)
+        rec = ToolCallRecord(
+            tool="t",
+            started_at=datetime.now(UTC),
+            duration_ms=1.0,
+            ok=True,
+            mode="meta",
+        )
+        # SQLite raises OperationalError, Postgres ProgrammingError.
+        with pytest.raises(DBAPIError, match="no such table|does not exist"):
+            await store.write([rec, event()])
+        assert len(await table_rows(store, store.table)) == 1
+
+    async def test_json_lines_record_field(self):
+        stream = io.StringIO()
+        app, _, mw, _ = make_app([JsonLinesSink(stream)])
+        await call(app, mw, ("submit", {"job_id": "j1"}))
+        lines = [json.loads(line) for line in stream.getvalue().splitlines()]
+        by_type = {line["record"]: line for line in lines}
+        assert set(by_type) == {"tool_call", "event"}
+        assert by_type["event"]["kind"] == "job.submitted"
+        assert by_type["event"]["call_id"] == by_type["tool_call"]["id"]
+        assert by_type["tool_call"]["tool"] == "submit"
+
+    async def test_memory_sink_calls_and_events(self):
+        app, sink, mw, _ = make_app()
+        await call(app, mw, ("submit", {"job_id": "j1"}), ("poll", {}))
+        assert [r.tool for r in sink.calls] == ["submit", "poll"]
+        assert [e.kind for e in sink.events] == ["job.submitted"]
+        assert len(sink.records) == 3
+
+
+class TestEventsFor:
+    async def test_merges_database_and_memory_in_time_order(self, db):
+        store = db()
+        app, _, mw, _ = make_app([store])
+        # Written by another worker: only in the database.
+        await store.write([event("job.submitted", minutes_ago=30)])
+        # Recorded here and flushed: in both, returned once.
+        mw.record_event("step.started", key="j1")
+        await mw.flush()
+        # Recorded here, not yet written: only in memory.
+        pending = event("step.finished", minutes_ago=-1)
+        mw.recent_events.append(pending)
+        # Another job, and a kind filtered out below.
+        mw.record_event("job.submitted", key="j2")
+        await mw.flush()
+
+        rows = await mw.events_for("j1")
+        assert [r["kind"] for r in rows] == [
+            "job.submitted",
+            "step.started",
+            "step.finished",
+        ]
+        assert all(r["occurred_at"].tzinfo is not None for r in rows)
+        assert rows[-1]["id"] == pending.id
+        assert len({r["id"] for r in rows}) == 3
+
+        steps = await mw.events_for("j1", kinds=["step.started", "step.finished"])
+        assert [r["kind"] for r in steps] == ["step.started", "step.finished"]
+        last = await mw.events_for("j1", limit=2)
+        assert [r["kind"] for r in last] == ["step.started", "step.finished"]
+
+    async def test_memory_only_without_database(self):
+        app, sink, mw, _ = make_app()
+        await call(app, mw, ("submit", {"job_id": "j1"}))
+        mw.record_event("job.finished", key="j1")
+        rows = await mw.events_for("j1")
+        assert [r["kind"] for r in rows] == ["job.submitted", "job.finished"]
+        assert rows[0]["call_id"] == sink.calls[0].id
+
+    async def test_database_failure_falls_back_to_memory(self, db):
+        store = db("nodb", create_tables=False)
+        app, _, mw, _ = make_app([store])
+        mw.recent_events.append(event("job.submitted"))
+        rows = await mw.events_for("j1")
+        assert [r["kind"] for r in rows] == ["job.submitted"]
+
+
+class TestPrune:
+    async def test_prune_deletes_old_events_in_batches(self, db):
+        store = db(retention=timedelta(days=30), prune_batch=2)
+        old_call = ToolCallRecord(
+            tool="old",
+            started_at=datetime.now(UTC) - timedelta(days=40),
+            duration_ms=1.0,
+            ok=True,
+            mode="meta",
+        )
+        store._last_prune = float("inf")  # keep write() from pruning first
+        await store.write(
+            [old_call]
+            + [event(f"old.{i}", minutes_ago=60 * 24 * 40) for i in range(5)]
+            + [event("new.one"), event("new.two", minutes_ago=60 * 24 * 29)]
+        )
+        statements = []
+        from sqlalchemy import event as sa_event
+
+        def count_deletes(conn, cursor, statement, *args):
+            if statement.lstrip().upper().startswith("DELETE FROM FFB_EVENTS"):
+                statements.append(statement)
+
+        sa_event.listen(
+            store.engine.sync_engine, "before_cursor_execute", count_deletes
+        )
+        deleted = await store.prune()
+        assert deleted == 6  # 1 call + 5 events
+        assert len(statements) == 3  # 2 + 2 + 1
+        kinds = sorted(r["kind"] for r in await table_rows(store, store.events))
+        assert kinds == ["new.one", "new.two"]
+
+    async def test_auto_prune_covers_events(self, db):
+        store = db(retention=timedelta(days=30))
+        await store.write([event("old", minutes_ago=60 * 24 * 31), event("new")])
+        assert [r["kind"] for r in await table_rows(store, store.events)] == ["new"]
+
+    async def test_prune_without_events_table(self, db):
+        store = db("legacy", create_tables=False)
+        old = MetaData()
+        build_metadata("ffb_", old)
+        old.remove(old.tables["ffb_events"])
+        async with store.engine.begin() as conn:
+            await conn.run_sync(old.create_all)
+        rec = ToolCallRecord(
+            tool="old",
+            started_at=datetime.now(UTC) - timedelta(days=40),
+            duration_ms=1.0,
+            ok=True,
+            mode="meta",
+        )
+        await store.write([rec])
+        assert await store.prune(datetime.now(UTC) - timedelta(days=30)) == 1
+        async with store.engine.connect() as conn:
+            n = (
+                await conn.execute(select(func.count()).select_from(store.table))
+            ).scalar()
+        assert n == 0

tests/unit/test_instrumentation.py

@@ -373,9 +373,8 @@ class TestSinks:
 
 
 class TestDatabaseSink:
-    async def test_end_to_end_with_created_tables(self, tmp_path):
-        url = f"sqlite+aiosqlite:///{tmp_path}/calls.db"
-        db = DatabaseSink(url, create_tables=True)
+    async def test_end_to_end_with_created_tables(self, db_url):
+        db = DatabaseSink(db_url(), create_tables=True)
         app, _, mw = make_app(sinks=[db], mode="full", server_version="9.9")
         await call(app, "create_token", {"label": "ci"})
         with pytest.raises(ToolError):
@@ -397,8 +396,8 @@ class TestDatabaseSink:
         assert rows[0]["server_version"] == "9.9"
         assert "hunter2" not in json.dumps([dict(r) for r in rows], default=str)
 
-    async def test_prefix_and_host_owned_engine(self, tmp_path):
-        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
+    async def test_prefix_and_host_owned_engine(self, db_url):
+        engine = create_async_engine(db_url("host"))
         db = DatabaseSink(engine, prefix="qa_", create_tables=True)
         app, _, mw = make_app(sinks=[db])
         await call(app, "echo", {"text": "hi"})
@@ -408,10 +407,14 @@ class TestDatabaseSink:
         async with engine.connect() as conn:
             tables = await conn.run_sync(lambda c: inspect(c).get_table_names())
         await engine.dispose()
-        assert sorted(tables) == ["qa_feedback_call_links", "qa_tool_calls"]
+        assert sorted(tables) == [
+            "qa_events",
+            "qa_feedback_call_links",
+            "qa_tool_calls",
+        ]
 
-    async def test_no_tables_created_by_default(self, tmp_path):
-        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
+    async def test_no_tables_created_by_default(self, db_url):
+        engine = create_async_engine(db_url("host"))
         app, _, mw = make_app(sinks=[DatabaseSink(engine)])
         result = await call(app, "echo", {"text": "hi"})
         await mw.flush()
@@ -425,9 +428,9 @@ class TestDatabaseSink:
         assert tables == []
         assert mw.dispatcher.sink_errors == 1
 
-    async def test_absent_payloads_are_sql_null(self, tmp_path):
+    async def test_absent_payloads_are_sql_null(self, db_url):
         """Meta mode stores no payloads; they must be SQL NULL, not JSON null."""
-        db = DatabaseSink(f"sqlite+aiosqlite:///{tmp_path}/n.db", create_tables=True)
+        db = DatabaseSink(db_url("n"), create_tables=True)
         app, _, mw = make_app(sinks=[db])
         await call(app, "echo", {"text": "hi"})
         await mw.flush()
@@ -444,7 +447,11 @@ class TestDatabaseSink:
         host = MetaData()
         assert build_metadata("ffb_", host) is host
         build_metadata("ffb_", host)
-        assert sorted(host.tables) == ["ffb_feedback_call_links", "ffb_tool_calls"]
+        assert sorted(host.tables) == [
+            "ffb_events",
+            "ffb_feedback_call_links",
+            "ffb_tool_calls",
+        ]
 
 
 class TestIntegration:

tests/unit/test_retention_sampling.py

@@ -0,0 +1,400 @@
+"""Tests for per-tool exclusion, sampling and retention pruning.
+
+Real FastMCP servers through fastmcp.Client, as in test_instrumentation.py.
+"""
+
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from fastmcp import Client, FastMCP
+from fastmcp.exceptions import ToolError
+from sqlalchemy import event, func, select
+
+from fastmcp_feedback.instrumentation import (
+    DatabaseSink,
+    MemorySink,
+    ToolCallRecord,
+    instrument,
+)
+
+
+@pytest.fixture
+async def db(db_url):
+    """DatabaseSinks whose engines are always disposed, or the process hangs."""
+    made = []
+
+    def factory(name="calls", **kwargs):
+        sink = DatabaseSink(db_url(name), create_tables=True, **kwargs)
+        made.append(sink)
+        return sink
+
+    yield factory
+    for sink in made:
+        await sink.engine.dispose()
+
+
+class Rolls:
+    """Stand-in for random.random that returns the given values in turn."""
+
+    def __init__(self, *values):
+        self.values = list(values)
+        self.calls = 0
+
+    def __call__(self):
+        self.calls += 1
+        return self.values.pop(0)
+
+
+def make_app(sinks=None, **kwargs):
+    sink = MemorySink()
+    app = FastMCP("sampling")
+
+    @app.tool
+    def poll() -> dict:
+        return {"status": "idle"}
+
+    @app.tool
+    def render(status: str) -> dict:
+        return {"status": status, "error": f"render said {status}"}
+
+    @app.tool
+    def crash() -> dict:
+        raise RuntimeError("exploded")
+
+    @app.tool
+    def other() -> str:
+        return "fine"
+
+    mw = instrument(app, sinks if sinks is not None else [sink], **kwargs)
+    return app, sink, mw
+
+
+async def call(app, mw, *calls):
+    """Run (tool, args) pairs; return the results, None where the tool raised."""
+    results = []
+    async with Client(app) as client:
+        for tool, args in calls:
+            try:
+                results.append(await client.call_tool(tool, args))
+            except ToolError:
+                results.append(None)
+    await mw.flush()
+    return results
+
+
+def old_record(tool="old", days=40, **kwargs):
+    return ToolCallRecord(
+        tool=tool,
+        started_at=datetime.now(UTC) - timedelta(days=days),
+        duration_ms=1.0,
+        ok=True,
+        mode="meta",
+        **kwargs,
+    )
+
+
+async def rows(sink, *columns):
+    t = sink.table
+    cols = [t.c[c] for c in columns] if columns else [t.c.tool]
+    async with sink.engine.connect() as conn:
+        return [
+            tuple(r)
+            for r in (await conn.execute(select(*cols).order_by(t.c.started_at))).all()
+        ]
+
+
+async def count(sink):
+    async with sink.engine.connect() as conn:
+        return (
+            await conn.execute(select(func.count()).select_from(sink.table))
+        ).scalar()
+
+
+class TestExclusion:
+    async def test_excluded_tool_not_recorded_nor_kept(self):
+        app, sink, mw = make_app(exclude_tools={"poll"})
+        results = await call(app, mw, ("poll", {}), ("other", {}), ("poll", {}))
+        assert [r.tool for r in sink.records] == ["other"]
+        assert [r.tool for r in mw.recent] == ["other"]
+        # The call itself is untouched.
+        assert results[0].structured_content == {"status": "idle"}
+
+    async def test_excluded_tool_failures_not_recorded_either(self):
+        app, sink, mw = make_app(exclude_tools=["crash"])
+        [result] = await call(app, mw, ("crash", {}))
+        assert result is None  # the error still reached the client
+        assert sink.records == []
+        assert list(mw.recent) == []
+
+    async def test_excluded_tool_skips_hooks(self):
+        seen = []
+        app, sink, mw = make_app(
+            exclude_tools={"poll"}, identity_resolver=lambda c: seen.append(1) or {}
+        )
+        await call(app, mw, ("poll", {}))
+        assert seen == []
+
+
+class TestSampling:
+    async def test_recorded_when_random_below_rate(self):
+        rolls = Rolls(0.29, 0.3, 0.31, 0.0)
+        app, sink, mw = make_app(sample_rates={"poll": 0.3}, random=rolls)
+        await call(app, mw, *[("poll", {})] * 4)
+        assert rolls.calls == 4
+        assert len(sink.records) == 2
+        assert [r.sample_rate for r in sink.records] == [0.3, 0.3]
+        assert all(r.outcome == "ok" for r in sink.records)
+
+    async def test_sampled_out_calls_not_in_recent(self):
+        app, sink, mw = make_app(sample_rates={"poll": 0.5}, random=Rolls(0.9, 0.1))
+        results = await call(app, mw, ("poll", {}), ("poll", {}))
+        assert len(sink.records) == 1
+        assert [r.id for r in mw.recent] == [sink.records[0].id]
+        # A sampled-out call still returns its result.
+        assert results[0].structured_content == {"status": "idle"}
+
+    async def test_sampled_out_ok_calls_skip_hooks(self):
+        seen = []
+
+        def who(context):
+            seen.append("identity")
+            return {}
+
+        def enrich(tool, args, result, context):
+            seen.append("enrich")
+            return {}
+
+        app, sink, mw = make_app(
+            sample_rates={"poll": 0.1},
+            random=Rolls(0.99),
+            identity_resolver=who,
+            enricher=enrich,
+        )
+        await call(app, mw, ("poll", {}))
+        assert sink.records == []
+        assert seen == []
+
+    async def test_unsampled_tools_recorded_without_rate_or_roll(self):
+        rolls = Rolls()
+        app, sink, mw = make_app(sample_rates={"poll": 0.1}, random=rolls)
+        await call(app, mw, ("other", {}), ("other", {}))
+        assert rolls.calls == 0
+        assert [(r.tool, r.sample_rate) for r in sink.records] == [
+            ("other", None),
+            ("other", None),
+        ]
+
+    async def test_rate_of_one_records_everything_with_rate(self):
+        app, sink, mw = make_app(sample_rates={"poll": 1}, random=Rolls(0.999))
+        await call(app, mw, ("poll", {}))
+        [record] = sink.records
+        assert record.sample_rate == 1.0
+
+    async def test_exception_of_sampled_tool_always_recorded(self):
+        # Rolls would sample every call out; failures must not consult them.
+        rolls = Rolls(0.99, 0.99)
+        app, sink, mw = make_app(sample_rates={"crash": 0.01}, random=rolls)
+        await call(app, mw, ("crash", {}), ("crash", {}))
+        assert [(r.outcome, r.error_type, r.sample_rate) for r in sink.records] == [
+            ("error", "RuntimeError", None),
+            ("error", "RuntimeError", None),
+        ]
+        assert len(mw.recent) == 2
+
+    async def test_soft_error_of_sampled_tool_always_recorded(self):
+        rolls = Rolls(0.99, 0.99)
+        app, sink, mw = make_app(sample_rates={"render": 0.01}, random=rolls)
+        await call(
+            app, mw, ("render", {"status": "failed"}), ("render", {"status": "done"})
+        )
+        assert [(r.outcome, r.sample_rate) for r in sink.records] == [
+            ("soft_error", None)
+        ]
+        assert sink.records[0].error_message == "status=failed: render said failed"
+        assert [r.id for r in mw.recent] == [sink.records[0].id]
+
+    async def test_sampled_in_failure_has_no_rate(self):
+        app, sink, mw = make_app(sample_rates={"crash": 0.5}, random=Rolls(0.0))
+        await call(app, mw, ("crash", {}))
+        [record] = sink.records
+        assert record.sample_rate is None
+
+    @pytest.mark.parametrize("rate", [0, -0.1, 1.01, 2, float("nan")])
+    def test_invalid_rates_rejected(self, rate):
+        with pytest.raises(ValueError, match="poll"):
+            make_app(sample_rates={"poll": rate})
+
+    def test_non_numeric_rate_rejected(self):
+        with pytest.raises((TypeError, ValueError)):
+            make_app(sample_rates={"poll": "often"})
+
+    async def test_estimated_count(self):
+        """sum(1 / coalesce(sample_rate, 1)) estimates the true call count."""
+        rolls = Rolls(*[i / 10 for i in range(10)])
+        app, sink, mw = make_app(sample_rates={"poll": 0.2}, random=rolls)
+        await call(app, mw, *[("poll", {})] * 10, ("crash", {}))
+        estimate = sum(1 / (r.sample_rate or 1) for r in sink.records)
+        assert estimate == pytest.approx(11)
+
+
+class TestDatabaseSinkColumn:
+    async def test_writes_sample_rate(self, db):
+        sink = db()
+        app, _, mw = make_app(
+            sinks=[sink],
+            sample_rates={"poll": 0.25, "crash": 0.25},
+            random=Rolls(0.1, 0.9),
+        )
+        await call(app, mw, ("poll", {}), ("poll", {}), ("crash", {}), ("other", {}))
+        assert await rows(sink, "tool", "outcome", "sample_rate") == [
+            ("poll", "ok", 0.25),
+            ("crash", "error", None),
+            ("other", "ok", None),
+        ]
+
+
+class TestPrune:
+    async def test_deletes_old_rows_keeps_new(self, db):
+        sink = db()
+        await sink.write([old_record("a", days=40), old_record("b", days=31)])
+        await sink.write([old_record("c", days=29), old_record("d", days=0)])
+        deleted = await sink.prune(datetime.now(UTC) - timedelta(days=30))
+        assert deleted == 2
+        assert await rows(sink) == [("c",), ("d",)]
+
+    async def test_default_cutoff_uses_retention(self, db):
+        sink = db(retention=timedelta(days=30))
+        sink._last_prune = time.monotonic()  # keep write() from pruning
+        await sink.write([old_record("a", days=40), old_record("b", days=1)])
+        assert await count(sink) == 2
+        assert await sink.prune() == 1
+        assert await rows(sink) == [("b",)]
+
+    async def test_no_retention_no_cutoff_is_a_no_op(self, db):
+        sink = db()
+        await sink.write([old_record(days=400)])
+        assert await sink.prune() == 0
+        assert await count(sink) == 1
+
+    async def test_keeps_linked_calls_and_links(self, db):
+        sink = db()
+        linked, plain = old_record("linked"), old_record("plain")
+        await sink.write([linked, plain])
+        await sink.replace_links(
+            "bug-1",
+            [
+                {
+                    "feedback_ref": "bug-1",
+                    "call_id": linked.id,
+                    "position": 0,
+                    "rule": "session",
+                    "linked_at": datetime.now(UTC),
+                },
+                {
+                    # A link to a call that is not in this database.
+                    "feedback_ref": "bug-1",
+                    "call_id": "missing",
+                    "position": 1,
+                    "rule": "session",
+                    "linked_at": datetime.now(UTC),
+                },
+            ],
+        )
+        assert await sink.prune(datetime.now(UTC)) == 1
+        assert await rows(sink) == [("linked",)]
+        context = await sink.linked_calls("bug-1")
+        assert [(r["call_id"], r["tool"]) for r in context] == [
+            (linked.id, "linked"),
+            ("missing", None),
+        ]
+
+    async def test_deletes_in_batches(self, db):
+        sink = db(prune_batch=3)
+        await sink.write(
+            [old_record(f"t{i}") for i in range(10)] + [old_record("new", 0)]
+        )
+        deletes = []
+
+        def spy(conn, cursor, statement, params, context, executemany):
+            if statement.lstrip().upper().startswith("DELETE"):
+                deletes.append(statement)
+
+        event.listen(sink.engine.sync_engine, "before_cursor_execute", spy)
+        try:
+            assert await sink.prune(datetime.now(UTC) - timedelta(days=1)) == 10
+        finally:
+            event.remove(sink.engine.sync_engine, "before_cursor_execute", spy)
+        assert len(deletes) == 4  # 3 + 3 + 3 + 1
+        assert await rows(sink) == [("new",)]
+
+    async def test_exact_multiple_of_batch(self, db):
+        sink = db(prune_batch=5)
+        await sink.write([old_record(f"t{i}") for i in range(10)])
+        assert await sink.prune(datetime.now(UTC)) == 10
+        assert await count(sink) == 0
+
+    def test_invalid_settings_rejected(self):
+        with pytest.raises(ValueError):
+            DatabaseSink("sqlite+aiosqlite://", retention=timedelta(0))
+        with pytest.raises(ValueError):
+            DatabaseSink("sqlite+aiosqlite://", prune_batch=0)
+
+
+class TestAutomaticPrune:
+    async def test_first_write_prunes_then_waits_for_interval(self, db):
+        sink = db(retention=timedelta(days=30), prune_interval=timedelta(hours=1))
+        await sink.write([old_record("a"), old_record("new", 0)])
+        assert await rows(sink) == [("new",)]
+
+        await sink.write([old_record("b")])
+        assert await count(sink) == 2  # inside the interval: not pruned
+
+        sink._last_prune = time.monotonic() - 3601
+        await sink.write([old_record("c", 0)])
+        assert sorted(await rows(sink)) == [("c",), ("new",)]
+
+    async def test_no_retention_never_prunes(self, db):
+        sink = db()
+        await sink.write([old_record(days=4000)])
+        assert await count(sink) == 1
+        assert sink._last_prune is None
+
+    async def test_failing_prune_does_not_stop_inserts(self, db, caplog):
+        sink = db(retention=timedelta(days=30))
+        attempts = []
+
+        async def broken(older_than=None):
+            attempts.append(older_than)
+            raise RuntimeError("prune blew up")
+
+        sink.prune = broken
+        with caplog.at_level(logging.WARNING):
+            await sink.write([old_record("a", 0)])
+            await sink.write([old_record("b", 0)])
+        assert await rows(sink) == [("a",), ("b",)]
+        assert len(attempts) == 1  # not retried on every write
+        assert "prune blew up" in caplog.text
+
+    async def test_through_the_middleware(self, db):
+        sink = db(retention=timedelta(days=30))
+        sink._last_prune = time.monotonic()
+        await sink.write([old_record("stale")])
+        sink._last_prune = None
+        app, _, mw = make_app(sinks=[sink])
+        await call(app, mw, ("other", {}))
+        assert mw.dispatcher.sink_errors == 0
+        assert await rows(sink) == [("other",)]
+
+    async def test_prune_failure_through_middleware_keeps_records(self, db):
+        sink = db(retention=timedelta(days=30))
+
+        async def broken(older_than=None):
+            raise RuntimeError("no")
+
+        sink.prune = broken
+        app, _, mw = make_app(sinks=[sink])
+        await call(app, mw, ("other", {}), ("poll", {}))
+        assert mw.dispatcher.sink_errors == 0
+        assert sorted(await rows(sink)) == [("other",), ("poll",)]

tests/unit/test_soft_errors.py

@@ -0,0 +1,395 @@
+"""Tests for classifying error results that tools return instead of raising.
+
+Real FastMCP servers through fastmcp.Client, as in test_instrumentation.py.
+"""
+
+import contextlib
+import logging
+
+import pytest
+from fastmcp import Client, FastMCP
+from fastmcp.exceptions import ToolError
+from mcp.types import CallToolResult, TextContent
+from sqlalchemy import select
+
+from fastmcp_feedback import add_feedback_tools
+from fastmcp_feedback.instrumentation import (
+    DEFAULT_RESULT_CLASSIFIER,
+    REDACTED,
+    DatabaseSink,
+    MemorySink,
+    ToolCallRecord,
+    instrument,
+    soft_error_classifier,
+)
+from fastmcp_feedback.mixins import submission
+
+BMCP_TOKEN = "bmcp_" + "A1b2C3d4E5f6G7h8I9j0"
+
+try:
+    from fastmcp.tools.tool import ToolResult
+except ImportError:  # FastMCP 4
+    from fastmcp.tools import ToolResult
+
+try:
+    ToolResult(content=[TextContent(type="text", text="x")], is_error=True)
+    TOOL_RESULT_IS_ERROR = True
+except TypeError:  # added in FastMCP 3.4
+    TOOL_RESULT_IS_ERROR = False
+
+
+@pytest.fixture
+async def db(db_url):
+    """DatabaseSinks whose engines are always disposed, or the process hangs."""
+    made = []
+
+    def factory(name="calls"):
+        sink = DatabaseSink(db_url(name), create_tables=True)
+        made.append(sink)
+        return sink
+
+    yield factory
+    for sink in made:
+        await sink.engine.dispose()
+
+
+def make_app(sinks=None, **kwargs):
+    sink = MemorySink()
+    app = FastMCP("soft")
+
+    @app.tool
+    def render(status: str) -> dict:
+        return {"status": status, "error": f"render said {status}"}
+
+    @app.tool
+    def respond(payload: dict) -> dict:
+        return payload
+
+    @app.tool
+    def text(value: str) -> str:
+        return value
+
+    @app.tool
+    def items() -> list:
+        return [{"status": "failed"}]
+
+    @app.tool
+    def crash() -> dict:
+        raise RuntimeError("exploded")
+
+    @app.tool
+    def mcp_error() -> CallToolResult:
+        return CallToolResult(
+            content=[TextContent(type="text", text="target missing")], isError=True
+        )
+
+    mw = instrument(app, sinks if sinks is not None else [sink], **kwargs)
+    return app, sink, mw
+
+
+async def run(app, mw, tool, args=None):
+    async with Client(app) as client:
+        with contextlib.suppress(ToolError):
+            await client.call_tool(tool, args or {})
+    await mw.flush()
+
+
+async def record_for(tool, args=None, **kwargs) -> ToolCallRecord:
+    app, sink, mw = make_app(**kwargs)
+    await run(app, mw, tool, args)
+    [record] = sink.records
+    return record
+
+
+class TestClassifier:
+    @pytest.mark.parametrize(
+        "result, expected",
+        [
+            ({"status": "failed", "error": "disk full"}, "status=failed: disk full"),
+            ({"state": "ERROR"}, "state=ERROR"),
+            ({"ok": False, "message": "no scene"}, "ok=False: no scene"),
+            ({"success": False, "error": "nope"}, "success=False: nope"),
+            ({"success": False, "detail": "d", "reason": "r"}, "success=False: d"),
+            (
+                {"status": "failure", "error": {"code": 7, "message": "inner"}},
+                "status=failure: inner",
+            ),
+            ({"isError": True, "content": ["boom"]}, "isError: boom"),
+            (
+                {"is_error": True, "content": [{"type": "text", "text": "a"}]},
+                "isError: a",
+            ),
+        ],
+    )
+    def test_flagged(self, result, expected):
+        assert DEFAULT_RESULT_CLASSIFIER("t", result) == expected
+
+    @pytest.mark.parametrize(
+        "result",
+        [
+            {"status": "done"},
+            {"status": "unknown_target"},
+            {"ok": True},
+            {"ok": 0},
+            {"success": None},
+            {"result": False},
+            {"state": 3},
+            {"isError": False},
+            [{"status": "failed"}],
+            "failed",
+            None,
+        ],
+    )
+    def test_not_flagged(self, result):
+        assert DEFAULT_RESULT_CLASSIFIER("t", result) is None
+
+    def test_extra_statuses(self):
+        classify = soft_error_classifier(extra_statuses=("unknown_target",))
+        assert (
+            classify("t", {"status": "unknown_target", "error": "no cube"})
+            == "status=unknown_target: no cube"
+        )
+        assert classify("t", {"status": "failed"}) == "status=failed"
+
+    def test_custom_keys_replace_defaults(self):
+        classify = soft_error_classifier(
+            statuses=("bad",), status_keys=("phase",), flag_keys=("passed",)
+        )
+        assert classify("t", {"phase": "bad"}) == "phase=bad"
+        assert classify("t", {"passed": False}) == "passed=False"
+        assert classify("t", {"status": "failed", "ok": False}) is None
+
+
+class TestMiddleware:
+    async def test_status_failed_is_soft_error(self):
+        record = await record_for("render", {"status": "failed"})
+        assert (record.outcome, record.ok, record.error_type) == (
+            "soft_error",
+            False,
+            "SoftError",
+        )
+        assert record.error_message == "status=failed: render said failed"
+
+    @pytest.mark.parametrize("key", ["ok", "success"])
+    async def test_false_flags_are_soft_errors(self, key):
+        record = await record_for("respond", {"payload": {key: False, "error": "x"}})
+        assert (record.outcome, record.ok) == ("soft_error", False)
+        assert record.error_message == f"{key}=False: x"
+
+    async def test_message_is_redacted_and_truncated(self):
+        record = await record_for(
+            "respond",
+            {"payload": {"status": "error", "error": f"bad key {BMCP_TOKEN} " * 50}},
+            max_error_chars=60,
+        )
+        assert BMCP_TOKEN not in record.error_message
+        assert REDACTED in record.error_message
+        assert len(record.error_message) == 60
+
+    async def test_extra_statuses(self):
+        record = await record_for(
+            "render",
+            {"status": "unknown_target"},
+            result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)),
+        )
+        assert record.outcome == "soft_error"
+        assert (
+            record.error_message == "status=unknown_target: render said unknown_target"
+        )
+
+    async def test_unknown_status_is_ok_by_default(self):
+        record = await record_for("render", {"status": "unknown_target"})
+        assert (record.outcome, record.ok, record.error_type) == ("ok", True, None)
+
+    @pytest.mark.parametrize(
+        "tool, args",
+        [
+            ("text", {"value": "failed"}),
+            ("items", {}),
+            ("respond", {"payload": {"status": "done", "count": 3}}),
+        ],
+    )
+    async def test_normal_results_stay_ok(self, tool, args):
+        record = await record_for(tool, args)
+        assert (record.outcome, record.ok, record.error_message) == ("ok", True, None)
+
+    async def test_none_disables(self):
+        record = await record_for(
+            "render", {"status": "failed"}, result_classifier=None
+        )
+        assert (record.outcome, record.ok, record.error_type) == ("ok", True, None)
+
+    async def test_custom_classifier(self):
+        seen = []
+
+        def classify(tool, result):
+            seen.append((tool, result))
+            return "empty" if result.get("count") == 0 else None
+
+        record = await record_for(
+            "respond", {"payload": {"count": 0}}, result_classifier=classify
+        )
+        assert seen == [("respond", {"count": 0})]
+        assert (record.outcome, record.error_message) == ("soft_error", "empty")
+
+    async def test_raising_classifier_records_ok(self, caplog):
+        def classify(tool, result):
+            raise KeyError("classifier bug")
+
+        app, sink, mw = make_app(result_classifier=classify)
+        with caplog.at_level(logging.WARNING):
+            async with Client(app) as client:
+                result = await client.call_tool("render", {"status": "failed"})
+            await mw.flush()
+        assert result.data["status"] == "failed"
+        [record] = sink.records
+        assert (record.outcome, record.ok, record.error_type) == ("ok", True, None)
+        assert "Result classifier" in caplog.text
+
+    async def test_classifier_not_run_on_exceptions(self):
+        calls = []
+        record = await record_for(
+            "crash", result_classifier=lambda t, r: calls.append(t) or "x"
+        )
+        assert calls == []
+        assert (record.outcome, record.ok, record.error_type) == (
+            "error",
+            False,
+            "RuntimeError",
+        )
+        assert record.error_message == "exploded"
+
+    async def test_mcp_error_result_without_raising(self):
+        """A CallToolResult with isError, on every FastMCP version."""
+        record = await record_for("mcp_error")
+        assert (record.outcome, record.ok) == ("soft_error", False)
+        assert record.error_message == "isError: target missing"
+
+    @pytest.mark.skipif(not TOOL_RESULT_IS_ERROR, reason="ToolResult.is_error is 3.4+")
+    async def test_tool_result_is_error(self):
+        app = FastMCP("is_error")
+
+        @app.tool
+        def lookup():
+            return ToolResult(
+                content=[TextContent(type="text", text="not found")], is_error=True
+            )
+
+        sink = MemorySink()
+        mw = instrument(app, [sink])
+        async with Client(app) as client:
+            await client.call_tool("lookup", {}, raise_on_error=False)
+        await mw.flush()
+        [record] = sink.records
+        assert (record.outcome, record.error_message) == (
+            "soft_error",
+            "isError: not found",
+        )
+
+    def test_record_outcome_defaults_from_ok(self):
+        from datetime import UTC, datetime
+
+        now = datetime.now(UTC)
+        assert ToolCallRecord("t", now, 1.0, True, "meta").outcome == "ok"
+        assert ToolCallRecord("t", now, 1.0, False, "meta").outcome == "error"
+
+
+class TestStorageAndCorrelation:
+    async def test_database_sink_writes_outcome(self, db):
+        sink = db()
+        app, _, mw = make_app(sinks=[sink])
+        async with Client(app) as client:
+            await client.call_tool("render", {"status": "done"})
+            await client.call_tool("render", {"status": "failed"})
+            with contextlib.suppress(ToolError):
+                await client.call_tool("crash", {})
+        await mw.flush()
+
+        t = sink.table
+        async with sink.engine.connect() as conn:
+            rows = (
+                await conn.execute(
+                    select(t.c.outcome, t.c.ok, t.c.error_type).order_by(t.c.started_at)
+                )
+            ).all()
+        assert [tuple(r) for r in rows] == [
+            ("ok", True, None),
+            ("soft_error", False, "SoftError"),
+            ("error", False, "RuntimeError"),
+        ]
+        async with sink.engine.connect() as conn:
+            [(failing,)] = (
+                await conn.execute(
+                    select(t.c.tool).where(~t.c.ok, t.c.outcome != "error")
+                )
+            ).all()
+        assert failing == "render"
+
+    async def test_links_and_feedback_context_show_soft_errors(self, db):
+        app, _, mw = make_app(
+            sinks=[db()], identity_resolver=lambda c: {"user_sub": "u1"}
+        )
+        captured = {}
+
+        @app.tool
+        async def report() -> int:
+            captured["links"] = await mw.capture_recent_calls()
+            await mw.link_feedback("bug-1", captured["links"])
+            return len(captured["links"])
+
+        async with Client(app) as client:
+            await client.call_tool("render", {"status": "failed"})
+            await client.call_tool("report", {})
+
+        links = captured["links"]
+        assert [(link.tool, link.ok) for link in links] == [("render", False)]
+        # Filled from memory or read from the database, depending on timing.
+        [row] = await mw.feedback_context("bug-1")
+        assert (row["outcome"], row["ok"], row["error_type"]) == (
+            "soft_error",
+            False,
+            "SoftError",
+        )
+        await mw.flush()
+        [row] = await mw.feedback_context("bug-1")
+        assert (row["tool"], row["outcome"], row["ok"]) == (
+            "render",
+            "soft_error",
+            False,
+        )
+        await mw.aclose()
+
+    async def test_submit_feedback_failure_is_soft_error(self, monkeypatch):
+        def broken(request):
+            raise RuntimeError("database down, password=hunter2")
+
+        monkeypatch.setattr(submission, "create_feedback_from_request", broken)
+        app = FastMCP("fb")
+        sink = MemorySink()
+        mw = instrument(app, [sink])
+        add_feedback_tools(app, database_url="sqlite:///:memory:", instrumentation=mw)
+
+        async with Client(app) as client:
+            result = await client.call_tool(
+                "submit_feedback",
+                {
+                    "request": {
+                        "type": "bug",
+                        "title": "t",
+                        "description": "d",
+                        "submitter": "m",
+                    }
+                },
+            )
+        await mw.flush()
+        assert result.data["success"] is False
+        [record] = sink.records
+        assert (record.tool, record.outcome, record.ok, record.error_type) == (
+            "submit_feedback",
+            "soft_error",
+            False,
+            "SoftError",
+        )
+        assert record.error_message == (
+            f"success=False: database down, password={REDACTED}"
+        )

tests/unit/test_test_databases.py

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

uv.lock

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