Compare commits

..
Author SHA1 Message Date
rsp2k 84b7591858 Merge fix/closed-dispatcher: the dispatcher stays closed after aclose()
Tests / test (==2.12.2) (push) Canceled after 0s
Tests / test (==2.14.3) (push) Canceled after 0s
Tests / test (>=3.3,<4) (push) Canceled after 0s
Tests / test (>=4,<5) (push) Canceled after 0s
Tests / test-postgres (==2.12.2) (push) Canceled after 0s
Tests / test-postgres (>=4,<5) (push) Canceled after 0s
Tests / canary (push) Canceled after 0s
2026-10-05 01:19:37 -06:00
rsp2k 8fa36d6838 Release 2026.10.05: the dispatcher stays closed after aclose()
Tests / test (==2.12.2) (push) Canceled after 0s
Tests / test (==2.14.3) (push) Canceled after 0s
Tests / test (>=3.3,<4) (push) Canceled after 0s
Tests / test (>=4,<5) (push) Canceled after 0s
Tests / test-postgres (==2.12.2) (push) Canceled after 0s
Tests / test-postgres (>=4,<5) (push) Canceled after 0s
Tests / canary (push) Canceled after 0s
2026-10-05 01:17:06 -06:00
rsp2k b28482fffc Keep the dispatcher 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.
2026-10-05 01:16:32 -06:00
rsp2k 15dfe17095 Merge feature/payload-cap-dotted-keys: payload size limit, dotted classifier keys
Tests / test (==2.12.2) (push) Canceled after 0s
Tests / test (==2.14.3) (push) Canceled after 0s
Tests / test (>=3.3,<4) (push) Canceled after 0s
Tests / test (>=4,<5) (push) Canceled after 0s
Tests / test-postgres (==2.12.2) (push) Canceled after 0s
Tests / test-postgres (>=4,<5) (push) Canceled after 0s
Tests / canary (push) Canceled after 0s
2026-10-04 14:45:18 -06:00
rsp2k 4206e15293 Release 2026.10.04: payload size limit, dotted classifier keys
Tests / test (>=4,<5) (push) Canceled after 0s
Tests / test (==2.12.2) (push) Canceled after 0s
Tests / test (==2.14.3) (push) Canceled after 0s
Tests / test (>=3.3,<4) (push) Canceled after 0s
Tests / test-postgres (==2.12.2) (push) Canceled after 0s
Tests / test-postgres (>=4,<5) (push) Canceled after 0s
Tests / canary (push) Canceled after 0s
2026-10-04 14:43:32 -06:00
rsp2k 8bf0801cf6 Test a preview that ends inside a secret 2026-10-04 14:42:16 -06:00
rsp2k 72d2845502 Cap stored payloads and event attrs; dotted paths in the classifier
max_payload_bytes (default 64 KiB, None for no limit) bounds the arguments
and results stored in mode full and every event's attrs. A larger value is
stored as {"_truncated": true, "size": ..., "preview": ...}, the preview
taken from the redacted value. Call payloads reuse the size already
measured for args_size and result_size; attrs keep their keys, with only
the largest values replaced.

soft_error_classifier takes message_keys, and message_keys and error_keys
accept dotted paths into nested dicts ("snapshot.error"). An exact key is
matched before splitting; a missing or non-dict step is absent.
2026-10-04 14:41:13 -06:00
rsp2k 69a09f4400 Merge fix/narrow-late-identity: second identity run only for an empty answer or a missing late key
Tests / test (==2.12.2) (push) Canceled after 0s
Tests / test (==2.14.3) (push) Canceled after 0s
Tests / test (>=3.3,<4) (push) Canceled after 0s
Tests / test (>=4,<5) (push) Canceled after 0s
Tests / test-postgres (==2.12.2) (push) Canceled after 0s
Tests / test-postgres (>=4,<5) (push) Canceled after 0s
Tests / canary (push) Canceled after 0s
2026-10-02 00:11:38 -06:00
14 changed files with 906 additions and 35 deletions
+50
View File
@@ -5,6 +5,56 @@ 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
+3 -1
View File
@@ -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_late_identity.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
+61 -1
View File
@@ -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
@@ -945,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
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "fastmcp-feedback"
version = "2026.10.02.1" # 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"
+1 -1
View File
@@ -41,7 +41,7 @@ Quick Start:
import logging
# Version information - CalVer YYYY.MM.DD
__version__ = "2026.10.02.1"
__version__ = "2026.10.05"
__author__ = "Ryan Malloy"
__email__ = "ryan@supported.systems"
@@ -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
@@ -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
@@ -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):
@@ -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.
@@ -184,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
@@ -215,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:
@@ -247,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
@@ -343,7 +352,9 @@ 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":
@@ -453,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,
@@ -469,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,
@@ -506,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":
@@ -550,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:
@@ -855,7 +878,13 @@ 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()
@@ -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
+101
View File
@@ -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"
+159
View File
@@ -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"]
+250
View File
@@ -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
Generated
+1 -1
View File
@@ -601,7 +601,7 @@ wheels = [
[[package]]
name = "fastmcp-feedback"
version = "2026.10.2.1"
version = "2026.10.5"
source = { editable = "." }
dependencies = [
{ name = "fastmcp" },