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,113 @@ 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.
+
+### Added
+- `InstrumentationMiddleware.capture_recent_calls(ctx, n=20)` returns the
+  calls that led up to now. It matches the same client session first
+  (`session` rule), then fills remaining slots with calls by the same
+  `user_sub` inside `window` (`user_window` rule, 15 minutes by default), and
+  each link records which rule matched.
+- Calls are kept in an in-memory buffer as they happen, so calls still
+  waiting in the sink queue count. With a `DatabaseSink` the lookup falls back
+  to the database, which covers calls from other workers and from before a
+  restart.
+- `link_feedback(ref, links)` stores links against a feedback id of any
+  format (`bug-7Q2X`, `42`) in a new `{prefix}feedback_call_links` table.
+  `build_metadata()` includes it; hosts that manage the schema with Alembic
+  need a new migration.
+- `feedback_context(ref)` returns the linked calls in order with tool,
+  outcome, error type and matching rule, filling in calls not yet written
+  from memory.
+- `add_feedback_tools(..., instrumentation=mw)`: `submit_feedback` links the
+  preceding calls automatically and returns `linked_calls` in its response.
+- `session_key(ctx)` gives a per-client key that stays stable across calls:
+  the HTTP `mcp-session-id` header when present, then one key per process for
+  stdio, then `ctx.session_id`.
+
+### Fixed
+- With `create_tables=True`, the background writer and `link_feedback` could
+  both run `create_all` at once and fail with "table already exists". Table
+  creation is now serialized by a lock.
+
+### Note
+- Clients on MCP protocol 2026-07-28 (Claude Code, FastMCP 4's `Client`)
+  open no HTTP sessions, so their calls link by user only. HTTP servers should
+  configure an `identity_resolver`. Stdio servers link by session with every
+  client.
+
 ## [2026.09.27.2] - 2026-09-27
 
 ### Fixed

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
@@ -223,6 +279,137 @@ go in `identity`. Enricher output goes in `extra`, except `server_version`,
 which has its own column. Async hooks that take longer than `hook_timeout`
 (0.25 s by default) are skipped for that call.
 
+### Linking feedback to the calls behind it
+
+When a model reports "the render tool hung", the report is far more useful with
+the calls that led up to it attached. Pass the middleware to
+`add_feedback_tools` and every `submit_feedback` does this for you:
+
+```python
+from fastmcp_feedback import add_feedback_tools
+from fastmcp_feedback.instrumentation import DatabaseSink, instrument
+
+mw = instrument(app, [DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True)])
+add_feedback_tools(app, database_url="sqlite:///feedback.db", instrumentation=mw)
+# submit_feedback now answers {"feedback_id": ..., "linked_calls": 3, ...}
+```
+
+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["outcome"], call["error_type"], call["rule"])
+```
+
+If you have your own feedback tools, use the same pieces directly. Feedback ids
+can be anything (`bug-7Q2X`, `42`):
+
+```python
+from fastmcp import Context
+
+@app.tool
+async def report_problem(summary: str, ctx: Context) -> dict:
+    ref = save_my_feedback(summary)                 # your storage, your id
+    links = await mw.capture_recent_calls(ctx, n=20)
+    await mw.link_feedback(ref, links)
+    return {"id": ref, "linked_calls": len(links)}
+```
+
+Calls are matched in two ways, and each link records which one (`rule`):
+
+- `session`: the same MCP client session. Over HTTP this is the
+  `mcp-session-id`; over stdio it is the server process, which serves one client.
+- `user_window`: the same `user_sub` within the last 15 minutes (`window=`),
+  for clients that open a new session per call (personal access tokens, CI).
+  This needs an `identity_resolver`.
+
+Session matches come first, and user matches fill any remaining slots. Calls
+still waiting in the queue are included, and with a `DatabaseSink`, so are
+calls from other workers or from before a restart.
+
+Links live in a `ffb_feedback_call_links` table next to `ffb_tool_calls`.
+`build_metadata()` includes it, so hosts that manage the schema with Alembic
+need one new migration before calling `link_feedback`.
+
+**Stateless clients:** clients on MCP protocol 2026-07-28, including Claude
+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()`
@@ -297,7 +484,7 @@ version passes the suite.
 
 ## API Reference
 
-### `add_feedback_tools(mcp, database_url=None, insights=None, prefix="", separator="_")`
+### `add_feedback_tools(mcp, database_url=None, insights=None, prefix="", separator="_", instrumentation=None)`
 
 Add all five feedback tools to a FastMCP server.
 
@@ -307,6 +494,9 @@ Add all five feedback tools to a FastMCP server.
 - `insights`: a `FeedbackInsights` instance. Defaults to one with analytics disabled
 - `prefix`: prepended to every tool name
 - `separator`: placed between `prefix` and the tool name
+- `instrumentation`: the middleware returned by `instrument()`. When given,
+  `submit_feedback` links the calls that preceded it to the new item and
+  returns `linked_calls` (see [Linking feedback to the calls behind it](#linking-feedback-to-the-calls-behind-it))
 
 ### Feedback Types
 
@@ -330,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 comprehensive examples demonstrating all capabilities:
+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
@@ -381,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.2"  # 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.2"
+__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/correlation.py

@@ -0,0 +1,106 @@
+"""Linking feedback to the tool calls that came before it."""
+
+from collections.abc import Iterable
+from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
+from typing import Any
+
+RULE_SESSION = "session"
+RULE_USER_WINDOW = "user_window"
+
+
+@dataclass
+class CallLink:
+    """A tool call linked to a piece of feedback, and why it matched."""
+
+    call_id: str
+    rule: str
+    tool: str
+    started_at: datetime
+    ok: bool
+    position: int = 0
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["started_at"] = self.started_at.isoformat()
+        return data
+
+
+@dataclass
+class Candidate:
+    """The fields correlation needs, from a buffered record or a database row."""
+
+    id: str
+    tool: str
+    started_at: datetime
+    ok: bool
+    session_id: str | None
+    user_sub: str | None
+
+
+class ResolverContext:
+    """What an identity resolver receives when called during correlation.
+
+    Outside a tool call there is no MiddlewareContext, so this carries only the
+    ``fastmcp_context`` attribute, which is what resolvers normally read.
+    """
+
+    def __init__(self, fastmcp_context: Any):
+        self.fastmcp_context = fastmcp_context
+        self.message = None
+
+
+def as_utc(value: datetime) -> datetime:
+    # SQLite drops the timezone from DateTime(timezone=True) columns.
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value
+
+
+def select_calls(
+    candidates: Iterable[Candidate],
+    *,
+    session_id: str | None,
+    user_sub: str | None,
+    since: datetime,
+    limit: int,
+) -> list[CallLink]:
+    """Pick up to ``limit`` recent calls for a feedback item.
+
+    Calls from the same MCP session come first. If there are fewer than
+    ``limit``, the rest are filled with the same user's calls since ``since``,
+    which covers clients that open a new session per call (PATs, CI). The
+    result is in chronological order, numbered from 0.
+    """
+    unique: dict[str, Candidate] = {}
+    for c in candidates:
+        unique.setdefault(c.id, c)
+    newest_first = sorted(
+        unique.values(), key=lambda c: as_utc(c.started_at), reverse=True
+    )
+
+    chosen: list[tuple[Candidate, str]] = []
+    if session_id:
+        chosen += [
+            (c, RULE_SESSION) for c in newest_first if c.session_id == session_id
+        ][:limit]
+    if user_sub and len(chosen) < limit:
+        taken = {c.id for c, _ in chosen}
+        chosen += [
+            (c, RULE_USER_WINDOW)
+            for c in newest_first
+            if c.id not in taken
+            and c.user_sub == user_sub
+            and as_utc(c.started_at) >= since
+        ][: limit - len(chosen)]
+
+    chosen.sort(key=lambda pair: as_utc(pair[0].started_at))
+    return [
+        CallLink(
+            call_id=c.id,
+            rule=rule,
+            tool=c.tool,
+            started_at=as_utc(c.started_at),
+            ok=c.ok,
+            position=i,
+        )
+        for i, (c, rule) in enumerate(chosen)
+    ]

src/fastmcp_feedback/instrumentation/middleware.py

@@ -5,18 +5,30 @@ import inspect
 import json
 import logging
 import os
+import random as _random
 import time
-from collections.abc import Awaitable, Callable, Iterable, Sequence
-from datetime import UTC, datetime
+import uuid
+from collections import deque
+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,
+    ResolverContext,
+    as_utc,
+    select_calls,
+)
 from .dispatch import RecordDispatcher
 from .record import ToolCallRecord
 from .redaction import Redactor
 from .sinks import JsonLinesSink, Sink
+from .storage import DatabaseSink
 
 logger = logging.getLogger(__name__)
 
@@ -43,6 +55,39 @@ def _attr(obj: Any, name: str) -> Any:
     return str(value) if value is not None else None
 
 
+# stdio serves exactly one client per process, so the process is the session.
+_STDIO_SESSION = f"stdio:{os.getpid()}:{uuid.uuid4().hex[:8]}"
+
+
+def session_key(ctx: Any) -> str | None:
+    """A per-client key that stays the same across that client's calls.
+
+    Clients on MCP protocol 2026-07-28 (Claude Code, FastMCP 4's Client) do
+    not open sessions, and FastMCP then generates a new ``ctx.session_id`` for
+    every request. So stable signals come first: the HTTP ``mcp-session-id``
+    header (older protocol clients), then one key per process for stdio, then
+    ``ctx.session_id``. With a stateless HTTP client there is no session, and
+    correlation relies on the same-user rule.
+    """
+    if ctx is None:
+        return None
+    try:
+        from fastmcp.server.dependencies import get_http_headers
+
+        header = (get_http_headers(include_all=True) or {}).get("mcp-session-id")
+        if header:
+            return header
+    except Exception:
+        pass
+    if _attr(ctx, "transport") == "stdio":
+        return _STDIO_SESSION
+    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:
@@ -88,6 +133,23 @@ class InstrumentationMiddleware(Middleware):
             May be async.
         server_version: default for the ``server_version`` column.
         hook_timeout: seconds an async hook may take before it is skipped.
+        recent_calls: how many recent records to keep in memory for
+            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__(
@@ -105,6 +167,12 @@ class InstrumentationMiddleware(Middleware):
         batch_size: int = 100,
         sink_timeout: float = 10.0,
         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:
@@ -117,6 +185,19 @@ class InstrumentationMiddleware(Middleware):
         self.server_version = server_version
         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()],
             max_queue=max_queue,
@@ -127,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()
@@ -141,13 +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.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 {}
@@ -160,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)
@@ -184,24 +305,33 @@ 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,
-            session_id=_attr(ctx, "session_id"),
+            session_id=session_key(ctx),
             request_id=_attr(ctx, "request_id"),
             client_id=columns.get("client_id") or _attr(ctx, "client_id"),
             user_sub=columns.get("user_sub"),
@@ -215,8 +345,143 @@ class InstrumentationMiddleware(Middleware):
             if full and payload is not None
             else None,
             extra=extra,
+            sample_rate=sample_rate,
         )
 
+    @property
+    def database_sink(self) -> DatabaseSink | None:
+        """The first DatabaseSink among the sinks, if any."""
+        return next(
+            (s for s in self.dispatcher.sinks if isinstance(s, DatabaseSink)), None
+        )
+
+    async def capture_recent_calls(
+        self,
+        ctx: Any = None,
+        n: int = 20,
+        *,
+        user_sub: str | None = None,
+        window: timedelta = timedelta(minutes=15),
+    ) -> list[CallLink]:
+        """The calls that led up to now, for attaching to a feedback item.
+
+        Call from inside a tool. Returns up to ``n`` calls in chronological
+        order: first from the current MCP session, then, if that yields fewer
+        than ``n``, the same user's calls within ``window``. Each link says
+        which rule matched. Never raises; returns [] on any failure.
+
+        Args:
+            ctx: the FastMCP Context. Looked up automatically when omitted.
+            n: maximum number of calls.
+            user_sub: the caller's user id. When omitted, the identity
+                resolver is asked, given an object whose ``fastmcp_context``
+                is ``ctx``.
+            window: how far back the same-user fallback reaches.
+        """
+        try:
+            if ctx is None:
+                ctx = _current_context()
+            session_id = session_key(ctx)
+            if user_sub is None and self.identity_resolver is not None:
+                identity = await self._call_hook(
+                    self.identity_resolver, ResolverContext(ctx)
+                )
+                user_sub = identity.get("user_sub")
+            since = datetime.now(UTC) - window
+
+            candidates = [
+                Candidate(r.id, r.tool, r.started_at, r.ok, r.session_id, r.user_sub)
+                for r in self.recent
+            ]
+            links = select_calls(
+                candidates,
+                session_id=session_id,
+                user_sub=user_sub,
+                since=since,
+                limit=n,
+            )
+            sink = self.database_sink
+            if sink is not None and len(links) < n:
+                # Other workers, or calls from before a restart.
+                rows = await asyncio.wait_for(
+                    sink.recent_calls(
+                        session_id=session_id, user_sub=user_sub, since=since, limit=n
+                    ),
+                    self.correlation_timeout,
+                )
+                candidates += [
+                    Candidate(
+                        r["id"],
+                        r["tool"],
+                        as_utc(r["started_at"]),
+                        r["ok"],
+                        r["session_id"],
+                        r["user_sub"],
+                    )
+                    for r in rows
+                ]
+                links = select_calls(
+                    candidates,
+                    session_id=session_id,
+                    user_sub=user_sub,
+                    since=since,
+                    limit=n,
+                )
+            return links
+        except Exception:
+            logger.warning("Failed to capture recent calls", exc_info=True)
+            return []
+
+    async def link_feedback(self, feedback_ref: str, links: Sequence[CallLink]) -> bool:
+        """Store ``links`` against a feedback id of any format (``bug-XXXX``, 42).
+
+        Needs a DatabaseSink. Replaces earlier links for the same id. Returns
+        False, without raising, if they could not be stored.
+        """
+        sink = self.database_sink
+        if sink is None:
+            logger.warning("link_feedback needs a DatabaseSink; links not stored")
+            return False
+        linked_at = datetime.now(UTC)
+        try:
+            await sink.replace_links(
+                str(feedback_ref),
+                [
+                    {
+                        "feedback_ref": str(feedback_ref),
+                        "call_id": link.call_id,
+                        "position": link.position,
+                        "rule": link.rule,
+                        "linked_at": linked_at,
+                    }
+                    for link in links
+                ],
+            )
+            return True
+        except Exception:
+            logger.warning("Failed to link feedback %s", feedback_ref, exc_info=True)
+            return False
+
+    async def feedback_context(self, feedback_ref: str) -> list[dict[str, Any]]:
+        """The calls linked to a feedback id, in order, with their details.
+
+        Calls not yet written to the database are filled in from memory.
+        """
+        sink = self.database_sink
+        if sink is None:
+            return []
+        rows = await sink.linked_calls(str(feedback_ref))
+        in_memory = {r.id: r for r in self.recent}
+        for row in rows:
+            if row["tool"] is None and row["call_id"] in in_memory:
+                record = in_memory[row["call_id"]]
+                for key in row:
+                    if key not in ("call_id", "position", "rule"):
+                        row[key] = getattr(record, key, None)
+            if row["started_at"] is not None:
+                row["started_at"] = as_utc(row["started_at"])
+        return rows
+
     async def flush(self) -> None:
         """Wait until queued records have reached the sinks."""
         await self.dispatcher.flush()
@@ -226,6 +491,15 @@ class InstrumentationMiddleware(Middleware):
         await self.dispatcher.aclose()
 
 
+def _current_context() -> Any:
+    try:
+        from fastmcp.server.dependencies import get_context
+
+        return get_context()
+    except Exception:
+        return None
+
+
 def instrument(app: FastMCP, sinks: Sequence[Sink] | None = None, **kwargs: Any):
     """Add an InstrumentationMiddleware to ``app`` and return it.
 

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

@@ -1,11 +1,16 @@
-"""Async SQLAlchemy storage for tool-call records.
+"""Async SQLAlchemy storage for tool-call records and feedback links.
 
 Hosts that manage their own schema include ``build_metadata(prefix)`` in their
 Alembic ``env.py`` (``target_metadata``) and pass their existing AsyncEngine to
 ``DatabaseSink``. Nothing is created unless ``create_tables=True``.
 """
 
+import asyncio
+import logging
+import time
 from collections.abc import Sequence
+from datetime import UTC, datetime, timedelta
+from typing import Any
 
 from sqlalchemy import (
     JSON,
@@ -18,12 +23,19 @@ from sqlalchemy import (
     String,
     Table,
     Text,
+    and_,
+    delete,
+    exists,
+    or_,
+    select,
 )
 from sqlalchemy.dialects.postgresql import JSONB
 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.
@@ -33,38 +45,55 @@ _JSON = JSON(none_as_null=True).with_variant(JSONB(none_as_null=True), "postgres
 def build_metadata(
     prefix: str = DEFAULT_TABLE_PREFIX, metadata: MetaData | None = None
 ) -> MetaData:
-    """Return MetaData holding the ``{prefix}tool_calls`` table.
+    """Return MetaData holding ``{prefix}tool_calls`` and ``{prefix}feedback_call_links``.
 
-    Pass an existing MetaData to add the table to it instead.
+    Pass an existing MetaData to add the tables to it instead.
     """
     metadata = metadata if metadata is not None else MetaData()
-    name = f"{prefix}tool_calls"
-    if name in metadata.tables:
-        return metadata
-    Table(
-        name,
-        metadata,
-        Column("id", String(36), primary_key=True),
-        Column("started_at", DateTime(timezone=True), nullable=False, index=True),
-        Column("duration_ms", Float, nullable=False),
-        Column("tool", String(255), nullable=False, index=True),
-        Column("mode", String(8), nullable=False),
-        Column("ok", Boolean, nullable=False, index=True),
-        Column("error_type", String(255)),
-        Column("error_message", Text),
-        Column("session_id", String(255), index=True),
-        Column("request_id", String(255)),
-        Column("client_id", String(255)),
-        Column("user_sub", String(255), index=True),
-        Column("caller_kind", String(64)),
-        Column("identity", _JSON),
-        Column("server_version", String(64), index=True),
-        Column("args_size", Integer),
-        Column("result_size", Integer),
-        Column("args", _JSON),
-        Column("result", _JSON),
-        Column("extra", _JSON),
-    )
+    calls = f"{prefix}tool_calls"
+    links = f"{prefix}feedback_call_links"
+    if calls not in metadata.tables:
+        Table(
+            calls,
+            metadata,
+            Column("id", String(36), primary_key=True),
+            Column("started_at", DateTime(timezone=True), nullable=False, index=True),
+            Column("duration_ms", Float, nullable=False),
+            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),
+            Column("request_id", String(255)),
+            Column("client_id", String(255)),
+            Column("user_sub", String(255), index=True),
+            Column("caller_kind", String(64)),
+            Column("identity", _JSON),
+            Column("server_version", String(64), index=True),
+            Column("args_size", Integer),
+            Column("result_size", Integer),
+            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
+        # record has left the queue, and hosts may keep calls elsewhere.
+        Table(
+            links,
+            metadata,
+            Column("feedback_ref", String(255), primary_key=True),
+            Column("call_id", String(36), primary_key=True, index=True),
+            Column("position", Integer, nullable=False),
+            Column("rule", String(32), nullable=False),
+            Column("linked_at", DateTime(timezone=True), nullable=False),
+        )
     return metadata
 
 
@@ -76,9 +105,16 @@ class DatabaseSink:
             ``sqlite+aiosqlite:///calls.db`` or ``postgresql+asyncpg://...``.
             An engine built from a URL is disposed on ``aclose()``; a passed-in
             engine is left alone.
-        prefix: table name prefix, so the table cannot collide with the host's.
-        create_tables: create the table on first write. Off by default; hosts
+        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__(
@@ -87,27 +123,174 @@ class DatabaseSink:
         *,
         prefix: str = DEFAULT_TABLE_PREFIX,
         create_tables: bool = False,
+        retention: timedelta | None = None,
+        prune_interval: timedelta = timedelta(hours=1),
+        prune_batch: int = 5000,
     ):
+        if retention is not None and retention <= timedelta(0):
+            raise ValueError(f"retention must be positive, got {retention!r}")
+        if prune_batch < 1:
+            raise ValueError(f"prune_batch must be at least 1, got {prune_batch!r}")
         self._owns_engine = isinstance(engine, str)
         self.engine = create_async_engine(engine) if isinstance(engine, str) else engine
         self.metadata = build_metadata(prefix)
         self.table = self.metadata.tables[f"{prefix}tool_calls"]
+        self.links = self.metadata.tables[f"{prefix}feedback_call_links"]
         self.create_tables = create_tables
         self._tables_ready = False
+        # The background writer and link_feedback can both reach _ready() at
+        # once; without the lock both run create_all and one fails.
+        self._tables_lock = asyncio.Lock()
+        self.retention = retention
+        self.prune_interval = prune_interval
+        self.prune_batch = prune_batch
+        # time.monotonic() of the last automatic prune; None until the first.
+        self._last_prune: float | None = None
 
     async def ensure_tables(self) -> None:
-        """Create the table if missing. Only called when create_tables=True."""
+        """Create the tables if missing. Only called when create_tables=True."""
         async with self.engine.begin() as conn:
             await conn.run_sync(self.metadata.create_all)
         self._tables_ready = True
 
+    async def _ready(self) -> None:
+        if self.create_tables and not self._tables_ready:
+            async with self._tables_lock:
+                if not self._tables_ready:
+                    await self.ensure_tables()
+
     async def write(self, records: Sequence[ToolCallRecord]) -> None:
         if not records:
             return
-        if self.create_tables and not self._tables_ready:
-            await self.ensure_tables()
+        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,
+        *,
+        session_id: str | None,
+        user_sub: str | None,
+        since: datetime,
+        limit: int,
+    ) -> list[dict[str, Any]]:
+        """Calls from this session, or from this user since ``since``, newest first."""
+        t = self.table
+        conditions = []
+        if session_id:
+            conditions.append(t.c.session_id == session_id)
+        if user_sub:
+            conditions.append(and_(t.c.user_sub == user_sub, t.c.started_at >= since))
+        if not conditions:
+            return []
+        await self._ready()
+        query = (
+            select(
+                t.c.id, t.c.tool, t.c.started_at, t.c.ok, t.c.session_id, t.c.user_sub
+            )
+            .where(or_(*conditions))
+            .order_by(t.c.started_at.desc())
+            .limit(limit)
+        )
+        async with self.engine.connect() as conn:
+            return [dict(row) for row in (await conn.execute(query)).mappings()]
+
+    async def replace_links(
+        self, feedback_ref: str, links: Sequence[dict[str, Any]]
+    ) -> None:
+        """Set the calls linked to ``feedback_ref``, replacing any earlier set."""
+        await self._ready()
+        async with self.engine.begin() as conn:
+            await conn.execute(
+                delete(self.links).where(self.links.c.feedback_ref == feedback_ref)
+            )
+            if links:
+                await conn.execute(self.links.insert(), list(links))
+
+    async def linked_calls(self, feedback_ref: str) -> list[dict[str, Any]]:
+        """Links for ``feedback_ref`` in order, joined to their call records.
+
+        Call columns are None for calls not (yet) in this database.
+        """
+        await self._ready()
+        link, call = self.links, self.table
+        query = (
+            select(
+                link.c.call_id,
+                link.c.position,
+                link.c.rule,
+                call.c.tool,
+                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,
+                call.c.user_sub,
+                call.c.caller_kind,
+                call.c.server_version,
+                call.c.extra,
+            )
+            .select_from(link.outerjoin(call, link.c.call_id == call.c.id))
+            .where(link.c.feedback_ref == feedback_ref)
+            .order_by(link.c.position)
+        )
+        async with self.engine.connect() as conn:
+            return [dict(row) for row in (await conn.execute(query)).mappings()]
 
     async def aclose(self) -> None:
         if self._owns_engine:

src/fastmcp_feedback/mixins/submission.py

@@ -1,15 +1,37 @@
 """Submission mixin for feedback collection."""
 
 from datetime import UTC, datetime
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from ..insights import record_feedback_submission
 from ..models import SubmitFeedbackRequest, create_feedback_from_request
-from .base import BaseFeedbackMixin, logger, mcp_tool
+from .base import (
+    BaseFeedbackMixin,
+    FeedbackDatabase,
+    FeedbackInsights,
+    logger,
+    mcp_tool,
+)
+
+if TYPE_CHECKING:
+    from ..instrumentation import InstrumentationMiddleware
 
 
 class SubmissionMixin(BaseFeedbackMixin):
-    """Mixin for feedback submission functionality."""
+    """Mixin for feedback submission functionality.
+
+    Pass an InstrumentationMiddleware to link each submission to the tool
+    calls that led up to it.
+    """
+
+    def __init__(
+        self,
+        database: FeedbackDatabase,
+        insights: FeedbackInsights | None = None,
+        instrumentation: "InstrumentationMiddleware | None" = None,
+    ):
+        super().__init__(database, insights)
+        self.instrumentation = instrumentation
 
     @mcp_tool(
         description="Submit feedback about your MCP experience - Use this tool when you encounter issues, have suggestions for improvement, or want to report positive experiences. AI models should proactively use this to help improve future MCP sessions. PRIVACY: Only include information you've already shared via previous MCP calls. Ask human permission before including any new sensitive data."
@@ -62,14 +84,23 @@ class SubmissionMixin(BaseFeedbackMixin):
                     source="mcp_tool",
                 )
 
+            linked_calls = 0
+            if self.instrumentation is not None:
+                links = await self.instrumentation.capture_recent_calls()
+                if await self.instrumentation.link_feedback(feedback_id, links):
+                    linked_calls = len(links)
+
             self._record_tool_usage("submit_feedback", start_time, True)
             logger.info(f"Feedback submitted successfully: {feedback_id}")
 
-            return {
+            response = {
                 "success": True,
                 "feedback_id": feedback_id,
                 "message": "Feedback submitted successfully",
             }
+            if self.instrumentation is not None:
+                response["linked_calls"] = linked_calls
+            return response
 
         except Exception as e:
             logger.error(f"Failed to submit feedback: {e}")

src/fastmcp_feedback/tools.py

@@ -17,6 +17,7 @@ def add_feedback_tools(
     insights: FeedbackInsights | None = None,
     prefix: str = "",
     separator: str = "_",
+    instrumentation=None,
 ) -> None:
     """Add all feedback tools to a FastMCP server.
 
@@ -29,6 +30,9 @@ def add_feedback_tools(
         insights: Optional insights instance. Creates default if not provided.
         prefix: Optional prefix for tool names.
         separator: Separator between prefix and tool name.
+        instrumentation: an InstrumentationMiddleware (see ``instrument``).
+            When given, each submission is linked to the tool calls that led
+            up to it, and the response reports ``linked_calls``.
     """
     try:
         # Initialize database
@@ -39,7 +43,7 @@ def add_feedback_tools(
             insights = FeedbackInsights()
 
         # Create mixin instances
-        submission_mixin = SubmissionMixin(database, insights)
+        submission_mixin = SubmissionMixin(database, insights, instrumentation)
         retrieval_mixin = RetrievalMixin(database, insights)
         management_mixin = ManagementMixin(database, insights)
 

tests/fixtures/stdio_correlation_server.py

@@ -0,0 +1,22 @@
+"""A stdio MCP server used by test_correlation to check session linking end to end."""
+
+from fastmcp import Context, FastMCP
+
+from fastmcp_feedback.instrumentation import MemorySink, instrument
+
+app = FastMCP("stdio-correlation")
+mw = instrument(app, [MemorySink()])
+
+
+@app.tool
+def echo(text: str) -> str:
+    return text
+
+
+@app.tool
+async def report(ctx: Context) -> list:
+    return [[link.tool, link.rule] for link in await mw.capture_recent_calls(ctx)]
+
+
+if __name__ == "__main__":
+    app.run(show_banner=False)

tests/unit/test_correlation.py

@@ -0,0 +1,416 @@
+"""Tests for linking feedback to the tool calls that preceded it."""
+
+import asyncio
+import contextlib
+import json
+import sys
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import fastmcp
+import pytest
+from fastmcp import Client, Context, FastMCP
+from fastmcp.client.transports import PythonStdioTransport
+from fastmcp.exceptions import ToolError
+
+from fastmcp_feedback import add_feedback_tools
+from fastmcp_feedback.instrumentation import (
+    DatabaseSink,
+    MemorySink,
+    instrument,
+)
+from fastmcp_feedback.instrumentation import middleware as mw_module
+from fastmcp_feedback.instrumentation.correlation import (
+    RULE_SESSION,
+    RULE_USER_WINDOW,
+    Candidate,
+    select_calls,
+)
+
+NOW = datetime(2026, 9, 27, 12, 0, tzinfo=UTC)
+
+# FastMCP 4's Client speaks MCP 2026-07-28, which has no sessions, so each
+# in-memory call gets a new ctx.session_id and there is no header or stdio
+# signal to fall back on. Calls from one test client link by user instead.
+FASTMCP_4 = int(fastmcp.__version__.split(".")[0]) >= 4
+SAME_CLIENT_RULE = RULE_USER_WINDOW if FASTMCP_4 else RULE_SESSION
+
+
+@pytest.fixture
+async def db(tmp_path):
+    """Make DatabaseSinks whose engines are always disposed, even on failure.
+
+    An undisposed aiosqlite engine keeps a worker thread alive and the test
+    process never exits.
+    """
+    made = []
+
+    def factory(name="calls", cls=DatabaseSink):
+        sink = cls(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 cand(id, minutes_ago, session="s1", user="u1", tool="t", ok=True):
+    return Candidate(id, tool, NOW - timedelta(minutes=minutes_ago), ok, session, user)
+
+
+class TestSelectCalls:
+    def test_session_first_then_user_window_chronological(self):
+        links = select_calls(
+            [cand("a", 1), cand("b", 3, session="s2"), cand("c", 2)],
+            session_id="s1",
+            user_sub="u1",
+            since=NOW - timedelta(minutes=15),
+            limit=10,
+        )
+        assert [(link.call_id, link.rule, link.position) for link in links] == [
+            ("b", RULE_USER_WINDOW, 0),
+            ("c", RULE_SESSION, 1),
+            ("a", RULE_SESSION, 2),
+        ]
+
+    def test_session_matches_take_priority_over_newer_user_calls(self):
+        links = select_calls(
+            [cand("old", 10), cand("new", 1, session="s2")],
+            session_id="s1",
+            user_sub="u1",
+            since=NOW - timedelta(minutes=15),
+            limit=1,
+        )
+        assert [(link.call_id, link.rule) for link in links] == [("old", RULE_SESSION)]
+
+    def test_user_window_excludes_old_calls_and_other_users(self):
+        links = select_calls(
+            [
+                cand("recent", 5, session="s2"),
+                cand("stale", 30, session="s2"),
+                cand("other", 1, session="s3", user="u2"),
+            ],
+            session_id="s9",
+            user_sub="u1",
+            since=NOW - timedelta(minutes=15),
+            limit=10,
+        )
+        assert [link.call_id for link in links] == ["recent"]
+
+    def test_limit_keeps_the_most_recent(self):
+        links = select_calls(
+            [cand(str(i), i) for i in range(10)],
+            session_id="s1",
+            user_sub=None,
+            since=NOW,
+            limit=3,
+        )
+        assert [link.call_id for link in links] == ["2", "1", "0"]
+
+    def test_duplicates_counted_once_and_naive_times_treated_as_utc(self):
+        naive = Candidate(
+            "a",
+            "t",
+            (NOW - timedelta(minutes=1)).replace(tzinfo=None),
+            True,
+            "s1",
+            "u1",
+        )
+        links = select_calls(
+            [naive, cand("a", 1)], session_id="s1", user_sub=None, since=NOW, limit=5
+        )
+        assert len(links) == 1 and links[0].started_at.tzinfo is not None
+
+    def test_no_session_or_user_means_no_links(self):
+        assert (
+            select_calls(
+                [cand("a", 1)], session_id=None, user_sub=None, since=NOW, limit=5
+            )
+            == []
+        )
+
+
+def build(sinks=None, user="u1", **kwargs):
+    app = FastMCP("correlated")
+
+    @app.tool
+    def echo(text: str) -> str:
+        return text
+
+    @app.tool
+    def render(scene: str) -> str:
+        raise RuntimeError(f"render of {scene} hung")
+
+    captured = {}
+
+    @app.tool
+    async def report(ctx: Context) -> list:
+        links = await mw.capture_recent_calls(ctx)
+        captured["links"] = links
+        return [link.call_id for link in links]
+
+    identity = {"user": user}
+    mw = instrument(
+        app,
+        sinks if sinks is not None else [MemorySink()],
+        identity_resolver=lambda c: {"user_sub": identity["user"]},
+        **kwargs,
+    )
+    return app, mw, captured, identity
+
+
+class TestCaptureRecentCalls:
+    async def test_same_session_calls_including_failures(self):
+        app, mw, captured, _ = build()
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "a"})
+            with contextlib.suppress(ToolError):
+                await client.call_tool("render", {"scene": "castle"})
+            await client.call_tool("report", {})
+
+        links = captured["links"]
+        assert [(link.tool, link.ok, link.rule) for link in links] == [
+            ("echo", True, SAME_CLIENT_RULE),
+            ("render", False, SAME_CLIENT_RULE),
+        ]
+
+    async def test_new_session_per_call_falls_back_to_user_window(self):
+        """PAT/CI clients open a new MCP session for every call."""
+        app, mw, captured, _ = build()
+        for text in ("one", "two"):
+            async with Client(app) as client:
+                await client.call_tool("echo", {"text": text})
+        async with Client(app) as client:
+            await client.call_tool("report", {})
+
+        assert [link.rule for link in captured["links"]] == [RULE_USER_WINDOW] * 2
+
+    async def test_other_users_calls_are_not_linked(self):
+        app, mw, captured, identity = build()
+        identity["user"] = "someone-else"
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "not mine"})
+        identity["user"] = "u1"
+        async with Client(app) as client:
+            await client.call_tool("report", {})
+        assert captured["links"] == []
+
+    async def test_calls_still_queued_are_found(self):
+        release = asyncio.Event()
+
+        class Stalled:
+            async def write(self, records):
+                await release.wait()
+
+        app, mw, captured, _ = build(sinks=[Stalled()])
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "queued"})
+            await client.call_tool("report", {})
+        release.set()
+        assert [link.tool for link in captured["links"]] == ["echo"]
+
+    async def test_database_fallback_after_restart(self, db):
+        app1, mw1, _, _ = build(sinks=[db("restart")])
+        async with Client(app1) as client:
+            await client.call_tool("echo", {"text": "before restart"})
+        await mw1.aclose()
+
+        # A new process: empty in-memory buffer, same database.
+        app2, mw2, captured, _ = build(sinks=[db("restart")])
+        async with Client(app2) as client:
+            await client.call_tool("report", {})
+        await mw2.aclose()
+
+        [link] = captured["links"]
+        assert link.tool == "echo" and link.rule == RULE_USER_WINDOW
+
+    @pytest.mark.skipif(
+        FASTMCP_4, reason="FastMCP 4 Client is sessionless (MCP 2026-07-28)"
+    )
+    async def test_failing_resolver_still_links_by_session(self):
+        app, mw, captured, _ = build()
+        mw.identity_resolver = lambda c: 1 / 0
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "a"})
+            await client.call_tool("report", {})
+        assert [link.rule for link in captured["links"]] == [RULE_SESSION]
+
+    async def test_slow_database_is_skipped(self, db):
+        class SlowDb(DatabaseSink):
+            async def recent_calls(self, **kwargs):
+                await asyncio.sleep(5)
+                return []
+
+        sink = db("slow", SlowDb)
+        app, mw, captured, _ = build(sinks=[sink], correlation_timeout=0.05)
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "a"})
+            await client.call_tool("report", {})
+        await mw.aclose()
+        # The lookup timed out, so capture returned [] rather than hanging.
+        assert captured["links"] == []
+
+    async def test_outside_a_tool_call_returns_empty(self):
+        _, mw, _, _ = build()
+        mw.identity_resolver = None
+        assert await mw.capture_recent_calls() == []
+
+
+class TestLinksAndContext:
+    async def test_link_and_read_back_with_external_ids(self, db):
+        sink = db("links")
+        app, mw, captured, _ = build(sinks=[sink])
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "a"})
+            with contextlib.suppress(ToolError):
+                await client.call_tool("render", {"scene": "castle"})
+            await client.call_tool("report", {})
+
+        assert await mw.link_feedback("bug-7Q2X", captured["links"])
+        await mw.flush()
+        context = await mw.feedback_context("bug-7Q2X")
+        await mw.aclose()
+
+        assert [(c["position"], c["tool"], c["ok"], c["rule"]) for c in context] == [
+            (0, "echo", True, SAME_CLIENT_RULE),
+            (1, "render", False, SAME_CLIENT_RULE),
+        ]
+        assert context[1]["error_type"] == "RuntimeError"
+
+    async def test_relinking_replaces_links(self, db):
+        sink = db("relink")
+        app, mw, captured, _ = build(sinks=[sink])
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "a"})
+            await client.call_tool("report", {})
+        links = captured["links"]
+        await mw.link_feedback("fb-1", links)
+        await mw.link_feedback("fb-1", links)
+        assert len(await mw.feedback_context("fb-1")) == 1
+        await mw.aclose()
+
+    async def test_context_fills_unflushed_calls_from_memory(self, db):
+        release = asyncio.Event()
+
+        class StalledDb(DatabaseSink):
+            async def write(self, records):
+                await release.wait()
+
+        sink = db("unflushed", StalledDb)
+        app, mw, captured, _ = build(sinks=[sink])
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "a"})
+            await client.call_tool("report", {})
+        await mw.link_feedback("fb-2", captured["links"])
+
+        [row] = await mw.feedback_context("fb-2")
+        release.set()
+        await mw.aclose()
+        assert row["tool"] == "echo" and row["ok"] is True
+
+    async def test_link_without_database_sink_is_false(self):
+        _, mw, _, _ = build()
+        assert await mw.link_feedback("x", []) is False
+        assert await mw.feedback_context("x") == []
+
+
+class TestSubmitFeedbackAutoLinks:
+    async def test_submission_links_preceding_calls(self, db):
+        app = FastMCP("app")
+
+        @app.tool
+        def render(scene: str) -> str:
+            raise RuntimeError("hung")
+
+        sink = db("submit")
+        mw = instrument(app, [sink], identity_resolver=lambda c: {"user_sub": "u1"})
+        add_feedback_tools(app, database_url="sqlite:///:memory:", instrumentation=mw)
+
+        async with Client(app) as client:
+            with contextlib.suppress(ToolError):
+                await client.call_tool("render", {"scene": "castle"})
+            result = await client.call_tool(
+                "submit_feedback",
+                {
+                    "request": {
+                        "type": "bug",
+                        "title": "render hangs",
+                        "description": "the render tool hung",
+                        "submitter": "model",
+                    }
+                },
+            )
+
+        assert result.data["linked_calls"] == 1
+        await mw.flush()
+        [row] = await mw.feedback_context(result.data["feedback_id"])
+        await mw.aclose()
+        assert (row["tool"], row["ok"], row["error_type"]) == (
+            "render",
+            False,
+            "RuntimeError",
+        )
+
+    async def test_without_instrumentation_response_unchanged(self):
+        app = FastMCP("plain")
+        add_feedback_tools(app, database_url="sqlite:///:memory:")
+        async with Client(app) as client:
+            result = await client.call_tool(
+                "submit_feedback",
+                {
+                    "request": {
+                        "type": "bug",
+                        "title": "t",
+                        "description": "no instrumentation",
+                        "submitter": "m",
+                    }
+                },
+            )
+        assert "linked_calls" not in result.data
+
+
+class FakeContext:
+    def __init__(self, transport=None, session_id="ctx-session"):
+        self.transport = transport
+        self.session_id = session_id
+
+
+class TestSessionKey:
+    def test_http_session_header_wins(self, monkeypatch):
+        import fastmcp.server.dependencies as deps
+
+        monkeypatch.setattr(
+            deps,
+            "get_http_headers",
+            lambda include_all=False: {"mcp-session-id": "hdr-1"},
+        )
+        assert (
+            mw_module.session_key(FakeContext(transport="streamable-http")) == "hdr-1"
+        )
+
+    def test_stdio_uses_one_key_per_process(self):
+        a = mw_module.session_key(FakeContext(transport="stdio", session_id="x"))
+        b = mw_module.session_key(FakeContext(transport="stdio", session_id="y"))
+        assert a == b and a.startswith("stdio:")
+
+    def test_falls_back_to_ctx_session_id(self):
+        assert mw_module.session_key(FakeContext()) == "ctx-session"
+
+    def test_no_context(self):
+        assert mw_module.session_key(None) is None
+
+
+class TestStdioEndToEnd:
+    async def test_same_stdio_client_links_by_session(self):
+        """On a real stdio transport, one client's calls link by session on every FastMCP version."""
+        server = (
+            Path(__file__).parent.parent / "fixtures" / "stdio_correlation_server.py"
+        )
+        transport = PythonStdioTransport(str(server), python_cmd=sys.executable)
+        async with Client(transport) as client:
+            await client.call_tool("echo", {"text": "a"})
+            await client.call_tool("echo", {"text": "b"})
+            result = await client.call_tool("report", {})
+        pairs = json.loads(result.content[0].text)
+        assert pairs == [["echo", RULE_SESSION], ["echo", RULE_SESSION]]

tests/unit/test_instrumentation.py

@@ -408,7 +408,7 @@ class TestDatabaseSink:
         async with engine.connect() as conn:
             tables = await conn.run_sync(lambda c: inspect(c).get_table_names())
         await engine.dispose()
-        assert tables == ["qa_tool_calls"]
+        assert sorted(tables) == ["qa_feedback_call_links", "qa_tool_calls"]
 
     async def test_no_tables_created_by_default(self, tmp_path):
         engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
@@ -444,7 +444,7 @@ class TestDatabaseSink:
         host = MetaData()
         assert build_metadata("ffb_", host) is host
         build_metadata("ffb_", host)
-        assert list(host.tables) == ["ffb_tool_calls"]
+        assert sorted(host.tables) == ["ffb_feedback_call_links", "ffb_tool_calls"]
 
 
 class TestIntegration:

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.2"
+version = "2026.9.28"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },