Release 2026.10.05: 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,28 @@ 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.

Makefile

@@ -62,4 +62,5 @@ lint:
 		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_payload_cap.py tests/unit/test_classifier_paths.py
+		tests/unit/test_payload_cap.py tests/unit/test_classifier_paths.py \
+		tests/unit/test_closed_dispatcher.py

README.md

@@ -1000,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.04"  # 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.04"
+__version__ = "2026.10.05"
 __author__ = "Ryan Malloy"
 __email__ = "ryan@supported.systems"
 

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/middleware.py

@@ -352,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":
@@ -522,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":
@@ -874,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()
 
 

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"]

uv.lock

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