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

The showcase repo is private and fastmcp-feedback-showcase was never
published, so both the link and the uvx command failed for everyone else.
Tested the new clone-and-run instructions against the now-public repo.

CHANGELOG.md

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

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.
@@ -201,6 +201,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 +298,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 +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()`
@@ -390,10 +520,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 +578,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

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.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.3"
+__version__ = "2026.09.28"
 __author__ = "Ryan Malloy"
 __email__ = "ryan@supported.systems"
 

src/fastmcp_feedback/instrumentation/__init__.py

@@ -4,8 +4,14 @@ 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",)))
 """
 
+from .classify import (
+    DEFAULT_RESULT_CLASSIFIER,
+    ResultClassifier,
+    soft_error_classifier,
+)
 from .dispatch import RecordDispatcher
 from .middleware import MODE_ENV_VAR, MODES, InstrumentationMiddleware, instrument
 from .record import ToolCallRecord
@@ -21,6 +27,7 @@ from .storage import DEFAULT_TABLE_PREFIX, DatabaseSink, build_metadata
 __all__ = [
     "CallbackSink",
     "DatabaseSink",
+    "DEFAULT_RESULT_CLASSIFIER",
     "DEFAULT_SECRET_KEYS",
     "DEFAULT_SECRET_PATTERNS",
     "DEFAULT_TABLE_PREFIX",
@@ -32,8 +39,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/middleware.py

@@ -5,16 +5,18 @@ 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,
@@ -82,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:
@@ -131,6 +137,19 @@ 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.
     """
 
     def __init__(
@@ -150,6 +169,10 @@ 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,
     ):
         mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
         if mode not in MODES:
@@ -163,6 +186,17 @@ 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.dispatcher = RecordDispatcher(
             sinks if sinks is not None else [JsonLinesSink()],
@@ -174,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()
@@ -188,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 {}
@@ -208,19 +265,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 +305,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,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

@@ -12,6 +12,14 @@ 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.
     """
 
     tool: str
@@ -19,6 +27,7 @@ class ToolCallRecord:
     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 +43,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/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.
@@ -57,6 +62,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 +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
@@ -98,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__(
@@ -106,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)
@@ -117,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."""
@@ -136,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,
@@ -194,6 +276,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/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",)]

tests/unit/test_soft_errors.py

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

uv.lock

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