Add Tiled streaming API specific to Bluesky specs

[thopkins32]

Assisted-by: oh-my-pi:gpt-5.6-terra

See #64 (comment) for changes discussed based on review.

Summary

Closes #6

Implements a bluesky_tiled_plugins.streaming module that exposes public interfaces:

• BlueskyStreamUpdate: a dataclass wrapping a Tiled update that contains additional information
• BlueskyStreamSubscription is a manager class for handling the recursive subscriptions and complicated disconnect behavior, it is a contextmanager
• subscribe_to_stream is the primary entry-point, returning a BlueskyStreamSubscription
• subscribe_to_stream_by_metadata is the same function with explicit filtering by metadata
• subscribe_to_stream_by_specs is the same function with explicit filtering by non-empty spec set intersection
• subscribe_to_stream_filtered is the same function for more custom filtering
• TiledWriter.stop now closes the stream tree of all nodes it created in order to notify consumers that no more updates are expected, the websocket connection can be closed, and subscriber threads can be released.

Tests are written in tests/test_streaming.py.

Docs are written in docs/tutorials/streaming.md.

There were other miscellaneous formatting fixes applied for some reason.

Design Choices

Subscriptions are created as follows:

catalog
└── BlueskyRun
   └── named BlueskyEventStream
       ├── appendable table(s)
       └── named array child(ren)

There is no buffering, synthetic joining, row alignment, or missing-value policy. That is up to the implementer of the callback to deal with.

Lifecycle is handled via TiledWriter explicitly declaring a stream closed, an explicit disconnect, or the context manager (which calls disconnect on exit). disconnect blocks until all subscriptions disconnected, done in reverse creation order, however Tiled's client natively handles that.

Testing

• Tested with a real workflow (see comment below)

Future Work

A narrower notification streaming API that simply notifies when a run starts and stops (or when a stream opens and closes), this could be useful when some code needs to react to a run stopping and act on the full batch of run data (which can be loaded directly from .read())

[codecov]

Codecov Report

❌ Patch coverage is 84.82759% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 78.38%. Comparing base (9b2d192) to head (1332834).
⚠️ Report is 8 commits behind head on main.

Files with missing lines	Patch %	Lines
src/bluesky_tiled_plugins/streaming.py	87.30%	16 Missing ⚠️
src/bluesky_tiled_plugins/writing/tiled_writer.py	66.66%	6 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main      #64      +/-   ##
==========================================
+ Coverage   78.23%   78.38%   +0.14%     
==========================================
  Files          15       16       +1     
  Lines        2343     2387      +44     
==========================================
+ Hits         1833     1871      +38     
- Misses        510      516       +6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jwlodek]

I like that it's a context manager.

Add ability to filter based on specs or specific metadata in the start document

I think we should aim to include this in the first iteration of this - I think the most common use case here will be to spin up a service that makes this subscription, that will only process a subset of runs. So some kind of filtering I think is key. Now that #60 is merged and any release with this streaming feature should include that, I think just adding a way to filter by specs should be sufficient (and easy to implement I think?)

[thopkins32]

Another API for this might simply be a notification channel. For example, I may want to be able to simply know when a Bluesky run ended and capture its UID, so I can query Tiled normally for the data. The alternative to this would be to repeatedly poll the catalog for new runs and not use websockets at all.

Question for Tiled developers reviewing this work: does a 'stream_closed' update on the websocket guarantee that that same data is queryable via something like tiled_client[uid]["primary/some_signal"].read()? My gut tells me "no" since it would probably delay the final update too much, but I would be happy if the answer was "yes" and final update delivery time was not significantly impacted.

[thopkins32]

Tested this out in the linked PR to tst-sim-tools and can show that the thread cleanup is behaving nicely. I've run ~20 runs all from the same initial subscription. If cleanup wasn't working, then I would expect there to be many more active threads with tiled subscriptions.

This was done with a local Tiled server, as you can tell from the thread name.

This image shows during a run:

This image shows after the run has stopped (still one catalog-level subscription):

[danielballan]

This is great to see, @thopkins32! Ever since you opened #6 I've agreed this is very needed. Thank you for taking it on. The API design looks great. Seconding @jwlodek, I love the context manager.

I left one question in line. It's based on an admittedly speedy read, so I've missed the point feel free to disregard.

Regarding a "notification" channel, it's interesting to consider how this would complement the web hooks feature added in Spring 2026. Webhooks serve the notification use case well because it does not require a long-lived connection (like a web socket). However, it does require a running HTTP endpoint that the server can reach. If the recipient is, say, running in a Jupyter notebook, a websocket option would be good to have.

Question for Tiled developers reviewing this work: does a 'stream_closed' update on the websocket guarantee that that same data is queryable via something like tiled_client[uid]["primary/some_signal"].read()? My gut tells me "no" since it would probably delay the final update too much, but I would be happy if the answer was "yes" and final update delivery time was not significantly impacted.

It would be nice! But, indeed, unfortunately not. To be precise:

• Changes to the catalog only (node creation or metadata update) are committed before the update is broadcast. Thus, when a streaming client is told about a new/updated node, and then issues a REST request about it, the server should never answer with a 404 ("What node?"). That is---except in some odd case where someone deleted the node in the mean time.
• Uploaded data is streamed before it is written to disk. The point is to get it to clients, via Redis, before we wait for disks.
• Registered data (e..g., data written by Area Detector) has an additional race condition. Ophyd[-async] and Bluesky report to Tiled that the detector was told to write data, but it may still be writing, or the network storage appliance may still be in the progress of achieving consistency. There is no way around this.

[thopkins32]

@danielballan thanks!

Registered data (e..g., data written by Area Detector) has an additional race condition. Ophyd[-async] and Bluesky report to Tiled that the detector was told to write data, but it may still be writing, or the network storage appliance may still be in the progress of achieving consistency. There is no way around this.

Ah, I was always told not to emit these docs until the data was actually written. I get that Tiled can't assume this to be true, generally speaking.

Overall, I think it's fine, it typically shouldn't be a long delay between seeing that a data stream is closed and the data being persisted to Tiled's queryable storage. And if there is a long delay, then it's likely there is a wider problem with file writing or network issues.

[jwlodek]

In ophyd async at least we guarantee for ADs that the docs are only emitted once the data has been flushed to disk, although the storage side of things of course still remains a consideration.

[danielballan]

That is good to know. Does ophyd-async block [on] WaitForPlugins before adding a stream datum to the cache returned by collect_asset_docs?

[genematx]

One of the key ideas of bluesky-tiled-plugins is to blur the distinction between data_keys stored in tables (i.e. those values coming from Event documents, written to the internal table) and arrays (either internal or external). I wonder if we should be converting LiveTableData into LiveArrayData too.

[thopkins32]

I'm not sure I like that approach. If things are already bundled into discrete events then I would like them to maintain that coupling. If we split into LiveArrayData for every field in LiveTableData, I would likely put it back into that same table structure as the first step of my callback.

[genematx]

I agree that it's good to keep the bundling. I just want to flag that the user should not be able to see (and will not, with the plugins client) the internal table directly. Therefore, it might be confusing to receive table updates.
Perhaps, we need to rething how we structure bluesky data in the BlueskyRun (consolidated) container.

[jwlodek]

That is good to know. Does ophyd-async block [on] WaitForPlugins before adding a stream datum to the cache returned by collect_asset_docs?

We only WaitForPlugins if we are also reading out signal data from the stats plugin for example, instead for the datum docs we rely on watching the frames written by the file plugin rather than the detector acquisition counter, and before generating the doc we call Flush on the file plugin which is a busy record that guarantees that the reported number of frames captured are all on-disk before it clears.

Similarly we emit watcher updates on frames written to disk rather than for frames collected by the detector.

[jwlodek]

Generally looks really good to me. I have a use case at IOS that I plan on giving this a spin with.

One thing I'd also be interested in is the ability to just have the stream fetch all the datakeys in the stream without me having to know them ahead of time. Or at least to have a way to determine which keys are interesting automatically somehow (something like hints).

[thopkins32]

Thanks for all the feedback, it has really helped me think through this in more detail. I overfit my implementation to my specific use-case a bit too much (single Bluesky data stream, metadata context required through updates)

Here is what I will do before the next review round:

• Remove start document freezing and passing around in favor of callbacks querying Tiled for it directly (from the run_uid)
• Collapse down to a single function subscribe_to_streams, now plural for multi-stream uses, with combined run filtering optional args
• Update docs with typing and clearer descriptions
• Add None as a select-all option for data keys
• Add tests for ragged and bytes data