Merge fix/closed-dispatcher: the dispatcher stays closed after aclose()

A record submitted after aclose(), from the loop or a thread, rebound the
dispatcher: a new queue and task that wrote to the sinks just closed. Now
aclose() closes it for good. Later records are dropped and counted under
"dispatcher closed" (new drop_reasons), with one warning; it never binds a
loop or starts a task again, and aclose() twice is a no-op. Calls made
after close are not kept in recent for linking.

flush() also waits for a record put just after queue.join() woke, which a
single join missed: a record a thread handed over just before aclose()
is delivered if it lands during the flush, else dropped and counted.

CHANGELOG.md

@@ -5,6 +5,97 @@ 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.10.05] - 2026-10-05
+
+A closed middleware stays closed.
+
+### Fixed
+- After `aclose()`, a later `record_event` or tool call record rebuilt the
+  dispatcher's queue and background task and wrote to the sinks that had just
+  been closed. The dispatcher now stays closed: every later record, from the
+  loop or another thread, is dropped and counted in `dispatcher.dropped` and
+  the new `dispatcher.drop_reasons` under `"dispatcher closed"`, with a single
+  warning; it never binds a loop or starts a task again. `record_event`
+  returns False. Tool calls still run normally, and their records are not
+  kept in memory for feedback linking.
+- `flush()` (and so `aclose()`) could return while a record put just after
+  `asyncio.Queue.join()` woke was still queued. A record a thread hands over
+  just before `aclose()` is now delivered if it lands while the flush runs,
+  or dropped and counted if it lands after; never written after close.
+
+### Changed
+- `aclose()` is idempotent: a second call does nothing, and sinks are closed
+  once.
+
+## [2026.10.04] - 2026-10-04
+
+Bounded payloads, and classifier keys that reach into nested results.
+
+### Added
+- `max_payload_bytes` on `InstrumentationMiddleware` and `instrument()`,
+  64 KiB by default, `None` for no limit. Arguments and results stored in
+  mode `full`, and every event's `attrs` (including `record_llm_call`'s),
+  are bounded by it. A larger value is stored as
+  `{"_truncated": true, "size": <bytes>, "preview": "<start of its JSON>"}`;
+  the preview is 1000 characters (a quarter of the limit when smaller) and
+  is taken after redaction. Call payloads reuse the size already measured
+  for `args_size` and `result_size`, so the limit adds no encoding; the
+  preview encodes only the start of the value. In `attrs`, only the largest
+  values are replaced, so the dict keeps its keys. `EmbeddingSink` embeds a
+  marker's preview. `fastmcp_feedback.instrumentation.payload` holds the
+  marker helpers (`TRUNCATED`, `is_truncated`).
+- `message_keys` on `soft_error_classifier` (default `error`, `message`,
+  `detail`, `reason`, as before).
+- Dotted paths in `message_keys` and `error_keys`: `"snapshot.error"` reads
+  `result["snapshot"]["error"]`. A key that literally contains a dot is
+  matched first; a missing step or a non-dict step counts as absent and
+  never raises. Default outputs are unchanged.
+
+### Changed
+- Stored payloads and event attrs over 64 KiB become markers. Earlier
+  releases stored them whole; `max_payload_bytes=None` keeps that.
+
+## [2026.10.02.1] - 2026-10-02
+
+The identity resolver's second run is opt-in again for any resolver that
+answers.
+
+### Changed
+- The second resolver run after the tool now happens only when the first
+  answer was empty (no keys, only None values, or a run that failed or timed
+  out) or lacks a key in `late_identity_keys`. In 2026.10.02 a missing
+  `user_sub`, `caller_kind` or `client_id` also triggered it, so a resolver
+  that never returns one of them (mcspeak's returns no `client_id`) ran twice
+  on every kept call. `late_identity_keys` stays empty by default: a resolver
+  that answers anything runs once per call unless the server opts in.
+  blender-mcp's registration fix still needs
+  `late_identity_keys=["bus_client_uuid"]`, as in 2026.10.02.
+
+## [2026.10.02] - 2026-10-02
+
+A tool that creates its caller's identity records it on its own row again.
+
+### Fixed
+- Regression in 2026.10.01.4: with the identity resolver moved before the
+  tool, a call whose tool creates the identity the resolver looks up (the
+  call that registers a client, or logs in) recorded its own row without
+  it. blender-mcp's `register_client` lost `bus_client_uuid`, `bus_role` and
+  `parent_uuid` from its row's `identity` this way. When the first answer is
+  incomplete, missing `user_sub`, `caller_kind` or `client_id`, or any key in
+  the new `late_identity_keys`, the resolver now runs a second time after the
+  tool, concurrently with the enricher, for calls that are kept (not for
+  sampled-out ones). Its values fill the row's gaps and replace `None`; on
+  conflict the first answer wins, so the row matches the events recorded
+  during the call, which are not rewritten. Both runs keep `hook_timeout`,
+  redaction and never raising. A complete first answer still means one run.
+
+### Added
+- `late_identity_keys` on `InstrumentationMiddleware` and `instrument()`:
+  identity keys that may only exist once the tool has run. A resolver that
+  already returns all three columns before such a tool (blender-mcp's does)
+  needs it to get the second run, e.g.
+  `late_identity_keys=["bus_client_uuid"]`.
+
 ## [2026.10.01.4] - 2026-10-01
 
 Statistics speak the same enum values as the rest of the API, events carry

Makefile

@@ -61,4 +61,6 @@ lint:
 	uv run --no-sync ruff check src/ examples/ tests/_compat.py tests/_databases.py \
 		tests/unit/test_test_databases.py tests/unit/test_instrumentation.py \
 		tests/unit/test_embeddings.py tests/unit/test_backfill.py \
-		tests/unit/test_consistency.py
+		tests/unit/test_consistency.py tests/unit/test_late_identity.py \
+		tests/unit/test_payload_cap.py tests/unit/test_classifier_paths.py \
+		tests/unit/test_closed_dispatcher.py

README.md

@@ -194,6 +194,42 @@ Set it with `mode=` or the `FEEDBACK_INSTRUMENTATION_MODE` environment variable.
 `meta_only_tools={"run_code", "create_token"}` keeps chosen tools at `meta`
 even in `full` mode.
 
+### Payload size limit
+
+`max_payload_bytes` (64 KiB by default, `None` for no limit) bounds what a
+row stores: the arguments and result in `full` mode, and every event's
+`attrs`. A larger value is stored as a marker instead:
+
+```json
+{"_truncated": true, "size": 812345, "preview": "{\"frames\": [[0.1, 0.2], ..."}
+```
+
+`size` is the value's JSON size in bytes, the same measurement as
+`args_size` and `result_size`, which the limit reuses, so it costs no second
+encoding. `preview` is the start of the JSON text, 1000 characters (a quarter
+of the limit when that is smaller), taken after redaction, so a secret cannot
+survive in it. For event `attrs` only the largest values are replaced, until
+the whole fits, so small fields such as `ok` keep their place:
+
+```python
+import asyncio
+from fastmcp_feedback.instrumentation import InstrumentationMiddleware, MemorySink
+
+async def main():
+    sink = MemorySink()
+    mw = InstrumentationMiddleware([sink])
+    mw.record_event("job.finished", key="j1", attrs={"ok": False, "error": "E" * 100_000})
+    await mw.aclose()
+    attrs = sink.events[0].attrs
+    print(attrs["ok"], attrs["error"]["_truncated"], attrs["error"]["size"])  # False True 100002
+
+asyncio.run(main())
+```
+
+**Upgrading to 2026.10.04:** before this release payloads and attrs were
+stored whole. A row that used to hold more than 64 KiB there now holds the
+marker; pass `max_payload_bytes=None` to keep storing everything.
+
 ### Redaction
 
 Secrets are removed from arguments, results, error messages and hook output
@@ -311,6 +347,25 @@ instrument(app, result_classifier=soft_error_classifier(error_keys=("failure_rea
 instrument(app, result_classifier=soft_error_classifier(error_keys=()))
 ```
 
+`message_keys` sets where the message comes from (`error`, `message`,
+`detail`, `reason` by default), and both `message_keys` and `error_keys`
+take dotted paths into nested dicts. A key that literally contains a dot is
+tried first; a missing step, or one that is not a dict, counts as absent:
+
+```python
+from fastmcp_feedback.instrumentation import soft_error_classifier
+
+# mcardupilot: a wait that returned because the simulator died, cause nested
+classify = soft_error_classifier(
+    status_keys=("status", "state", "reason"),
+    extra_statuses=("died",),
+    message_keys=("snapshot.error", "error", "message", "detail", "reason"),
+)
+died = {"reason": "died", "snapshot": {"error": "SITL exited: segfault"}}
+assert classify("wait_until", died) == "reason=died: SITL exited: segfault"
+assert classify("wait_until", {"reason": "timeout", "snapshot": {}}) is None
+```
+
 Add your own status values, or pass any `(tool, result) -> str | None`:
 
 ```python
@@ -382,15 +437,43 @@ 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.
 
-The identity resolver runs once per call, **before** the tool. Events the
-tool records get the same `user_sub`, `caller_kind` and `client_id` as the
-call's row, and the call's record reuses the result instead of asking again.
-So keep it cheap: an async resolver can hold the tool back by up to
-`hook_timeout`, and it also runs for calls that sampling later drops (sampling
-cannot know beforehand whether a call will fail). It does not run in mode
-`off` or for `exclude_tools`. The enricher still runs after the call, since it
-receives the result. Before 2026.10.01.4 the resolver ran after the call and
-events never carried identity.
+The identity resolver runs **before** the tool. Events the tool records get
+the same `user_sub`, `caller_kind` and `client_id` as the call's row. So keep
+it cheap: an async resolver can hold the tool back by up to `hook_timeout`,
+and it also runs for calls that sampling later drops (sampling cannot know
+beforehand whether a call will fail). It does not run in mode `off` or for
+`exclude_tools`. The enricher still runs after the call, since it receives
+the result.
+
+Some tools create the identity they run under: a registration that makes the
+client record the resolver looks up, or a login. The resolver runs a second
+time after the tool, alongside the enricher and only for calls that are kept,
+when its first answer was empty (nothing, only None values, or a run that
+failed or timed out) or lacks a key named in `late_identity_keys`. A missing
+`user_sub`, `caller_kind` or `client_id` alone does not trigger it, and
+`late_identity_keys` is empty by default, so a resolver that answers anything
+runs once per call unless you opt in.
+The call's own row takes the second answer's values where the first had
+none; where both have a value, the first wins, so the row agrees with the
+events recorded during the call. Those events are not rewritten. A complete
+first answer means one run.
+
+```python
+registered: dict[str, str] = {}  # session -> bus client, filled by a tool
+
+def who(context):
+    out = {"user_sub": "user-1", "caller_kind": "addon", "client_id": "dcr-9"}
+    session = getattr(context.fastmcp_context, "session_id", None)
+    if session in registered:
+        out["bus_client_uuid"] = registered[session]
+    return out
+
+instrument(app, identity_resolver=who, late_identity_keys=["bus_client_uuid"])
+```
+
+Without `late_identity_keys` this resolver's first answer is not empty, so
+there is no second run and the call that registers the client is recorded
+without `bus_client_uuid`.
 
 ### Linking feedback to the calls behind it
 
@@ -917,7 +1000,12 @@ error. `--help` lists the rest (`--until`, `--limit`, `--batch-size`,
 FastMCP runs the first middleware added as the outermost. Call `instrument()`
 before adding other middleware to include their time and record calls they
 reject; call it last to time only the tool. On shutdown, `await mw.aclose()`
-flushes queued records and closes the sinks.
+flushes queued records and closes the sinks. After that the middleware
+records nothing: tool calls still run, and a late `record_event` (from a
+worker thread finishing up, say) returns False. Those records are dropped and
+counted in `mw.dispatcher.drop_reasons["dispatcher closed"]`, never written,
+so wait for your own threads before closing. Calling `aclose()` again does
+nothing.
 
 ## Logging
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "fastmcp-feedback"
-version = "2026.10.01.4"  # YYYYMMDD.patch versioning - update for each release
+version = "2026.10.05"  # 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

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

src/fastmcp_feedback/instrumentation/classify.py

@@ -19,14 +19,36 @@ 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)
+def lookup(result: dict, key: str) -> Any:
+    """``result[key]``, where ``key`` may be a dotted path into nested dicts.
+
+    A key present exactly as written wins, so a literal ``"a.b"`` key is
+    found before ``result["a"]["b"]``. Otherwise ``"snapshot.error"`` walks
+    ``result["snapshot"]["error"]``. A missing step, or a step that is not a
+    dict, gives None: the same as the key not being there. Never raises.
+    """
+    if key in result:
+        return result[key]
+    if "." not in key:
+        return None
+    value: Any = result
+    for part in key.split("."):
+        if not isinstance(value, dict) or part not in value:
+            return None
+        value = value[part]
+    return value
+
+
+def _message(
+    result: dict, depth: int = 0, keys: tuple[str, ...] = MESSAGE_KEYS
+) -> str | None:
+    for key in keys:
+        value = lookup(result, key)
         if value is None or value == "":
             continue
         if isinstance(value, dict):
             # {"error": {"code": 404, "message": "..."}}
-            if depth == 0 and (inner := _message(value, 1)):
+            if depth == 0 and (inner := _message(value, 1, keys)):
                 return inner
             continue
         if isinstance(value, list | tuple):
@@ -59,14 +81,17 @@ def soft_error_classifier(
     flag_keys: Iterable[str] = ("ok", "success"),
     extra_statuses: Iterable[str] = (),
     error_keys: Iterable[str] = ("error",),
+    message_keys: Iterable[str] = MESSAGE_KEYS,
 ) -> 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``.
+    ``isError``. The message is the first of ``message_keys`` present
+    (``error``, ``message``, ``detail``, ``reason`` by default), e.g.
+    ``status=failed: disk full``. A dict found there is searched one level
+    down with the same keys, as in ``{"error": {"message": "..."}}``.
 
     Failing those, it is a soft error when any of ``error_keys`` holds a
     non-empty string, as in ``{"played": False, "error": "output stalled"}``,
@@ -81,6 +106,13 @@ def soft_error_classifier(
     ``isError`` still flag it. Truthy non-bools such as ``1`` or ``"yes"``
     are not an explicit success.
 
+    ``message_keys`` and ``error_keys`` accept dotted paths into nested
+    dicts: ``"snapshot.error"`` reads ``result["snapshot"]["error"]``. A key
+    that literally contains a dot is matched first; a missing step or one
+    that is not a dict counts as the key not being there. The message label
+    for an ``error_keys`` hit is the path as written. ``status_keys`` and
+    ``flag_keys`` are plain keys.
+
     >>> classify = soft_error_classifier(extra_statuses=("unknown_target",))
     >>> classify("render", {"status": "unknown_target", "error": "no such object"})
     'status=unknown_target: no such object'
@@ -88,11 +120,22 @@ def soft_error_classifier(
     'error: output stalled'
     >>> print(classify("speak", {"ok": True, "error": "deprecated arg"}))
     None
+    >>> sitl = soft_error_classifier(
+    ...     status_keys=("status", "state", "reason"),
+    ...     extra_statuses=("died",),
+    ...     message_keys=("snapshot.error", "error", "message", "detail", "reason"),
+    ... )
+    >>> sitl("wait_until", {"reason": "died", "snapshot": {"error": "SITL exited"}})
+    'reason=died: SITL exited'
     """
     bad = frozenset(s.lower() for s in (*statuses, *extra_statuses))
     status_keys = tuple(status_keys)
     flag_keys = tuple(flag_keys)
     error_keys = tuple(error_keys)
+    message_keys = tuple(message_keys)
+
+    def message(result: dict) -> str | None:
+        return _message(result, 0, message_keys)
 
     def classify(tool: str, result: Any) -> str | None:
         if not isinstance(result, dict):
@@ -100,18 +143,18 @@ def soft_error_classifier(
         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))
+                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))
+                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"))
+                detail = message(result) or _content_text(result.get("content"))
                 return _describe("isError", detail)
         if any(result.get(key) is True for key in flag_keys):
             return None
         for key in error_keys:
-            value = result.get(key)
+            value = lookup(result, key)
             if isinstance(value, str) and value.strip():
                 return _describe(key, value.strip())
         return None

src/fastmcp_feedback/instrumentation/dispatch.py

@@ -10,6 +10,8 @@ from .sinks import Record, Sink
 
 logger = logging.getLogger(__name__)
 
+CLOSED = "dispatcher closed"
+
 
 class RecordDispatcher:
     """Queue records and hand them to sinks from a background task.
@@ -17,6 +19,10 @@ class RecordDispatcher:
     ``submit`` never blocks and never raises: when the queue is full the record
     is dropped and counted, so a slow or broken sink cannot slow down or fail
     a tool call. Sink errors are logged and counted, never propagated.
+
+    After ``aclose()`` the dispatcher stays closed: every later record, from
+    the loop or another thread, is dropped and counted under ``CLOSED``, and
+    it never binds a loop or starts a task again.
     """
 
     def __init__(
@@ -32,15 +38,26 @@ class RecordDispatcher:
         self.batch_size = batch_size
         self.sink_timeout = sink_timeout
         self.dropped = 0
+        # Drops per reason, e.g. {"queue full": 3, "dispatcher closed": 1}.
+        self.drop_reasons: dict[str, int] = {}
         self.sink_errors = 0
+        self._closed = False
         self._queue: asyncio.Queue[Record] | None = None
         self._task: asyncio.Task | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
         # submit_threadsafe can drop from any thread.
         self._drop_lock = threading.Lock()
 
+    @property
+    def closed(self) -> bool:
+        """True once ``aclose()`` has flushed; later records are dropped."""
+        return self._closed
+
     def submit(self, record: Record) -> bool:
         """Queue a record for delivery. Returns False if it was dropped."""
+        if self._closed:
+            self._drop(CLOSED)
+            return False
         try:
             loop = asyncio.get_running_loop()
         except RuntimeError:
@@ -60,6 +77,8 @@ class RecordDispatcher:
 
         Called by the middleware as each tool call starts. Never raises.
         """
+        if self._closed:
+            return
         with contextlib.suppress(Exception):
             self._bind(asyncio.get_running_loop())
 
@@ -83,7 +102,13 @@ class RecordDispatcher:
         it was handed over, and a queue that fills before it lands still drops
         and counts it. With no running loop anywhere the record is dropped.
         ``on_queued(record)`` runs on the loop thread once the record is queued.
+
+        A record handed over before ``aclose()`` whose turn on the loop comes
+        after the dispatcher closed is dropped and counted there, not written.
         """
+        if self._closed:
+            self._drop(CLOSED)
+            return False
         try:
             try:
                 running = asyncio.get_running_loop()
@@ -123,6 +148,16 @@ class RecordDispatcher:
         with self._drop_lock:
             self.dropped += 1
             dropped = self.dropped
+            for_reason = self.drop_reasons.get(reason, 0) + 1
+            self.drop_reasons[reason] = for_reason
+        if reason == CLOSED:
+            # Late records at shutdown are expected; say so once.
+            if for_reason == 1:
+                logger.warning(
+                    "Dropped instrumentation record recorded after aclose(); "
+                    "later ones are dropped silently (see drop_reasons)"
+                )
+            return
         if dropped == 1 or dropped % 1000 == 0:
             logger.warning(
                 "Dropped instrumentation record (%s); %d dropped so far",
@@ -158,13 +193,33 @@ class RecordDispatcher:
                 )
 
     async def flush(self) -> None:
-        """Wait until every queued record has been handed to the sinks."""
-        if self._queue is not None and self._loop is asyncio.get_running_loop():
-            await self._queue.join()
+        """Wait until every queued record has been handed to the sinks.
+
+        Records queued while it waits, from this loop or handed over from a
+        thread, are waited for too.
+        """
+        queue = self._queue
+        if queue is None or self._loop is not asyncio.get_running_loop():
+            return
+        await queue.join()
+        # join() can return while a record put just after its wake-up is
+        # still queued, so check again rather than trusting one join.
+        while _unfinished(queue):
+            await queue.join()
 
     async def aclose(self) -> None:
-        """Flush, stop the background task and close the sinks."""
+        """Flush, stop the background task and close the sinks. Idempotent.
+
+        Records queued before this returns its flush are delivered; from then
+        on the dispatcher is closed and drops everything (``CLOSED``).
+        """
+        if self._closed:
+            return
         await self.flush()
+        # No await between the flush ending and this: a record handed over
+        # from a thread that lands later is dropped by submit, never queued
+        # for a task that is about to stop.
+        self._closed = True
         if self._task is not None:
             self._task.cancel()
             with contextlib.suppress(asyncio.CancelledError, RuntimeError):
@@ -180,3 +235,12 @@ class RecordDispatcher:
                     logger.warning(
                         "Sink %s failed to close", type(sink).__name__, exc_info=True
                     )
+
+
+def _unfinished(queue: asyncio.Queue) -> bool:
+    """Whether ``queue`` holds records not yet marked done.
+
+    ``_unfinished_tasks`` is the counter ``join()`` itself waits on (CPython
+    3.11 to 3.14). Without it, report False: flush then does a single join.
+    """
+    return getattr(queue, "_unfinished_tasks", 0) > 0

src/fastmcp_feedback/instrumentation/embeddings.py

@@ -22,6 +22,7 @@ from sqlalchemy import Float, and_, literal, not_, select, tuple_
 
 from .constants import DEFAULT_EVENT_TEXT_KEYS, EMBEDDING_SOURCES
 from .events import EventRecord
+from .payload import is_truncated
 from .record import ToolCallRecord
 from .redaction import Redactor
 from .sinks import Record
@@ -158,6 +159,10 @@ def _clean(value: Any, depth: int = 0) -> str | None:
     """
     if value is None:
         return None
+    if is_truncated(value):
+        # A value over the middleware's max_payload_bytes: its start is all
+        # that was kept.
+        value = value.get("preview")
     if isinstance(value, str):
         text = value
     elif isinstance(value, list | tuple | dict):

src/fastmcp_feedback/instrumentation/middleware.py

@@ -4,7 +4,6 @@ import asyncio
 import contextlib
 import contextvars
 import inspect
-import json
 import logging
 import os
 import random as _random
@@ -35,6 +34,7 @@ from .events import (
     EventRecord,
     jsonable,
 )
+from .payload import DEFAULT_MAX_PAYLOAD_BYTES, cap, cap_attrs, json_size
 from .record import ToolCallRecord
 from .redaction import Redactor
 from .sinks import JsonLinesSink, Sink
@@ -114,13 +114,6 @@ def _result_payload(result: Any) -> Any:
     return result
 
 
-def _json_size(value: Any) -> int | None:
-    try:
-        return len(json.dumps(value, default=str).encode())
-    except Exception:
-        return None
-
-
 class InstrumentationMiddleware(Middleware):
     """Record every tool call without ever slowing or breaking it.
 
@@ -140,11 +133,20 @@ class InstrumentationMiddleware(Middleware):
         meta_only_tools: tools that are never recorded above ``meta``.
         identity_resolver: ``(context) -> dict`` returning ``user_sub``,
             ``caller_kind`` and ``client_id`` (stored as columns) plus any
-            other keys (stored in ``identity``). May be async. Runs once per
-            recorded call, before the tool, so events recorded during the
-            call get the same identity. It also runs for calls that sampling
-            later drops, and an async one can delay the tool by up to
-            ``hook_timeout``; keep it cheap.
+            other keys (stored in ``identity``). May be async. Runs before
+            the tool, so events recorded during the call get that identity;
+            it also runs for calls that sampling later drops, and an async
+            one can delay the tool by up to ``hook_timeout``; keep it cheap.
+            When that first answer is empty (nothing, or only None values,
+            including a run that failed or timed out) or lacks any of
+            ``late_identity_keys``, it runs again after the tool, alongside
+            the enricher, and the call's own record takes the second
+            answer's values where the first had none. Any other answer means
+            one run, even without all three columns.
+        late_identity_keys: identity keys that may only exist once the tool
+            has run, such as an id a registration or login tool creates. A
+            first answer without one of them gets the second run. Empty by
+            default.
         enricher: ``(tool, args, result, context) -> dict`` of extra fields,
             stored in ``extra``. A ``server_version`` key fills that column.
             May be async.
@@ -175,6 +177,11 @@ class InstrumentationMiddleware(Middleware):
             ``prompt``/``completion`` keys passed in its ``attrs``.
         max_llm_text_chars: longest prompt or completion kept. Defaults to
             ``8 * max_error_chars`` (4000).
+        max_payload_bytes: largest stored arguments, result (mode ``full``)
+            or event ``attrs``, in bytes of JSON; None for no limit. Larger
+            ones are stored as ``{"_truncated": true, "size": ..., "preview":
+            ...}`` markers, after redaction; for attrs, only the largest
+            values are replaced. See ``fastmcp_feedback.instrumentation.payload``.
 
     Exclusion and sampling apply to tool calls only; events are always
     recorded. A sampled tool's call is kept whenever an event was recorded
@@ -189,6 +196,7 @@ class InstrumentationMiddleware(Middleware):
         redactor: Redactor | None = None,
         meta_only_tools: Iterable[str] = (),
         identity_resolver: IdentityResolver | None = None,
+        late_identity_keys: Iterable[str] = (),
         enricher: Enricher | None = None,
         server_version: str | None = None,
         hook_timeout: float = 0.25,
@@ -205,6 +213,7 @@ class InstrumentationMiddleware(Middleware):
         recent_events: int = 2000,
         capture_llm_text: bool = False,
         max_llm_text_chars: int | None = None,
+        max_payload_bytes: int | None = DEFAULT_MAX_PAYLOAD_BYTES,
     ):
         mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
         if mode not in MODES:
@@ -213,6 +222,7 @@ class InstrumentationMiddleware(Middleware):
         self.redactor = redactor or Redactor()
         self.meta_only_tools = frozenset(meta_only_tools)
         self.identity_resolver = identity_resolver
+        self.late_identity_keys = tuple(late_identity_keys)
         self.enricher = enricher
         self.server_version = server_version
         self.hook_timeout = hook_timeout
@@ -236,6 +246,16 @@ class InstrumentationMiddleware(Middleware):
             contextvars.ContextVar(f"ffb_call_scope_{id(self):x}", default=None)
         )
         self.capture_llm_text = capture_llm_text
+        if max_payload_bytes is not None and (
+            isinstance(max_payload_bytes, bool)
+            or not isinstance(max_payload_bytes, int)
+            or max_payload_bytes <= 0
+        ):
+            raise ValueError(
+                f"max_payload_bytes must be a positive int or None, "
+                f"got {max_payload_bytes!r}"
+            )
+        self.max_payload_bytes = max_payload_bytes
         self.max_llm_text_chars = (
             max_llm_text_chars
             if max_llm_text_chars is not None
@@ -273,9 +293,9 @@ class InstrumentationMiddleware(Middleware):
             scope = CallScope(
                 id=str(uuid.uuid4()),
                 session_id=session_key(context.fastmcp_context),
-                user_sub=who.columns.get("user_sub"),
-                caller_kind=who.columns.get("caller_kind"),
-                client_id=who.columns.get("client_id"),
+                user_sub=who.column("user_sub"),
+                caller_kind=who.column("caller_kind"),
+                client_id=who.column("client_id"),
             )
             token = self._call_scope.set(scope)
         except Exception:
@@ -332,25 +352,40 @@ class InstrumentationMiddleware(Middleware):
         )
         if scope is not None:
             record.id = scope.id
-        self.recent.append(record)
+        if not self.dispatcher.closed:
+            # After aclose() calls are neither stored nor kept for linking.
+            self.recent.append(record)
         self.dispatcher.submit(record)
 
     async def _resolve_identity(self, context: MiddlewareContext) -> "_Identity":
-        """Run the identity resolver once for a call. Never raises.
+        """Run the identity resolver once. Never raises.
 
         Returns the redacted result split into the three columns and the
-        rest. ``client_id`` falls back to the FastMCP context's.
+        rest, plus the FastMCP context's ``client_id`` as the fallback for a
+        resolver that gives none.
         """
+        fallback = None
         try:
+            fallback = _attr(context.fastmcp_context, "client_id")
             identity = await self._call_hook(self.identity_resolver, context)
             identity = self.redactor.redact(identity)
             columns = {k: identity.pop(k) for k in _IDENTITY_COLUMNS if k in identity}
-            if not columns.get("client_id"):
-                columns["client_id"] = _attr(context.fastmcp_context, "client_id")
-            return _Identity(columns, identity)
+            return _Identity(columns, identity, fallback)
         except Exception:
             logger.warning("Failed to resolve identity", exc_info=True)
-            return _Identity({}, {})
+            return _Identity({}, {}, fallback)
+
+    def _identity_incomplete(self, who: "_Identity") -> bool:
+        """Whether the resolver's first answer lacks something it may know later."""
+        if self.identity_resolver is None:
+            return False
+        answered = any(
+            v is not None for v in (*who.columns.values(), *who.rest.values())
+        )
+        return not answered or any(
+            who.rest.get(k) is None and who.columns.get(k) is None
+            for k in self.late_identity_keys
+        )
 
     async def _call_hook(self, hook: Callable | None, *args: Any) -> dict[str, Any]:
         if hook is None:
@@ -395,9 +430,17 @@ class InstrumentationMiddleware(Middleware):
         args = getattr(context.message, "arguments", None) or {}
         ctx = context.fastmcp_context
 
-        columns = who.columns
+        enrich = self._call_hook(self.enricher, tool, args, payload, context)
+        if self._identity_incomplete(who):
+            # A tool can create the identity it runs under (a registration,
+            # a login). Ask again, concurrently with the enricher so the
+            # record is not built any later than the slower of the two.
+            later, extra = await asyncio.gather(self._resolve_identity(context), enrich)
+            who = who.filled_from(later)
+        else:
+            extra = await enrich
         identity = dict(who.rest)
-        extra = await self._call_hook(self.enricher, tool, args, payload, context)
+        columns = {k: who.column(k) for k in _IDENTITY_COLUMNS}
         extra = self.redactor.redact(extra)
         server_version = extra.pop("server_version", None) or self.server_version
 
@@ -421,6 +464,9 @@ class InstrumentationMiddleware(Middleware):
                 self.redactor.redact_text(soft)[: self.max_error_chars] or None
             )
 
+        # Measured once: the size columns and the payload limit share it.
+        args_size = json_size(args)
+        result_size = json_size(payload) if payload is not None else None
         return ToolCallRecord(
             tool=tool,
             started_at=started_at,
@@ -437,10 +483,14 @@ class InstrumentationMiddleware(Middleware):
             caller_kind=columns.get("caller_kind"),
             identity=identity,
             server_version=server_version,
-            args_size=_json_size(args),
-            result_size=_json_size(payload) if payload is not None else None,
-            args=self.redactor.redact(args) if full else None,
-            result=self.redactor.redact(payload)
+            args_size=args_size,
+            result_size=result_size,
+            args=cap(self.redactor.redact(args), args_size, self.max_payload_bytes)
+            if full
+            else None,
+            result=cap(
+                self.redactor.redact(payload), result_size, self.max_payload_bytes
+            )
             if full and payload is not None
             else None,
             extra=extra,
@@ -474,7 +524,9 @@ class InstrumentationMiddleware(Middleware):
         Returns False when the event was not recorded: mode ``off``, an invalid
         ``kind``, ``key`` or ``attrs``, a full queue, or no event loop to hand
         it to (called from a thread before anything was recorded on the
-        server's loop). From another thread, True means handed to the loop.
+        server's loop), or after ``aclose()``. From another thread, True means
+        handed to the loop; if the middleware closes before the event lands,
+        it is dropped and counted, never written.
         """
         try:
             if self.mode == "off":
@@ -518,7 +570,10 @@ class InstrumentationMiddleware(Middleware):
                 caller_kind=None if caller_kind is None else str(caller_kind),
                 client_id=None if client_id is None else str(client_id),
                 server_version=self.server_version,
-                attrs=jsonable(self.redactor.redact(dict(attrs or {}))),
+                attrs=cap_attrs(
+                    jsonable(self.redactor.redact(dict(attrs or {}))),
+                    self.max_payload_bytes,
+                ),
             )
             return self.dispatcher.submit_threadsafe(record, self.recent_events.append)
         except Exception:
@@ -823,15 +878,53 @@ class InstrumentationMiddleware(Middleware):
         await self.dispatcher.flush()
 
     async def aclose(self) -> None:
-        """Flush and close the sinks. Call on server shutdown."""
+        """Flush and close the sinks. Call on server shutdown. Idempotent.
+
+        Afterwards the middleware records nothing: tool calls still run, but
+        their records, and any ``record_event`` from the loop or a thread, are
+        dropped and counted in ``dispatcher.drop_reasons["dispatcher closed"]``,
+        and they are not kept in memory for linking or ``events_for``.
+        """
         await self.dispatcher.aclose()
 
 
 class _Identity(NamedTuple):
-    """One call's resolved identity: the column values and everything else."""
+    """One answer from the identity resolver, redacted.
 
-    columns: dict[str, str | None]
+    ``columns`` holds whichever of user_sub, caller_kind and client_id it
+    returned, ``rest`` everything else, and ``fallback_client_id`` the
+    FastMCP context's client_id for when it returned none.
+    """
+
+    columns: dict[str, Any]
     rest: dict[str, Any]
+    fallback_client_id: str | None = None
+
+    def column(self, key: str) -> Any:
+        value = self.columns.get(key)
+        if key == "client_id" and not value:
+            return self.fallback_client_id
+        return value
+
+    def filled_from(self, later: "_Identity") -> "_Identity":
+        """This answer, with gaps and None values filled from ``later``.
+
+        Values already here win, so the call's record agrees with the events
+        recorded during the call, which saw only this answer.
+        """
+
+        def fill(first: dict[str, Any], second: dict[str, Any]) -> dict[str, Any]:
+            merged = dict(first)
+            for key, value in second.items():
+                if merged.get(key) is None:
+                    merged[key] = value
+            return merged
+
+        return _Identity(
+            fill(self.columns, later.columns),
+            fill(self.rest, later.rest),
+            self.fallback_client_id or later.fallback_client_id,
+        )
 
 
 def _is_sink(sink: Any, module: str, name: str) -> bool:

src/fastmcp_feedback/instrumentation/payload.py

@@ -0,0 +1,108 @@
+"""Size limits for stored payloads: call arguments and results, event attrs.
+
+A value whose JSON encoding is larger than the limit is stored as a marker:
+
+    {"_truncated": true, "size": 812345, "preview": "{\"frames\": [[0.1, ..."}
+
+``size`` is the value's encoded size in bytes, measured as for ``args_size``
+(``json.dumps(value, default=str)``, non-ASCII escaped). ``preview`` is the
+start of its JSON text, ``preview_chars(limit)`` characters: ``PREVIEW_CHARS``
+(1000) under the default limit, a quarter of the limit under a smaller one,
+so the marker itself fits. It is taken from the value after redaction, so a
+secret cannot reach it. The keys and their meaning are stable.
+"""
+
+import json
+from typing import Any
+
+TRUNCATED = "_truncated"
+PREVIEW_CHARS = 1000
+DEFAULT_MAX_PAYLOAD_BYTES = 64 * 1024
+
+# Incremental, so a preview of a large value encodes only its start.
+_ENCODER = json.JSONEncoder(default=str, ensure_ascii=False)
+
+
+def json_size(value: Any) -> int | None:
+    """Bytes of ``value`` as JSON, or None when it cannot be encoded."""
+    try:
+        return len(json.dumps(value, default=str).encode())
+    except Exception:
+        return None
+
+
+def is_truncated(value: Any) -> bool:
+    """Whether ``value`` is a marker left by the size limit."""
+    return isinstance(value, dict) and value.get(TRUNCATED) is True
+
+
+def preview(value: Any, chars: int = PREVIEW_CHARS) -> str:
+    """The first ``chars`` characters of ``value``'s JSON text."""
+    if chars <= 0:
+        return ""
+    parts: list[str] = []
+    length = 0
+    try:
+        for chunk in _ENCODER.iterencode(value):
+            parts.append(chunk)
+            length += len(chunk)
+            if length >= chars:
+                break
+    except Exception:
+        return ""
+    return "".join(parts)[:chars]
+
+
+def preview_chars(limit: int) -> int:
+    """How much of a value's JSON text a marker keeps under ``limit``."""
+    return min(PREVIEW_CHARS, limit // 4)
+
+
+def marker(value: Any, size: int | None, limit: int) -> dict[str, Any]:
+    """The stand-in stored for a ``value`` over ``limit`` bytes."""
+    return {
+        TRUNCATED: True,
+        "size": size,
+        "preview": preview(value, preview_chars(limit)),
+    }
+
+
+def cap(value: Any, size: int | None, limit: int | None) -> Any:
+    """``value``, or its marker when ``size`` is over ``limit``.
+
+    ``size`` is passed in because the caller has usually measured it already
+    (``args_size``, ``result_size``), so this costs no second encoding.
+    """
+    if limit is None or size is None or size <= limit:
+        return value
+    return marker(value, size, limit)
+
+
+def cap_attrs(attrs: dict[str, Any], limit: int | None) -> dict[str, Any]:
+    """Event attrs within ``limit`` bytes, keeping the dict's keys.
+
+    The largest values are replaced by markers, one at a time, until the
+    whole fits, so small fields (``ok``, ``phase``, ``model``) stay readable.
+    If markers alone cannot get it under the limit (very many keys, or a
+    limit of a few dozen bytes), the whole dict becomes one marker, which
+    under such a tiny limit can itself be larger than the limit.
+    """
+    if limit is None or not attrs:
+        return attrs
+    total = json_size(attrs)
+    if total is None or total <= limit:
+        return attrs
+    sizes = {key: json_size(value) or 0 for key, value in attrs.items()}
+    out = dict(attrs)
+    for key in sorted(sizes, key=sizes.__getitem__, reverse=True):
+        if total <= limit:
+            break
+        stand_in = marker(attrs[key], sizes[key], limit)
+        smaller = sizes[key] - (json_size(stand_in) or 0)
+        if smaller <= 0:
+            continue
+        out[key] = stand_in
+        total -= smaller
+    if total > limit:
+        return marker(attrs, json_size(attrs), limit)
+    return out

tests/unit/test_classifier_paths.py

@@ -0,0 +1,101 @@
+"""soft_error_classifier: message_keys and dotted paths into nested results."""
+
+import pytest
+
+from fastmcp_feedback.instrumentation import soft_error_classifier
+from fastmcp_feedback.instrumentation.classify import MESSAGE_KEYS, lookup
+
+# mcardupilot: a wait that returned because the SITL died, the cause nested.
+SITL = soft_error_classifier(
+    status_keys=("status", "state", "reason"),
+    extra_statuses=("died",),
+    message_keys=("snapshot.error", "error", "message", "detail", "reason"),
+)
+
+
+class TestMcardupilot:
+    def test_died_reads_the_nested_cause(self):
+        result = {"reason": "died", "snapshot": {"error": "SITL exited: segfault"}}
+        assert SITL("wait_until", result) == "reason=died: SITL exited: segfault"
+
+    def test_timeout_stays_ok(self):
+        result = {"reason": "timeout", "snapshot": {"error": None, "alt": 12.0}}
+        assert SITL("wait_until", result) is None
+
+    def test_reached_stays_ok(self):
+        assert SITL("wait_until", {"reason": "reached", "snapshot": {}}) is None
+
+    def test_died_without_a_cause_falls_back(self):
+        assert SITL("wait_until", {"reason": "died"}) == "reason=died: died"
+
+    def test_dead_session_state(self):
+        result = {"state": "died", "error": "lost heartbeat"}
+        assert SITL("snapshot", result) == "state=died: lost heartbeat"
+
+    def test_default_message_keys_cannot_reach_it(self):
+        plain = soft_error_classifier(
+            status_keys=("status", "state", "reason"), extra_statuses=("died",)
+        )
+        result = {"reason": "died", "snapshot": {"error": "SITL exited: segfault"}}
+        assert plain("wait_until", result) == "reason=died: died"
+
+
+class TestLookup:
+    @pytest.mark.parametrize(
+        "result",
+        [
+            {},
+            {"snapshot": None},
+            {"snapshot": "a string"},
+            {"snapshot": ["error"]},
+            {"snapshot": 3},
+            {"snapshot": {}},
+            {"snapshot": {"other": 1}},
+        ],
+    )
+    def test_missing_or_non_dict_step_is_absent(self, result):
+        assert lookup(result, "snapshot.error") is None
+
+    def test_deep_path(self):
+        assert lookup({"a": {"b": {"c": "x"}}}, "a.b.c") == "x"
+
+    def test_literal_dotted_key_wins(self):
+        result = {"snapshot.error": "literal", "snapshot": {"error": "nested"}}
+        assert lookup(result, "snapshot.error") == "literal"
+
+    def test_plain_key(self):
+        assert lookup({"error": "x"}, "error") == "x"
+        assert lookup({"error": None}, "error") is None
+
+    def test_odd_paths_never_raise(self):
+        for key in ("", ".", "a.", ".a", "a..b"):
+            lookup({"a": {"": {"b": 1}}}, key)
+
+
+class TestOptions:
+    def test_default_message_keys(self):
+        assert MESSAGE_KEYS == ("error", "message", "detail", "reason")
+
+    def test_message_keys_order(self):
+        classify = soft_error_classifier(message_keys=("detail", "error"))
+        result = {"status": "failed", "error": "e", "detail": "d"}
+        assert classify("t", result) == "status=failed: d"
+
+    def test_dotted_error_keys(self):
+        classify = soft_error_classifier(error_keys=("result.error",))
+        assert classify("t", {"result": {"error": " broke "}}) == "result.error: broke"
+        assert classify("t", {"result": {"error": ""}}) is None
+        assert classify("t", {"result": "error"}) is None
+
+    def test_dotted_error_keys_respect_explicit_success(self):
+        classify = soft_error_classifier(error_keys=("result.error",))
+        assert classify("t", {"ok": True, "result": {"error": "warn"}}) is None
+
+    def test_nested_dict_message_searched_with_same_keys(self):
+        classify = soft_error_classifier(message_keys=("problem",))
+        result = {"status": "failed", "problem": {"problem": "inner"}}
+        assert classify("t", result) == "status=failed: inner"
+
+    def test_is_error_uses_message_keys(self):
+        classify = soft_error_classifier(message_keys=("why",))
+        assert classify("t", {"isError": True, "why": "nope"}) == "isError: nope"

tests/unit/test_closed_dispatcher.py

@@ -0,0 +1,159 @@
+"""After aclose() the dispatcher is closed: late records are dropped, never written."""
+
+import asyncio
+import logging
+import threading
+
+from fastmcp import Client, FastMCP
+
+from fastmcp_feedback.instrumentation import MemorySink, instrument
+from fastmcp_feedback.instrumentation.dispatch import CLOSED
+
+
+class ClosingSink(MemorySink):
+    def __init__(self):
+        super().__init__()
+        self.closes = 0
+
+    async def aclose(self):
+        self.closes += 1
+
+
+def make_app():
+    sink = ClosingSink()
+    app = FastMCP("closed")
+
+    @app.tool
+    def echo(text: str) -> dict:
+        return {"echo": text}
+
+    mw = instrument(app, [sink])
+    return app, sink, mw
+
+
+async def started(mw):
+    """A middleware whose dispatcher has a loop and a running task."""
+    assert mw.record_event("boot", key="b")
+    await mw.flush()
+
+
+def kinds(sink):
+    return [e.kind for e in sink.events]
+
+
+class TestAfterClose:
+    async def test_record_event_dropped_and_counted(self):
+        app, sink, mw = make_app()
+        await started(mw)
+        await mw.aclose()
+        before = mw.dispatcher.dropped
+        assert mw.record_event("late", key="k") is False
+        await asyncio.sleep(0.01)
+        assert kinds(sink) == ["boot"]
+        assert mw.dispatcher.dropped == before + 1
+        assert mw.dispatcher.drop_reasons == {CLOSED: 1}
+        assert [e.kind for e in mw.recent_events] == ["boot"]
+        assert mw.dispatcher.closed
+
+    async def test_thread_after_close_dropped(self):
+        app, sink, mw = make_app()
+        await started(mw)
+        await mw.aclose()
+        out = []
+        thread = threading.Thread(target=lambda: out.append(mw.record_event("late")))
+        thread.start()
+        await asyncio.to_thread(thread.join)
+        await asyncio.sleep(0.01)
+        assert out == [False] and kinds(sink) == ["boot"]
+        assert mw.dispatcher.drop_reasons[CLOSED] == 1
+
+    async def test_tool_call_after_close_runs_but_is_not_written(self):
+        app, sink, mw = make_app()
+        async with Client(app) as client:
+            await client.call_tool("echo", {"text": "before"})
+            await mw.aclose()
+            result = await client.call_tool("echo", {"text": "after"})
+        assert result.data == {"echo": "after"}
+        await asyncio.sleep(0.01)
+        assert [r.args_size for r in sink.calls] == [len('{"text": "before"}')]
+        assert len(mw.recent) == 1
+        assert mw.dispatcher.drop_reasons[CLOSED] == 1
+
+    async def test_close_twice_is_a_no_op(self):
+        app, sink, mw = make_app()
+        await started(mw)
+        await mw.aclose()
+        await mw.aclose()
+        assert sink.closes == 1 and mw.dispatcher.drop_reasons == {}
+
+    async def test_close_before_any_record(self):
+        app, sink, mw = make_app()
+        await mw.aclose()
+        assert mw.record_event("late") is False
+        assert mw.dispatcher._task is None and mw.dispatcher._loop is None
+
+    async def test_never_rebinds_or_starts_a_task(self):
+        app, sink, mw = make_app()
+        await started(mw)
+        home = mw.dispatcher._loop
+        await mw.aclose()
+        mw.dispatcher.bind_loop()
+
+        def other_loop():
+            async def submit():
+                return mw.record_event("elsewhere")
+
+            return asyncio.run(submit())
+
+        assert await asyncio.to_thread(other_loop) is False
+        assert mw.dispatcher._task is None and mw.dispatcher._queue is None
+        assert mw.dispatcher._loop is home
+        assert mw.dispatcher.drop_reasons == {CLOSED: 1}
+
+    async def test_warns_once(self, caplog):
+        app, sink, mw = make_app()
+        await started(mw)
+        await mw.aclose()
+        with caplog.at_level(logging.WARNING, "fastmcp_feedback"):
+            for _ in range(5):
+                mw.record_event("late")
+        warnings = [r for r in caplog.records if "after aclose" in r.getMessage()]
+        assert len(warnings) == 1
+        assert mw.dispatcher.drop_reasons == {CLOSED: 5}
+
+
+class TestHandoffAtClose:
+    """A record a thread handed to the loop just before aclose(), made
+    deterministic by blocking the loop while the thread hands it over."""
+
+    def hand_over(self, mw, name):
+        out = []
+        thread = threading.Thread(target=lambda: out.append(mw.record_event(name)))
+        thread.start()
+        thread.join()  # blocks the loop: the callback is scheduled, not run
+        assert out == [True]
+
+    async def test_landing_after_the_flush_is_dropped(self):
+        # Nothing queued: the flush returns without yielding, the dispatcher
+        # closes, and the handed-over record lands on a closed dispatcher.
+        app, sink, mw = make_app()
+        await started(mw)
+        self.hand_over(mw, "late")
+        await mw.aclose()
+        await asyncio.sleep(0.01)
+        assert kinds(sink) == ["boot"]
+        assert mw.dispatcher.drop_reasons == {CLOSED: 1}
+        assert "late" not in [e.kind for e in mw.recent_events]
+
+    async def test_landing_during_the_flush_is_delivered(self):
+        # Something still queued: the flush yields, the record lands while it
+        # runs, and the flush delivers it before the dispatcher closes.
+        app, sink, mw = make_app()
+        await started(mw)
+        assert mw.record_event("queued")  # on the loop, not yet delivered
+        self.hand_over(mw, "handed")
+        await mw.aclose()
+        assert kinds(sink) == ["boot", "queued", "handed"]
+        assert mw.dispatcher.drop_reasons == {}
+        await asyncio.sleep(0.01)
+        assert kinds(sink) == ["boot", "queued", "handed"]

tests/unit/test_consistency.py

@@ -364,7 +364,8 @@ class TestEventIdentity:
         [ev] = sink.events
         assert rec.ok and ev.call_id == rec.id
         assert ev.user_sub is None and rec.user_sub is None
-        assert calls == [1]
+        # An empty first answer gets a second run after the tool; it fails too.
+        assert calls == [1, 1]
 
     async def test_slow_resolver_cut_off_before_tool(self):
         async def slow(context):

tests/unit/test_late_identity.py

@@ -0,0 +1,320 @@
+"""The identity resolver's second run, after the tool, for identity the tool creates.
+
+Modelled on blender-mcp: an addon calls register_client, and only after that
+call can the resolver find the bus client (bus_client_uuid, bus_role) for the
+caller's session. The first run, before the tool, already knows the user.
+"""
+
+import asyncio
+import contextlib
+import time
+
+from fastmcp import Client, FastMCP
+from fastmcp.exceptions import ToolError
+
+from fastmcp_feedback.instrumentation import REDACTED, MemorySink, instrument
+
+
+class Bus:
+    """Who is registered, as the tool leaves it; read by the resolvers."""
+
+    def __init__(self):
+        self.client = None
+
+
+class BlenderResolver:
+    """blender-mcp's identify(): user and role always, bus keys once registered."""
+
+    def __init__(self, bus):
+        self.bus = bus
+        self.calls = 0
+
+    def __call__(self, context):
+        self.calls += 1
+        out = {"user_sub": "user-1", "caller_kind": "addon", "client_id": "dcr-9"}
+        if self.bus.client is not None:
+            out["bus_client_uuid"] = self.bus.client
+            out["bus_role"] = "interactive"
+        return out
+
+
+class EmptyUntilRegistered:
+    """Nothing before registration, every field after."""
+
+    def __init__(self, bus):
+        self.bus = bus
+        self.calls = 0
+
+    def __call__(self, context):
+        self.calls += 1
+        if self.bus.client is None:
+            return {}
+        return {
+            "user_sub": "user-1",
+            "caller_kind": "addon",
+            "client_id": "dcr-9",
+            "bus_client_uuid": self.bus.client,
+        }
+
+
+def make_app(resolver_cls, *, late=(), sample=None, **kwargs):
+    bus = Bus()
+    resolver = resolver_cls(bus)
+    sink = MemorySink()
+    app = FastMCP("late-identity")
+    holder = {}
+
+    @app.tool
+    async def register_client(client_uuid: str) -> dict:
+        holder["mw"].record_event("bus.registering", key=client_uuid)
+        bus.client = client_uuid
+        return {"status": "ok"}
+
+    @app.tool
+    async def register_then_fail(client_uuid: str) -> dict:
+        bus.client = client_uuid
+        raise RuntimeError("registered, then broke")
+
+    @app.tool
+    async def quiet() -> str:
+        return "ok"
+
+    @app.tool
+    async def ping() -> str:
+        holder["mw"].record_event("bus.ping")
+        return "pong"
+
+    kwargs.setdefault("identity_resolver", resolver)
+    if sample is not None:
+        kwargs.update(
+            sample_rates={"ping": sample, "quiet": sample}, random=lambda: 0.99
+        )
+    mw = instrument(app, [sink], late_identity_keys=late, **kwargs)
+    holder["mw"] = mw
+    return app, sink, mw, resolver
+
+
+async def call(app, mw, *calls):
+    async with Client(app) as client:
+        for tool, args in calls:
+            with contextlib.suppress(ToolError):
+                await client.call_tool(tool, args)
+    await mw.flush()
+
+
+def columns(record):
+    return (record.user_sub, record.caller_kind, record.client_id)
+
+
+class TestRegistration:
+    async def test_blender_shaped_registration_row_gets_bus_identity(self):
+        app, sink, mw, who = make_app(BlenderResolver, late=("bus_client_uuid",))
+        await call(app, mw, ("register_client", {"client_uuid": "c-1"}))
+        [rec] = sink.calls
+        assert columns(rec) == ("user-1", "addon", "dcr-9")
+        assert rec.identity == {"bus_client_uuid": "c-1", "bus_role": "interactive"}
+        assert who.calls == 2
+        # The event recorded during the call saw the first answer only.
+        [ev] = sink.events
+        assert ev.call_id == rec.id and columns(ev) == ("user-1", "addon", "dcr-9")
+
+    async def test_blender_shaped_without_late_keys_runs_once(self):
+        # The first answer is not empty, so without late_identity_keys there
+        # is no second run and the row lacks the bus keys. Opt-in by design.
+        app, sink, mw, who = make_app(BlenderResolver)
+        await call(app, mw, ("register_client", {"client_uuid": "c-1"}))
+        [rec] = sink.calls
+        assert rec.identity == {} and "bus_client_uuid" not in rec.identity
+        assert columns(rec) == ("user-1", "addon", "dcr-9")
+        assert who.calls == 1
+
+    async def test_registered_caller_runs_once(self):
+        app, sink, mw, who = make_app(BlenderResolver, late=("bus_client_uuid",))
+        await call(
+            app,
+            mw,
+            ("register_client", {"client_uuid": "c-1"}),
+            ("ping", {}),
+            ("ping", {}),
+        )
+        assert who.calls == 2 + 1 + 1
+        assert [r.identity.get("bus_client_uuid") for r in sink.calls] == [
+            "c-1",
+            "c-1",
+            "c-1",
+        ]
+
+    async def test_empty_first_answer_filled_after(self):
+        app, sink, mw, who = make_app(EmptyUntilRegistered)
+        await call(app, mw, ("register_client", {"client_uuid": "c-2"}))
+        [rec] = sink.calls
+        [ev] = sink.events
+        assert columns(rec) == ("user-1", "addon", "dcr-9")
+        assert rec.identity == {"bus_client_uuid": "c-2"}
+        assert columns(ev) == (None, None, None)  # not rewritten
+        assert who.calls == 2
+
+    async def test_tool_that_raises_gets_second_run_too(self):
+        app, sink, mw, who = make_app(EmptyUntilRegistered)
+        await call(app, mw, ("register_then_fail", {"client_uuid": "c-3"}))
+        [rec] = sink.calls
+        assert rec.outcome == "error"
+        assert columns(rec) == ("user-1", "addon", "dcr-9")
+        assert who.calls == 2
+
+
+class TestMerge:
+    async def test_first_answer_wins_on_conflict(self):
+        answers = iter(
+            [
+                {"user_sub": "early", "caller_kind": None, "org": None},
+                {
+                    "user_sub": "late",
+                    "caller_kind": "k",
+                    "client_id": "c",
+                    "org": "acme",
+                    "team": "qa",
+                },
+            ]
+        )
+        app, sink, mw, _ = make_app(
+            EmptyUntilRegistered,
+            identity_resolver=lambda c: next(answers),
+            late=("team",),
+        )
+        await call(app, mw, ("ping", {}))
+        [rec] = sink.calls
+        [ev] = sink.events
+        assert columns(rec) == ("early", "k", "c")
+        assert rec.identity == {"org": "acme", "team": "qa"}
+        assert columns(ev) == ("early", None, None)
+
+    async def test_second_answer_redacted(self):
+        token = "bmcp_abcdefgh12345678"
+        answers = iter([{}, {"user_sub": "u", "caller_kind": f"pat:{token}"}])
+        app, sink, mw, _ = make_app(
+            EmptyUntilRegistered, identity_resolver=lambda c: next(answers)
+        )
+        await call(app, mw, ("ping", {}))
+        assert sink.calls[0].caller_kind == f"pat:{REDACTED}"
+
+
+class TestFailures:
+    async def test_second_run_raising_keeps_first_answer(self):
+        runs = []
+
+        def who(context):
+            runs.append(1)
+            if len(runs) == 1:
+                return {"user_sub": "u1"}
+            raise RuntimeError("resolver bug")
+
+        app, sink, mw, _ = make_app(
+            EmptyUntilRegistered, identity_resolver=who, late=("bus_client_uuid",)
+        )
+        await call(app, mw, ("ping", {}))
+        [rec] = sink.calls
+        assert rec.ok and rec.user_sub == "u1" and len(runs) == 2
+
+    async def test_second_run_timing_out_keeps_first_answer(self):
+        runs = []
+
+        async def who(context):
+            runs.append(1)
+            if len(runs) == 1:
+                return {"user_sub": "u1"}
+            await asyncio.sleep(5)
+            return {"user_sub": "late", "caller_kind": "k"}
+
+        app, sink, mw, _ = make_app(
+            EmptyUntilRegistered,
+            identity_resolver=who,
+            hook_timeout=0.05,
+            late=("bus_client_uuid",),
+        )
+        await call(app, mw, ("ping", {}))
+        [rec] = sink.calls
+        assert rec.ok and rec.user_sub == "u1" and rec.caller_kind is None
+        assert len(runs) == 2
+
+
+class McspeakResolver:
+    """mcspeak's _who: user_sub and caller_kind, never client_id."""
+
+    def __init__(self, bus):
+        self.calls = 0
+
+    def __call__(self, context):
+        self.calls += 1
+        return {"caller_kind": "claude-code", "user_sub": "mctalkbox"}
+
+
+class TestWhenItRuns:
+    async def test_missing_column_alone_does_not_rerun(self):
+        app, sink, mw, who = make_app(McspeakResolver)
+        await call(app, mw, ("ping", {}), ("quiet", {}), ("ping", {}))
+        assert who.calls == 3 and len(sink.calls) == 3
+        assert {columns(r) for r in sink.calls} == {("mctalkbox", "claude-code", None)}
+
+    async def test_only_none_values_count_as_empty(self):
+        answers = iter([{"user_sub": None}, {"user_sub": "u2", "caller_kind": "k"}])
+        app, sink, mw, _ = make_app(
+            EmptyUntilRegistered, identity_resolver=lambda c: next(answers)
+        )
+        await call(app, mw, ("ping", {}))
+        assert columns(sink.calls[0])[:2] == ("u2", "k")
+
+    async def test_complete_answer_runs_once(self):
+        app, sink, mw, who = make_app(BlenderResolver)
+        await call(app, mw, ("ping", {}), ("ping", {}))
+        assert who.calls == 2 and len(sink.calls) == 2
+
+    async def test_sampled_out_call_makes_no_second_run(self):
+        app, sink, mw, who = make_app(EmptyUntilRegistered, sample=0.1)
+        await call(app, mw, ("quiet", {}))
+        assert sink.calls == [] and who.calls == 1
+
+    async def test_sampled_call_kept_for_its_event_gets_second_run(self):
+        app, sink, mw, who = make_app(EmptyUntilRegistered, sample=0.1)
+        await call(app, mw, ("ping", {}))
+        assert len(sink.calls) == 1 and who.calls == 2
+
+    async def test_no_resolver_no_runs(self):
+        app, sink, mw, _ = make_app(EmptyUntilRegistered, identity_resolver=None)
+        await call(app, mw, ("ping", {}))
+        [rec] = sink.calls
+        assert columns(rec) == (None, None, None)
+
+    async def test_second_run_overlaps_enricher(self):
+        # The record is built before the result reaches the client, so the
+        # second run must not add its time to the enricher's.
+        runs = []
+
+        async def who(context):
+            runs.append(1)
+            if len(runs) > 1:
+                await asyncio.sleep(0.4)
+            return {}
+
+        async def enrich(tool, args, result, context):
+            await asyncio.sleep(0.4)
+            return {}
+
+        app, sink, mw, _ = make_app(
+            EmptyUntilRegistered,
+            identity_resolver=who,
+            enricher=enrich,
+            hook_timeout=2,
+        )
+        async with Client(app) as client:
+            start = time.perf_counter()
+            await client.call_tool("ping", {})
+            elapsed = time.perf_counter() - start
+        assert len(runs) == 2
+        assert elapsed < 0.75, elapsed
+
+
+async def test_mode_off_runs_nothing():
+    app, sink, mw, who = make_app(EmptyUntilRegistered, mode="off")
+    await call(app, mw, ("ping", {}))
+    assert who.calls == 0

tests/unit/test_payload_cap.py

@@ -0,0 +1,250 @@
+"""max_payload_bytes: stored arguments, results and event attrs are bounded."""
+
+import json
+
+import pytest
+from fastmcp import Client, FastMCP
+
+from fastmcp_feedback.instrumentation import (
+    REDACTED,
+    DatabaseSink,
+    EventRecord,
+    MemorySink,
+    instrument,
+)
+from fastmcp_feedback.instrumentation import payload as payload_mod
+from fastmcp_feedback.instrumentation.embeddings import extract_texts
+from fastmcp_feedback.instrumentation.payload import (
+    DEFAULT_MAX_PAYLOAD_BYTES,
+    PREVIEW_CHARS,
+    TRUNCATED,
+    cap_attrs,
+    is_truncated,
+    json_size,
+    preview_chars,
+)
+
+SECRET = "sk-" + "a1b2c3d4e5" * 4  # matches the default sk- pattern
+
+
+def make_app(**kwargs):
+    sink = MemorySink()
+    app = FastMCP("cap")
+    holder = {}
+
+    @app.tool
+    def echo(blob: str) -> dict:
+        return {"blob": blob}
+
+    @app.tool
+    def big(n: int) -> dict:
+        return {"rows": ["x" * 100] * n}
+
+    @app.tool
+    def report(error: str) -> str:
+        holder["mw"].record_event(
+            "job.finished", key="j1", attrs={"ok": False, "error": error}
+        )
+        return "ok"
+
+    kwargs.setdefault("mode", "full")
+    mw = instrument(app, [sink], **kwargs)
+    holder["mw"] = mw
+    return app, sink, mw
+
+
+async def call(app, mw, tool, args):
+    async with Client(app) as client:
+        await client.call_tool(tool, args)
+    await mw.flush()
+
+
+class TestCallPayloads:
+    async def test_default_is_64_kib(self):
+        app, sink, mw = make_app()
+        assert mw.max_payload_bytes == DEFAULT_MAX_PAYLOAD_BYTES == 65536
+
+    async def test_under_the_cap_stored_whole(self):
+        app, sink, mw = make_app()
+        await call(app, mw, "echo", {"blob": "y" * 1000})
+        [rec] = sink.calls
+        assert rec.args == {"blob": "y" * 1000}
+        assert rec.result == {"blob": "y" * 1000}
+
+    async def test_over_the_cap_becomes_a_marker(self):
+        app, sink, mw = make_app()
+        await call(app, mw, "big", {"n": 2000})
+        [rec] = sink.calls
+        assert rec.args == {"n": 2000}
+        assert rec.result_size > 65536
+        assert set(rec.result) == {TRUNCATED, "size", "preview"}
+        assert rec.result[TRUNCATED] is True
+        assert rec.result["size"] == rec.result_size
+        assert len(rec.result["preview"]) == PREVIEW_CHARS
+        assert rec.result["preview"].startswith('{"rows": ["xxxx')
+        json.dumps(rec.result)  # JSON-safe
+
+    async def test_args_over_the_cap(self):
+        app, sink, mw = make_app(max_payload_bytes=500)
+        await call(app, mw, "echo", {"blob": "z" * 2000})
+        [rec] = sink.calls
+        assert rec.args["size"] == rec.args_size and is_truncated(rec.args)
+        # Under a small cap the preview shrinks so the marker fits.
+        assert len(rec.args["preview"]) == preview_chars(500) == 125
+        assert json_size(rec.args) <= 500
+
+    async def test_none_means_unlimited(self):
+        app, sink, mw = make_app(max_payload_bytes=None)
+        await call(app, mw, "big", {"n": 20000})
+        [rec] = sink.calls
+        assert len(rec.result["rows"]) == 20000
+
+    async def test_meta_mode_stores_no_payload_but_sizes(self):
+        app, sink, mw = make_app(mode="meta")
+        await call(app, mw, "big", {"n": 2000})
+        [rec] = sink.calls
+        assert rec.args is None and rec.result is None
+        assert rec.result_size > 65536
+
+    @pytest.mark.parametrize("before_end", [40, 15])
+    async def test_redaction_happens_before_the_cap(self, before_end):
+        # At 15 the preview ends inside the secret: cutting the raw JSON there
+        # would leave a fragment too short for the pattern to match.
+        app, sink, mw = make_app(max_payload_bytes=4000)
+        pad = preview_chars(4000) - len('{"blob": "') - before_end
+        blob = "p" * pad + " " + SECRET + " " + "q" * 5000
+        await call(app, mw, "echo", {"blob": blob})
+        [rec] = sink.calls
+        for stored in (rec.args, rec.result):
+            assert is_truncated(stored)
+            assert SECRET[:8] not in stored["preview"]
+            assert "[REDACT" in stored["preview"]
+
+    async def test_secret_key_redacted_in_preview(self):
+        app, sink, mw = make_app(max_payload_bytes=200)
+        app.tool(lambda api_key, pad: "ok", name="login")
+        await call(app, mw, "login", {"api_key": "hunter2-secret", "pad": "x" * 500})
+        [rec] = sink.calls
+        assert "hunter2" not in rec.args["preview"]
+        assert REDACTED in rec.args["preview"]
+
+    @pytest.mark.parametrize("bad", [0, -1, True, "64k", 1.5])
+    def test_invalid_cap_rejected(self, bad):
+        with pytest.raises(ValueError, match="max_payload_bytes"):
+            make_app(max_payload_bytes=bad)
+
+
+class TestNoSecondEncoding:
+    async def test_payloads_encoded_once(self, monkeypatch):
+        # The size columns and the cap share one measurement per payload,
+        # whether or not the payload is over the cap.
+        app, sink, mw = make_app(max_payload_bytes=1000)
+        measured = []
+        original = payload_mod.json_size
+
+        def counting(value):
+            measured.append(1)
+            return original(value)
+
+        monkeypatch.setattr(
+            "fastmcp_feedback.instrumentation.middleware.json_size", counting
+        )
+        await call(app, mw, "echo", {"blob": "small"})
+        await call(app, mw, "big", {"n": 5000})
+        assert len(measured) == 4  # args and result, two calls
+        assert is_truncated(sink.calls[1].result)
+
+    def test_preview_stops_early(self, monkeypatch):
+        chunks = []
+        encoder = payload_mod._ENCODER
+
+        class Counting:
+            def iterencode(self, value):
+                for chunk in encoder.iterencode(value):
+                    chunks.append(chunk)
+                    yield chunk
+
+        monkeypatch.setattr(payload_mod, "_ENCODER", Counting())
+        huge = {"rows": ["x" * 100] * 100_000}  # about 10 MB
+        text = payload_mod.preview(huge)
+        assert len(text) == PREVIEW_CHARS
+        assert len(chunks) < 50
+
+
+class TestEventAttrs:
+    async def test_large_attr_replaced_small_ones_kept(self):
+        app, sink, mw = make_app(mode="meta")
+        await call(app, mw, "report", {"error": "E" * 100_000})
+        [ev] = sink.events
+        assert ev.attrs["ok"] is False
+        assert is_truncated(ev.attrs["error"])
+        assert ev.attrs["error"]["size"] == json_size("E" * 100_000)
+        assert json_size(ev.attrs) <= 65536
+
+    async def test_small_attrs_untouched(self):
+        app, sink, mw = make_app(mode="meta")
+        await call(app, mw, "report", {"error": "short"})
+        assert sink.events[0].attrs == {"ok": False, "error": "short"}
+
+    async def test_attrs_redacted_before_the_cap(self):
+        app, sink, mw = make_app(mode="meta", max_payload_bytes=4000)
+        error = "x" * (preview_chars(4000) - 30) + f" token {SECRET} " + "y" * 9000
+        await call(app, mw, "report", {"error": error})
+        stored = sink.events[0].attrs["error"]
+        assert SECRET[:8] not in stored["preview"] and REDACTED in stored["preview"]
+
+    async def test_outside_a_call_and_unlimited(self):
+        app, sink, mw = make_app(mode="meta", max_payload_bytes=None)
+        assert mw.record_event("job.log", key="j", attrs={"log": "L" * 200_000})
+        await mw.flush()
+        assert sink.events[0].attrs["log"] == "L" * 200_000
+
+    async def test_llm_call_attrs_capped_text_bounded(self):
+        app, sink, mw = make_app(mode="meta", capture_llm_text=True)
+        assert mw.record_llm_call(
+            model="m",
+            prompt="P" * 100_000,
+            attrs={"messages": ["M" * 1000] * 100},
+        )
+        await mw.flush()
+        [ev] = sink.events
+        assert len(ev.attrs["prompt"]) == mw.max_llm_text_chars
+        assert is_truncated(ev.attrs["messages"])
+        assert ev.attrs["model"] == "m" and ev.attrs["ok"] is True
+
+    def test_many_small_values_become_one_marker(self):
+        attrs = {f"k{i}": "v" * 50 for i in range(500)}
+        capped = cap_attrs(attrs, 1000)
+        assert is_truncated(capped) and capped["size"] == json_size(attrs)
+
+    def test_tiny_cap(self):
+        capped = cap_attrs({"a": "x" * 100}, 10)
+        assert is_truncated(capped) and len(capped["preview"]) == 2
+
+
+class TestDownstream:
+    async def test_marker_round_trips_through_database(self, db_url):
+        sink = DatabaseSink(db_url("cap"), create_tables=True)
+        try:
+            app, _, mw = make_app(mode="meta")
+            mw.dispatcher.sinks.append(sink)
+            await call(app, mw, "report", {"error": "E" * 100_000})
+            rows = await mw.events_for("j1")
+            [row] = [r for r in rows if r["kind"] == "job.finished"]
+            assert row["attrs"]["error"][TRUNCATED] is True
+            assert row["attrs"]["ok"] is False
+        finally:
+            await sink.engine.dispose()
+
+    def test_embedding_text_uses_the_preview(self):
+        from datetime import UTC, datetime
+
+        marker = payload_mod.marker("disk full: " + "z" * 5000, 5013, 100)
+        record = EventRecord(
+            kind="job.failed",
+            occurred_at=datetime.now(UTC),
+            attrs={"error": marker},
+        )
+        [text] = extract_texts([record])
+        assert text.text.startswith('"disk full: zzz')
+        assert "_truncated" not in text.text

uv.lock

@@ -601,7 +601,7 @@ wheels = [
 
 [[package]]
 name = "fastmcp-feedback"
-version = "2026.10.1.4"
+version = "2026.10.5"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },