Release 2026.09.27.3: feedback-to-call correlation

Changelog entry and version bump for linking feedback items to the tool
calls that preceded them. README documents the instrumentation parameter
of add_feedback_tools and the new feedback_call_links table that Alembic
hosts need to migrate.

CHANGELOG.md

@@ -5,6 +5,54 @@ 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.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
+- `DatabaseSink` stored missing `args`/`result` (every `meta` mode row) as the
+  JSON literal `null`, so `WHERE args IS NULL` matched nothing. They are now
+  SQL `NULL`. No schema change; rows written by 2026.09.27.1 keep JSON `null`
+  and can be backfilled with
+  `UPDATE ffb_tool_calls SET args = NULL WHERE args = 'null'::jsonb` (same for
+  `result`). Reported from production by blender-mcp.
+
 ## [2026.09.27.1] - 2026-09-27
 
 Per-call instrumentation for any FastMCP server.

README.md

@@ -223,6 +223,63 @@ 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["ok"], 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.
+
 ### Middleware order and shutdown
 
 FastMCP runs the first middleware added as the outermost. Call `instrument()`
@@ -297,7 +354,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 +364,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,7 +390,7 @@ 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:
+See the [fastmcp-feedback-showcase](https://git.supported.systems/fastmcp-feedback/fastmcp-example) repository for examples of every feature:
 
 ```bash
 uvx fastmcp-feedback-showcase  # Run full demo

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "fastmcp-feedback"
-version = "2026.09.27.1"  # YYYYMMDD.patch versioning - update for each release
+version = "2026.09.27.3"  # 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.1"
+__version__ = "2026.09.27.3"
 __author__ = "Ryan Malloy"
 __email__ = "ryan@supported.systems"
 

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

@@ -6,17 +6,27 @@ import json
 import logging
 import os
 import time
+import uuid
+from collections import deque
 from collections.abc import Awaitable, Callable, Iterable, Sequence
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
 from fastmcp import FastMCP
 from fastmcp.server.middleware import Middleware, MiddlewareContext
 
+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 +53,35 @@ 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 _result_payload(result: Any) -> Any:
     structured = getattr(result, "structured_content", None)
     if structured is not None:
@@ -88,6 +127,10 @@ 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.
     """
 
     def __init__(
@@ -105,6 +148,8 @@ 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,
     ):
         mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
         if mode not in MODES:
@@ -117,6 +162,8 @@ 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.recent: deque[ToolCallRecord] = deque(maxlen=recent_calls)
         self.dispatcher = RecordDispatcher(
             sinks if sinks is not None else [JsonLinesSink()],
             max_queue=max_queue,
@@ -144,6 +191,7 @@ class InstrumentationMiddleware(Middleware):
                 record = await self._build_record(
                     context, started_at, duration_ms, result, error
                 )
+                self.recent.append(record)
                 self.dispatcher.submit(record)
             except Exception:
                 logger.warning("Failed to record tool call", exc_info=True)
@@ -201,7 +249,7 @@ class InstrumentationMiddleware(Middleware):
             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"),
@@ -217,6 +265,140 @@ class InstrumentationMiddleware(Middleware):
             extra=extra,
         )
 
+    @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 +408,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/storage.py

@@ -1,11 +1,14 @@
-"""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
 from collections.abc import Sequence
+from datetime import datetime
+from typing import Any
 
 from sqlalchemy import (
     JSON,
@@ -18,6 +21,10 @@ from sqlalchemy import (
     String,
     Table,
     Text,
+    and_,
+    delete,
+    or_,
+    select,
 )
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
@@ -26,44 +33,57 @@ from .record import ToolCallRecord
 
 DEFAULT_TABLE_PREFIX = "ffb_"
 
-_JSON = JSON().with_variant(JSONB(), "postgresql")
+# none_as_null: a missing payload is SQL NULL, so "WHERE args IS NULL" works.
+_JSON = JSON(none_as_null=True).with_variant(JSONB(none_as_null=True), "postgresql")
 
 
 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),
+            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),
+        )
+    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
 
 
@@ -75,8 +95,8 @@ 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``.
     """
 
@@ -91,23 +111,104 @@ class DatabaseSink:
         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()
 
     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])
 
+    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.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:
             await self.engine.dispose()

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")
@@ -425,11 +425,26 @@ class TestDatabaseSink:
         assert tables == []
         assert mw.dispatcher.sink_errors == 1
 
+    async def test_absent_payloads_are_sql_null(self, tmp_path):
+        """Meta mode stores no payloads; they must be SQL NULL, not JSON null."""
+        db = DatabaseSink(f"sqlite+aiosqlite:///{tmp_path}/n.db", create_tables=True)
+        app, _, mw = make_app(sinks=[db])
+        await call(app, "echo", {"text": "hi"})
+        await mw.flush()
+
+        t = db.table
+        async with db.engine.connect() as conn:
+            nulls = (
+                await conn.execute(select(t.c.args.is_(None), t.c.result.is_(None)))
+            ).one()
+        await mw.aclose()
+        assert tuple(nulls) == (True, True)
+
     def test_build_metadata_joins_host_metadata(self):
         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:

uv.lock

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