From 3185f5c2b26ad5c89713ee4ff2396d2df3756435 Mon Sep 17 00:00:00 2001
From: genematx <ymatviych@bnl.gov>
Date: Tue, 3 Nov 2026 11:38:43 +1300
Subject: [PATCH 1/6] Require tiled >=0.2.21 and use httpx2 to match the tiled
 client

tiled 0.2.21 migrated its client from httpx to httpx2, changing the
exception family raised through it: ClientError now subclasses
httpx2.HTTPStatusError, so our `except httpx.HTTPStatusError` clauses
silently stopped matching. Concretely, the remote-validation 404 fallbacks
in BlueskyRun.validate and TiledWriter raised instead of falling back to
local validation.

Pin tiled >=0.2.21 (runtime and test group) and switch both sites to
httpx2, declared as a direct dependency since we import it. A small test
guards the contract (ClientError is catchable via httpx2.HTTPStatusError)
so the next migration fails loudly instead of silently.
---
 pyproject.toml                                    |  5 +++--
 src/bluesky_tiled_plugins/clients/bluesky_run.py  |  6 +++---
 src/bluesky_tiled_plugins/writing/tiled_writer.py |  6 +++---
 tests/test_httpx2.py                              | 14 ++++++++++++++
 tests/test_tiled_writer.py                        |  6 +++---
 5 files changed, 26 insertions(+), 11 deletions(-)
 create mode 100644 tests/test_httpx2.py

diff --git a/pyproject.toml b/pyproject.toml
index 30d23f6..8ff15f1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -36,13 +36,14 @@ dynamic = ["version"]
 dependencies = [
   "dask",
   "event-model",
+  "httpx2",  # the generation used by the tiled client since 0.2.21
   "mongoquery",
   "ndindex",  # imported by tiled.ndslice
   "numpy",
   "pandas",
   "pyarrow",
   "pytz",
-  "tiled[minimal-client] >=0.2.19",
+  "tiled[minimal-client] >=0.2.21",
   "tzlocal",
   "xarray",
 ]
@@ -88,7 +89,7 @@ test = [
   "redis-json-dict",
   "suitcase-mongo",
   "tifffile",
-  "tiled[server] >=0.2.19",
+  "tiled[server] >=0.2.21",
 ]
 dev = [{ include-group = "test" }]
 docs = [
diff --git a/src/bluesky_tiled_plugins/clients/bluesky_run.py b/src/bluesky_tiled_plugins/clients/bluesky_run.py
index 7be9ccf..83da862 100644
--- a/src/bluesky_tiled_plugins/clients/bluesky_run.py
+++ b/src/bluesky_tiled_plugins/clients/bluesky_run.py
@@ -1,7 +1,7 @@
 import codecs
 import copy
 import functools
-import httpx
+import httpx2
 import io
 import json
 import keyword
@@ -534,13 +534,13 @@ def validate(
                 msg = "Remote validation failed: " + "; ".join(notes)
                 raise ValidationException(msg, self.item["id"])
 
-        except httpx.HTTPStatusError as e:
+        except httpx2.HTTPStatusError as e:
             # Backcompatibility: if the server does not support validation endpoint,
             # it will return 404 Not Found error; in this case, attempt to validate
             # the data structure locally by the client itself (requires multiple
             # round-trips to the server, but better than nothing).
 
-            if response.status_code == httpx.codes.NOT_FOUND:
+            if response.status_code == httpx2.codes.NOT_FOUND:
                 warnings.warn(
                     "Tiled server does not support remote validation. "
                     "Attempting to validate the data structure by the client.",
diff --git a/src/bluesky_tiled_plugins/writing/tiled_writer.py b/src/bluesky_tiled_plugins/writing/tiled_writer.py
index a95c63e..60f3484 100644
--- a/src/bluesky_tiled_plugins/writing/tiled_writer.py
+++ b/src/bluesky_tiled_plugins/writing/tiled_writer.py
@@ -9,7 +9,7 @@
 from typing import Any, Optional, cast
 import warnings
 
-import httpx
+import httpx2
 import numpy
 import pyarrow
 from event_model import (
@@ -908,9 +908,9 @@ def _append_with_retry(self, df_client: DataFrameClient, table, desc_name: str):
         try:
             df_client.append_partition(0, table)
             return
-        except (httpx.TransportError, httpx.HTTPStatusError) as err:
+        except (httpx2.TransportError, httpx2.HTTPStatusError) as err:
             if (
-                isinstance(err, httpx.HTTPStatusError)
+                isinstance(err, httpx2.HTTPStatusError)
                 and err.response.status_code < 500
                 and err.response.status_code != 429
             ):
diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py
new file mode 100644
index 0000000..7b06a6f
--- /dev/null
+++ b/tests/test_httpx2.py
@@ -0,0 +1,14 @@
+"""Guard the httpx generation contract with the tiled client.
+
+bluesky-tiled-plugins catches exceptions raised through the tiled client
+(e.g. in the remote-validation 404 fallbacks), so the exception family it
+catches must match the one tiled raises. Since tiled 0.2.21 that is httpx2;
+this test fails loudly if tiled migrates again.
+"""
+
+import httpx2
+from tiled.client.utils import ClientError
+
+
+def test_tiled_client_error_is_catchable_via_httpx2():
+    assert issubclass(ClientError, httpx2.HTTPStatusError)
diff --git a/tests/test_tiled_writer.py b/tests/test_tiled_writer.py
index c2def29..3b53c24 100644
--- a/tests/test_tiled_writer.py
+++ b/tests/test_tiled_writer.py
@@ -6,7 +6,7 @@
 from typing import cast
 from urllib.parse import parse_qs, urlparse
 
-import httpx
+import httpx2
 
 from bluesky import RunEngine
 import bluesky.plan_stubs as bps
@@ -1013,7 +1013,7 @@ def patched_event(name, doc):
     assert lines[6]["name"] == "stop"
 
 
-class _LoseAppendResponseOnce(httpx.BaseTransport):
+class _LoseAppendResponseOnce(httpx2.BaseTransport):
     """Forward the first PATCH /table/partition request to the wrapped transport
     -- so the server fully processes the append -- but raise ReadTimeout to the
     caller, simulating a lost response for an append that actually succeeded."""
@@ -1030,7 +1030,7 @@ def handle_request(self, request):
             and "/table/partition/" in str(request.url)
         ):
             self.tripped = True
-            raise httpx.ReadTimeout("Simulated lost response", request=request)
+            raise httpx2.ReadTimeout("Simulated lost response", request=request)
         return response
 
 

From 1b57e44c233dbd0ac017d812269dd3592dc6b421 Mon Sep 17 00:00:00 2001
From: genematx <ymatviych@bnl.gov>
Date: Tue, 3 Nov 2026 12:00:30 +1300
Subject: [PATCH 2/6] Require Python >=3.11 to match tiled >=0.2.21

tiled 0.2.21 (the httpx2 generation) dropped Python 3.10, which made the
dependency resolution unsatisfiable for the 3.10 split of the supported
range and failed CI on every job, including 3.13 and the docs build.

- Bump `requires-python` to >=3.11 and drop the 3.10 classifier
- Update mypy `python_version` and pylint `py-version` to 3.11
- Test on 3.11 (new floor) and 3.13 in CI
- Update the Python version requirement in the installation tutorials
- Use `itertools.pairwise` in StreamDatum batching, resolving an old TODO
---
 .github/workflows/ci.yml                          | 6 +++---
 docs/tutorials/installation.md                    | 2 +-
 docs/tutorials/saving.md                          | 2 +-
 pyproject.toml                                    | 7 +++----
 src/bluesky_tiled_plugins/writing/tiled_writer.py | 4 +---
 5 files changed, 9 insertions(+), 12 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 4561fa5..f8f3a33 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -54,7 +54,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.13"]
+        python-version: ["3.11", "3.13"]
 
     services:
       mongodb:
@@ -106,7 +106,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.13"]
+        python-version: ["3.11", "3.13"]
 
     steps:
       - uses: actions/checkout@v7
@@ -137,7 +137,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.13"]
+        python-version: ["3.11", "3.13"]
         runs-on: [windows-latest, macos-latest]
 
     steps:
diff --git a/docs/tutorials/installation.md b/docs/tutorials/installation.md
index 019933f..52451c8 100644
--- a/docs/tutorials/installation.md
+++ b/docs/tutorials/installation.md
@@ -8,7 +8,7 @@
 :::{tab-item} pip
 :sync: pip
 
-You will need Python 3.10 or later. You can check your version of Python by
+You will need Python 3.11 or later. You can check your version of Python by
 typing into a terminal:
 
 ```sh
diff --git a/docs/tutorials/saving.md b/docs/tutorials/saving.md
index c0a8d7f..178b30b 100644
--- a/docs/tutorials/saving.md
+++ b/docs/tutorials/saving.md
@@ -13,7 +13,7 @@ server dependencies.
 :::{tab-item} pip
 :sync: pip
 
-You will need Python 3.10 or later. You can check your version of Python by
+You will need Python 3.11 or later. You can check your version of Python by
 typing into a terminal:
 
 ```sh
diff --git a/pyproject.toml b/pyproject.toml
index 8ff15f1..c4d9a2b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -15,7 +15,7 @@ description = "Integrations for using Tiled with Bluesky"
 readme = { file = "README.md", content-type = "text/markdown" }
 license = "BSD-3-Clause"
 license-files = ["LICENSE"]
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 classifiers = [
   "Development Status :: 1 - Planning",
   "Intended Audience :: Science/Research",
@@ -24,7 +24,6 @@ classifiers = [
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3 :: Only",
-  "Programming Language :: Python :: 3.10",
   "Programming Language :: Python :: 3.11",
   "Programming Language :: Python :: 3.12",
   "Programming Language :: Python :: 3.13",
@@ -145,7 +144,7 @@ report.exclude_also = ['\.\.\.', 'if typing.TYPE_CHECKING:']
 
 [tool.mypy]
 files = ["src", "tests"]
-python_version = "3.10"
+python_version = "3.11"
 warn_unused_configs = true
 strict = false
 enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"]
@@ -175,7 +174,7 @@ ignore = [
 
 
 [tool.pylint]
-py-version = "3.10"
+py-version = "3.11"
 ignore-paths = [".*/_version.py"]
 reports.output-format = "colorized"
 similarities.ignore-imports = "yes"
diff --git a/src/bluesky_tiled_plugins/writing/tiled_writer.py b/src/bluesky_tiled_plugins/writing/tiled_writer.py
index 60f3484..be50eb0 100644
--- a/src/bluesky_tiled_plugins/writing/tiled_writer.py
+++ b/src/bluesky_tiled_plugins/writing/tiled_writer.py
@@ -129,9 +129,7 @@ def concatenate_stream_datums(*docs: StreamDatum):
             "All StreamDatum documents must reference the same stream_resource."
         )
     docs = tuple(sorted(docs, key=lambda doc: doc["indices"]["start"]))
-    for d1, d2 in zip(
-        docs[:-1], docs[1:]
-    ):  # TODO: use itertools.pairwise(docs) in python 3.10+
+    for d1, d2 in itertools.pairwise(docs):
         if d1["indices"]["stop"] != d2["indices"]["start"]:
             raise ValueError("StreamDatum documents must be consecutive.")
 

From b56fde3090c8fe6cdaf562468982674887658826 Mon Sep 17 00:00:00 2001
From: genematx <ymatviych@bnl.gov>
Date: Tue, 3 Nov 2026 12:02:59 +1300
Subject: [PATCH 3/6] Remove the httpx2 contract guard test

---
 tests/test_httpx2.py | 14 --------------
 1 file changed, 14 deletions(-)
 delete mode 100644 tests/test_httpx2.py

diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py
deleted file mode 100644
index 7b06a6f..0000000
--- a/tests/test_httpx2.py
+++ /dev/null
@@ -1,14 +0,0 @@
-"""Guard the httpx generation contract with the tiled client.
-
-bluesky-tiled-plugins catches exceptions raised through the tiled client
-(e.g. in the remote-validation 404 fallbacks), so the exception family it
-catches must match the one tiled raises. Since tiled 0.2.21 that is httpx2;
-this test fails loudly if tiled migrates again.
-"""
-
-import httpx2
-from tiled.client.utils import ClientError
-
-
-def test_tiled_client_error_is_catchable_via_httpx2():
-    assert issubclass(ClientError, httpx2.HTTPStatusError)

From 1f98dd58c2bd0bc5313a80b7ce782012bda5db86 Mon Sep 17 00:00:00 2001
From: genematx <ymatviych@bnl.gov>
Date: Tue, 3 Nov 2026 12:15:36 +1300
Subject: [PATCH 4/6] Ignore the SQLAlchemy 2.1 `noload` deprecation warning in
 tests

SQLAlchemy 2.1 deprecates the `noload` loader strategy, which tiled's
catalog ORM uses for the `Webhook.node` relationship. With warnings
raised as errors, the deprecation fails `configure_mappers()` and with
it every test that starts the in-process server. This is a third-party
issue, to be fixed in tiled.
---
 pyproject.toml | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/pyproject.toml b/pyproject.toml
index c4d9a2b..106307b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -133,6 +133,11 @@ filterwarnings = [
   # so they remain readable; zarr warns that these dtypes have no stable Zarr V3
   # specification yet. This warning is emitted by the (in-process) test server.
   "ignore:The data type .* does not have a Zarr V3 specification",
+  # SQLAlchemy 2.1 deprecates the `noload` loader strategy, which tiled's catalog
+  # ORM uses for the Webhook.node relationship. Raising this warning as an error
+  # inside `configure_mappers()` poisons every mapper in the (in-process) test
+  # server; third-party issue, to be fixed in tiled.
+  "ignore:The ``noload`` loader strategy is deprecated:DeprecationWarning",
 ]
 log_level = "INFO"
 testpaths = ["tests"]

From 4cabe0351fb685fa5609bfd799b2e7ccac3ca395 Mon Sep 17 00:00:00 2001
From: genematx <ymatviych@bnl.gov>
Date: Tue, 3 Nov 2026 09:20:29 +1300
Subject: [PATCH 5/6] Vendor BufferingWrapper with flush() from bluesky

Vendor bluesky.callbacks.buffer.BufferingWrapper (following the _dispatcher
precedent) with the addition of the flush() method, contributed upstream on
the bluesky add-buffer-flush branch but not yet released. This lets beamline
profiles adopt the read-after-write synchronization barrier via a
bluesky-tiled-plugins release: documents are processed in a background
thread, so code that reads a run back immediately after RE(plan) returns
(e.g. db[uid]) must call flush() first or it may observe a missing or
incomplete run, or silently pick up the previous one.

The vendored module is self-contained (standard library only) and exported
from the package root, so profiles can import TiledInserter/TiledWriter and
BufferingWrapper from one place. It can be replaced by a re-export once a
bluesky release includes flush().
---
 src/bluesky_tiled_plugins/__init__.py        |   2 +
 src/bluesky_tiled_plugins/writing/_buffer.py | 157 ++++++++++++++++++
 tests/test_buffering.py                      | 161 +++++++++++++++++++
 3 files changed, 320 insertions(+)
 create mode 100644 src/bluesky_tiled_plugins/writing/_buffer.py
 create mode 100644 tests/test_buffering.py

diff --git a/src/bluesky_tiled_plugins/__init__.py b/src/bluesky_tiled_plugins/__init__.py
index 3f2833e..69bc4b0 100644
--- a/src/bluesky_tiled_plugins/__init__.py
+++ b/src/bluesky_tiled_plugins/__init__.py
@@ -6,6 +6,7 @@
     BlueskyStreamUpdate,
     subscribe_to_streams,
 )
+from .writing._buffer import BufferingWrapper
 from .writing.tiled_writer import TiledWriter, TiledInserter
 
 __all__ = [
@@ -13,6 +14,7 @@
     "BlueskyRun",
     "BlueskyStreamSubscription",
     "BlueskyStreamUpdate",
+    "BufferingWrapper",
     "CatalogOfBlueskyRuns",
     "TiledInserter",
     "TiledWriter",
diff --git a/src/bluesky_tiled_plugins/writing/_buffer.py b/src/bluesky_tiled_plugins/writing/_buffer.py
new file mode 100644
index 0000000..663c1e5
--- /dev/null
+++ b/src/bluesky_tiled_plugins/writing/_buffer.py
@@ -0,0 +1,157 @@
+"""
+This module vendors BufferingWrapper from bluesky.callbacks.buffer, with the
+addition of the `flush` method, so that beamline profiles can use the read-after-write
+synchronization and avoids a bluesky dependency.
+
+Once a bluesky release includes `BufferingWrapper.flush`, this copy can be
+replaced by a re-export (or deprecated). The implementation is self-contained
+(standard library only); keep it in sync with the upstream module.
+"""
+
+import atexit
+import logging
+import threading
+import time
+from collections.abc import Callable
+from queue import Empty, Full, Queue
+
+logger = logging.getLogger(__name__)
+
+
+class BufferingWrapper:
+    """A wrapper for callbacks that processes documents in a separate thread.
+
+    This class allows a callback to be executed in a background thread, processing
+    documents as they are received. This prevent the blocking of RunEngine on any
+    slow I/O operations by the callback. It handles graceful shutdown on exit or signal
+    termination, ensuring that no new documents are accepted after shutdown has been
+    initiated.
+
+    The wrapped callback should be thread-safe and not subscribed to the RE directly.
+    If it maintains shared mutable state, it must protect it using internal locking.
+
+    Because documents are processed asynchronously, a run may not be fully written
+    by the time ``RE(plan)`` returns. Code that reads a run back immediately after
+    it completes (e.g. ``db[-1]`` or ``db[uid]``) must call :meth:`flush` first to
+    wait for the buffer to drain; otherwise it may observe a missing or incomplete
+    run, or silently pick up the previous one.
+
+    Parameters
+    ----------
+        target : callable
+            The instance of a callback that will be called with the documents.
+            It should accept two parameters: `name` and `doc`.
+        queue_size : int, optional
+            The maximum size of the internal queue. Default is 1,000,000.
+
+    Usage
+    -----
+        tw = TiledWriter(client)
+        buff_tw = BufferingWrapper(tw)
+        RE.subscribe(buff_tw)
+    """
+
+    def __init__(self, target: Callable, queue_size: int = 1_000_000):
+        self._wrapped_callback = target
+        self._queue: Queue = Queue(maxsize=queue_size)
+        self._stop_event = threading.Event()
+        self._shutdown_lock = threading.Lock()
+
+        self._thread = threading.Thread(target=self._process_queue, daemon=True)
+        self._thread.start()
+
+        atexit.register(self.shutdown)
+
+    def __call__(self, name, doc):
+        if self._stop_event.is_set():
+            raise RuntimeError("Cannot accept new data after shutdown.")
+        try:
+            self._queue.put((name, doc))
+        except Full as e:
+            logger.exception(
+                f"The buffer is full. The {self._wrapped_callback.__class__.__name__} can not keep up "
+                f"with the incoming data. "
+                f"Consider increasing the queue size or optimizing the callback processing: {e}"
+            )
+            raise RuntimeError(
+                f"The buffer is full. The {self._wrapped_callback.__class__.__name__} can not keep up "
+                f"with the incoming data. "
+                "Consider increasing the queue size or optimizing the callback processing."
+            ) from e
+        except Exception as e:
+            logger.exception(f"Failed to put document {name} in queue: {e}")
+            raise RuntimeError(f"Failed to put document {name} in queue: {e}") from e
+
+    def _process_queue(self):
+        while True:
+            try:
+                item = self._queue.get(timeout=1)
+            except Empty:
+                if self._stop_event.is_set():
+                    break
+                continue
+            try:
+                if item is None:
+                    break  # Received sentinel value to stop processing
+                self._wrapped_callback(*item)  # Delegate to wrapped callback
+            except Exception as e:
+                logger.exception(f"Exception in {self._wrapped_callback.__class__.__name__}: {e}")
+            finally:
+                # Account for every retrieved item (including the sentinel and
+                # items whose processing raised), so that `flush` can rely on
+                # the queue's unfinished-task counter.
+                self._queue.task_done()
+
+    def flush(self, timeout: "float | None" = None) -> bool:
+        """Block until all documents enqueued so far have been processed.
+
+        This provides a synchronization barrier for read-after-write patterns:
+        a plan or script that reads a run back (e.g. ``db[uid]``) immediately
+        after ``RE(plan)`` returns must call ``flush()`` first, because the
+        wrapped callback processes documents asynchronously and may not have
+        written the run yet by the time the plan finishes.
+
+        Documents enqueued by another thread *while* flushing extend the wait:
+        the method returns once the queue has no unfinished items.
+
+        Parameters
+        ----------
+        timeout : float, optional
+            Maximum time to wait, in seconds. By default, wait indefinitely.
+
+        Returns
+        -------
+        bool
+            True if the buffer was fully drained; False if the timeout expired
+            (or the worker thread is no longer running) with items remaining.
+        """
+        if timeout is not None:
+            deadline = time.monotonic() + timeout
+        with self._queue.all_tasks_done:
+            while self._queue.unfinished_tasks:
+                if not self._thread.is_alive():
+                    # The worker has exited (e.g. after shutdown); any items
+                    # still in the queue will never be processed.
+                    return not self._queue.unfinished_tasks
+                if timeout is None:
+                    # Wake up periodically to recheck worker liveness.
+                    self._queue.all_tasks_done.wait(timeout=1)
+                else:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        return False
+                    self._queue.all_tasks_done.wait(timeout=min(remaining, 1))
+        return True
+
+    def shutdown(self, wait: bool = True):
+        if self._stop_event.is_set():
+            return
+        self._stop_event.set()
+        self._queue.put(None)
+
+        atexit.unregister(self.shutdown)
+
+        if wait:
+            self._thread.join()
+
+        logger.info(f"{self._wrapped_callback.__class__.__name__} shut down gracefully.")
diff --git a/tests/test_buffering.py b/tests/test_buffering.py
new file mode 100644
index 0000000..3ec12bd
--- /dev/null
+++ b/tests/test_buffering.py
@@ -0,0 +1,161 @@
+"""Tests for the vendored `BufferingWrapper` (see writing/_buffer.py).
+
+Kept in sync with the upstream tests in bluesky (`tests/test_buffering.py`),
+minus the bluesky-specific fixtures.
+"""
+
+import threading
+import time
+
+import pytest
+
+from bluesky_tiled_plugins import BufferingWrapper
+
+
+class SlowDummyCallback:
+    """Simulates a slow (blocking) callback for testing threaded wrappers."""
+
+    def __init__(self, delay=0.1):
+        self.delay = delay
+        self.called = []
+
+    def __call__(self, name, doc):
+        time.sleep(self.delay)
+        self.called.append((name, doc))
+
+
+class CallbackWithException:
+    """A callback that raises an exception when called."""
+
+    def __call__(self, name, doc):
+        raise RuntimeError("This callback always raises an exception.")
+
+
+class BlockingCallback:
+    """A callback that blocks until released, for exercising flush timeouts."""
+
+    def __init__(self):
+        self.release = threading.Event()
+        self.called = []
+
+    def __call__(self, name, doc):
+        self.release.wait()
+        self.called.append((name, doc))
+
+
+@pytest.fixture
+def make_wrapper():
+    """Factory for `BufferingWrapper` instances that are shut down on teardown."""
+    created: list[BufferingWrapper] = []
+
+    def factory(*args, **kwargs):
+        wrapper = BufferingWrapper(*args, **kwargs)
+        created.append(wrapper)
+        return wrapper
+
+    yield factory
+
+    for wrapper in created:
+        wrapper.shutdown()
+
+
+def test_calls_are_delegated(make_wrapper):
+    cb = SlowDummyCallback(delay=0)
+    buff_cb = make_wrapper(cb)
+    buff_cb("start", {"uid": "abc"})
+    assert buff_cb.flush(timeout=5) is True
+    assert cb.called == [("start", {"uid": "abc"})]
+
+
+def test_shutdown_refuses_new_documents(make_wrapper):
+    cb = SlowDummyCallback(delay=0)
+    buff_cb = make_wrapper(cb)
+    buff_cb("one", {})
+    buff_cb.shutdown(wait=True)
+    with pytest.raises(RuntimeError):
+        buff_cb("two", {})
+    assert ("one", {}) in cb.called
+    assert ("two", {}) not in cb.called
+
+
+def test_exception_in_callback_does_not_crash_worker(make_wrapper):
+    buff_cb = make_wrapper(CallbackWithException())
+    buff_cb("event", {})
+    assert buff_cb.flush(timeout=5) is True
+    assert buff_cb._thread.is_alive()
+    buff_cb("event", {})  # still accepting documents
+
+
+def test_flush_waits_for_processing(make_wrapper):
+    """flush() returns only after all enqueued documents are processed."""
+    cb = SlowDummyCallback(delay=0.02)
+    buff_cb = make_wrapper(cb)
+
+    for i in range(10):
+        buff_cb("event", {"seq_num": i})
+
+    assert buff_cb.flush() is True
+    assert len(cb.called) == 10  # all documents delivered, in order
+    assert [doc["seq_num"] for _, doc in cb.called] == list(range(10))
+
+
+def test_flush_empty_buffer_returns_immediately(make_wrapper):
+    """flush() on an empty buffer is an idempotent no-op, however many times
+    it is called, with or without a timeout."""
+    buff_cb = make_wrapper(SlowDummyCallback(delay=0))
+    for _ in range(3):
+        assert buff_cb.flush(timeout=1) is True
+        assert buff_cb.flush() is True
+    assert buff_cb._thread.is_alive()
+
+
+def test_flush_timeout(make_wrapper):
+    """flush() returns False when the buffer can not drain within the timeout."""
+    cb = BlockingCallback()
+    buff_cb = make_wrapper(cb)
+
+    buff_cb("event", {"seq_num": 0})
+    assert buff_cb.flush(timeout=0.2) is False  # worker is blocked
+
+    cb.release.set()
+    assert buff_cb.flush(timeout=5) is True
+    assert len(cb.called) == 1
+
+
+def test_flush_with_raising_callback(make_wrapper):
+    """Documents whose processing raises still count as processed: flush
+    must not hang on them."""
+    buff_cb = make_wrapper(CallbackWithException())
+    for _ in range(5):
+        buff_cb("event", {})
+    assert buff_cb.flush(timeout=5) is True
+    assert buff_cb._thread.is_alive()
+
+
+def test_flush_after_shutdown(make_wrapper):
+    """After a clean shutdown everything has been processed; flush returns True."""
+    cb = SlowDummyCallback(delay=0.01)
+    buff_cb = make_wrapper(cb)
+    for i in range(5):
+        buff_cb("event", {"seq_num": i})
+    buff_cb.shutdown(wait=True)
+
+    assert buff_cb.flush(timeout=1) is True
+    assert buff_cb.flush(timeout=1) is True  # repeated calls remain safe
+    assert len(cb.called) == 5
+
+
+def test_read_after_write_pattern(make_wrapper):
+    """The canonical use: write documents, flush, then read back."""
+    store = {}
+
+    def writer(name, doc):
+        if name == "stop":
+            store[doc["run_start"]] = "complete"
+
+    buff_cb = make_wrapper(writer)
+    buff_cb("start", {"uid": "abc"})
+    buff_cb("stop", {"run_start": "abc"})
+
+    assert buff_cb.flush(timeout=5) is True
+    assert store.get("abc") == "complete"  # guaranteed visible after flush

From 7c95b46c3f800b35279fd1a4359a8ef0a4f589ed Mon Sep 17 00:00:00 2001
From: genematx <ymatviych@bnl.gov>
Date: Tue, 3 Nov 2026 09:35:33 +1300
Subject: [PATCH 6/6] MNT: format and lint

---
 src/bluesky_tiled_plugins/writing/_buffer.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/src/bluesky_tiled_plugins/writing/_buffer.py b/src/bluesky_tiled_plugins/writing/_buffer.py
index 663c1e5..8b888b4 100644
--- a/src/bluesky_tiled_plugins/writing/_buffer.py
+++ b/src/bluesky_tiled_plugins/writing/_buffer.py
@@ -95,7 +95,9 @@ def _process_queue(self):
                     break  # Received sentinel value to stop processing
                 self._wrapped_callback(*item)  # Delegate to wrapped callback
             except Exception as e:
-                logger.exception(f"Exception in {self._wrapped_callback.__class__.__name__}: {e}")
+                logger.exception(
+                    f"Exception in {self._wrapped_callback.__class__.__name__}: {e}"
+                )
             finally:
                 # Account for every retrieved item (including the sentinel and
                 # items whose processing raised), so that `flush` can rely on
@@ -154,4 +156,6 @@ def shutdown(self, wait: bool = True):
         if wait:
             self._thread.join()
 
-        logger.info(f"{self._wrapped_callback.__class__.__name__} shut down gracefully.")
+        logger.info(
+            f"{self._wrapped_callback.__class__.__name__} shut down gracefully."
+        )