Merge feature/retention-sampling: retention, exclusion and sampling

CHANGELOG.md

@@ -5,6 +5,40 @@ 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.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.

README.md

@@ -336,6 +336,80 @@ 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.
+
 ### Middleware order and shutdown
 
 FastMCP runs the first middleware added as the outermost. Call `instrument()`

pyproject.toml

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

src/fastmcp_feedback/__init__.py

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

src/fastmcp_feedback/instrumentation/middleware.py

@@ -5,10 +5,11 @@ 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
 
@@ -83,6 +84,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:
@@ -137,6 +142,14 @@ class InstrumentationMiddleware(Middleware):
             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.
     """
 
     def __init__(
@@ -157,6 +170,9 @@ class InstrumentationMiddleware(Middleware):
         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,
     ):
         mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
         if mode not in MODES:
@@ -171,6 +187,16 @@ class InstrumentationMiddleware(Middleware):
         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.dispatcher = RecordDispatcher(
             sinks if sinks is not None else [JsonLinesSink()],
@@ -182,6 +208,8 @@ class InstrumentationMiddleware(Middleware):
     async def on_call_tool(self, context: MiddlewareContext, call_next):
         if self.mode == "off":
             return await call_next(context)
+        if _tool_name(context) in self.exclude_tools:
+            return await call_next(context)
 
         started_at = datetime.now(UTC)
         start = time.perf_counter()
@@ -196,14 +224,35 @@ class InstrumentationMiddleware(Middleware):
         finally:
             duration_ms = (time.perf_counter() - start) * 1000
             try:
-                record = await self._build_record(
-                    context, started_at, duration_ms, result, error
-                )
-                self.recent.append(record)
-                self.dispatcher.submit(record)
+                await self._record(context, started_at, duration_ms, result, error)
             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,
+    ) -> 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.
+        if rate is not None and (error is not None or soft is not None):
+            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
+        )
+        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 {}
@@ -237,14 +286,14 @@ class InstrumentationMiddleware(Middleware):
         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)
@@ -266,14 +315,12 @@ class InstrumentationMiddleware(Middleware):
             error_message = self.redactor.redact_text(str(cause))[
                 : self.max_error_chars
             ]
-        elif self.result_classifier is not None and payload is not None:
-            soft = self._classify(tool, result, payload)
-            if soft is not None:
-                outcome = "soft_error"
-                error_type = "SoftError"
-                error_message = (
-                    self.redactor.redact_text(soft)[: self.max_error_chars] or None
-                )
+        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,
@@ -298,6 +345,7 @@ class InstrumentationMiddleware(Middleware):
             if full and payload is not None
             else None,
             extra=extra,
+            sample_rate=sample_rate,
         )
 
     @property

src/fastmcp_feedback/instrumentation/record.py

@@ -16,6 +16,10 @@ class ToolCallRecord:
     ``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.
     """
 
     tool: str
@@ -39,6 +43,7 @@ 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:

src/fastmcp_feedback/instrumentation/storage.py

@@ -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,6 +25,7 @@ from sqlalchemy import (
     Text,
     and_,
     delete,
+    exists,
     or_,
     select,
 )
@@ -31,6 +34,8 @@ from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 
 from .record import ToolCallRecord
 
+logger = logging.getLogger(__name__)
+
 DEFAULT_TABLE_PREFIX = "ffb_"
 
 # none_as_null: a missing payload is SQL NULL, so "WHERE args IS NULL" works.
@@ -73,6 +78,9 @@ 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 links not in metadata.tables:
         # No foreign key to tool_calls: a link can be written before its call
@@ -100,6 +108,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 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__(
@@ -108,7 +123,14 @@ 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)
@@ -119,6 +141,11 @@ class DatabaseSink:
         # 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."""
@@ -138,6 +165,59 @@ class DatabaseSink:
         await self._ready()
         async with self.engine.begin() as conn:
             await conn.execute(self.table.insert(), [r.to_row() for r in records])
+        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:
+            deleted = await self.prune()
+            if deleted:
+                logger.info(
+                    "Pruned %d tool-call rows from %s", deleted, self.table.name
+                )
+        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 rows that started 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. Deletes ``prune_batch`` rows per
+        transaction until none are left. Returns the number of rows deleted.
+        """
+        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
+        batch = (
+            select(t.c.id)
+            .where(
+                t.c.started_at < older_than,
+                ~exists().where(link.c.call_id == t.c.id),
+            )
+            .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(t).where(t.c.id.in_(ids)))
+            total += len(ids)
+            if len(ids) < self.prune_batch:
+                return total
 
     async def recent_calls(
         self,

tests/unit/test_retention_sampling.py

@@ -0,0 +1,402 @@
+"""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(tmp_path):
+    """DatabaseSinks whose engines are always disposed, or the process hangs."""
+    made = []
+
+    def factory(name="calls", **kwargs):
+        sink = DatabaseSink(
+            f"sqlite+aiosqlite:///{tmp_path}/{name}.db", create_tables=True, **kwargs
+        )
+        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",)]

uv.lock

@@ -553,7 +553,7 @@ wheels = [
 
 [[package]]
 name = "fastmcp-feedback"
-version = "2026.9.27.4"
+version = "2026.9.28"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },