Repository: bluesky/bluesky
Pull request #1542 (2022-08-23 23:30:59)

Add ability to pre-declare a stream
In discussions with PSI (SLS2) at NSLS-II today it became clear that we need the ability to pre-declare what the streams will be before we create the first event.  This will have at least three and a half major benefits

1. it makes the first measurement less special as we can ensure that the describe / read_configuration / describe_configuration happen when we think they wil
2. they will enable the bundler to live in a different process and still correctly accumulate all of the notifications it will need
3. it is now possible to get the descriptor is a custom plan without subscribing to the output stream
4. the RunBundler class is now pluggable by class level monkey patching or thin sub-classing (could be convinced to make it init time configurable...)

In local testing I added a "strict mode" flag to demand that streams are pre-declared and many tests still passed.

tacaswell: Open questions:

 - do we want this?
 - how to manage "strict mode" configuration (probably want 3 modes: YOLO, warn, strict)
 - how to thread the stream name through `count` and the `scan_nd` family of plans because they take an arbitrary user function than can use a different stream name that 'primary' ?
 - how configurable do we want to make the RunBundler class on the RE?
danielballan: I think we do want it.

I think we should start with "YOLO" mode only and get some experience with this before we add either "warn" or "strict".

I think that we can defer doing anything helpful or clever with `per_step`. If anything but the default `per_step` is used, `count` and `scan_nd` can drop the declaration. Users who want both a  declaration and a custom `per_step` have to write a custom plan. As we get experience with this we can consider adding another parameter to deal with this.

I think class attribute is a fine place to start.
coretl: Also, what does "YOLO" mean?
tacaswell: It looks like changes to ophyd broke some of the tests?
coretl: I'm guessing https://github.com/bluesky/ophyd/pull/1036 might have something to do with it? @callumforrester do you recognise the error?

cjtitus: I'm interested in getting this merged. I can try to look at why tests were failing, but as the logs have disappeared it would be helpful to re-run the tests first.

I don't think I can tell GH to re-run the tests, but someone has the power to do that, right?
danielballan: Power-cycled to re-run CI
cjtitus: Well it looks like turning it off and then back on has "fixed" the previously failing tests. Were the previously open questions ever answered satisfactorily?  

> Open questions:
> 
>     * do we want this?
> 
>     * how to manage "strict mode" configuration (probably want 3 modes: YOLO, warn, strict)
> 
>     * how to thread the stream name through `count` and the `scan_nd` family of plans because they take an arbitrary user function than can use a different stream name that 'primary' ?
> 
>     * how configurable do we want to make the RunBundler class on the RE?

I can at least answer the first question and say yes, I want this. 
The rest of these were addressed by @danielballan 

In order to manage `count` and `scan_nd` I think the existing code needs to be changed to avoid pre-declaring a stream if a non-default `per_shot` is passed in. There is already a check for `if per_shot is None` to assign the default. This could be when the stream pre-declaration is toggled.
cjtitus: Another important question is if strict checking option should be left in `RunBundler`, but `False` by default, or if it should be removed entirely for now. If it's `False`, there's really no way to ever turn it on, unless an option for strict checking were to be added to the RunEngine that could be passed to RunBundler instantiation during `open_run`

Maybe this is a plausible enough future addition that the code for strict checking ought to be left in?
tacaswell: Thank you @cjtitus for pushing on this!

One change I would like to make is to change the `strict_pre_declare` to an enum rather than a bool (with values "strict", "warn", and "permissive" and change the parameter to "stream_pre_declare_policy" or something like that), but it is not a disaster if it goes in as a bool (and we can promote later it is just _annoying_ to do so).

Patch:
diff --git a/bluesky/bundlers.py b/bluesky/bundlers.py
index a3850ea1..11fc0965 100644
--- a/bluesky/bundlers.py
+++ b/bluesky/bundlers.py
@@ -24,7 +24,9 @@ ObjDict = Dict[Any, Dict[str, T]]
 
 
 class RunBundler:
-    def __init__(self, md, record_interruptions, emit, emit_sync, log):
+    def __init__(self, md, record_interruptions, emit, emit_sync, log, *, strict_pre_declare=False):
+        # if create can YOLO implicitly create a stream
+        self._strict_pre_declare = strict_pre_declare
         # state stolen from the RE
         self.bundling = False  # if we are in the middle of bundling readings
         self._bundle_name = None  # name given to event descriptor
@@ -131,9 +133,73 @@ class RunBundler:
         self.run_is_open = False
         return doc["run_start"]
 
+    async def _prepare_stream(self, desc_key, objs_read):
+        # We do not have an Event Descriptor for this set
+        # so one must be created.
+        data_keys = {}
+        config = {}
+        object_keys = {}
+        hints = {}
+        for obj in objs_read:
+            dks = self._describe_cache[obj]
+            obj_name = obj.name
+            # dks is an OrderedDict. Record that order as a list.
+            object_keys[obj_name] = list(dks)
+            for field, dk in dks.items():
+                dk["object_name"] = obj_name
+            data_keys.update(dks)
+            config[obj_name] = {}
+            config[obj_name]["data"] = self._config_values_cache[obj]
+            config[obj_name]["timestamps"] = self._config_ts_cache[obj]
+            config[obj_name]["data_keys"] = self._config_desc_cache[obj]
+            maybe_update_hints(hints, obj)
+        descriptor_uid = new_uid()
+        descriptor_doc = dict(
+            run_start=self._run_start_uid,
+            time=ttime.time(),
+            data_keys=data_keys,
+            uid=descriptor_uid,
+            configuration=config,
+            name=desc_key,
+            hints=hints,
+            object_keys=object_keys,
+        )
+        await self.emit(DocumentNames.descriptor, descriptor_doc)
+        doc_logger.debug(
+            "[descriptor] document emitted with name %r containing "
+            "data keys %r (run_uid=%r)",
+            desc_key,
+            data_keys.keys(),
+            self._run_start_uid,
+            extra={
+                'doc_name': 'descriptor',
+                'run_uid': self._run_start_uid,
+                'data_keys': data_keys.keys()}
+        )
+        self._descriptors[desc_key] = (objs_read, descriptor_doc)
+        return descriptor_doc
+
+    async def _ensure_cached(self, obj):
+        if obj not in self._describe_cache:
+            await self._cache_describe(obj)
+        if obj not in self._config_desc_cache:
+            await self._cache_describe_config(obj)
+            await self._cache_read_config(obj)
+
+    async def declare_stream(self, msg):
+        """Generate and emit an EventDescriptor."""
+        command, no_obj, objs, kwargs, _ = msg
+        stream_name = kwargs['name']
+        assert no_obj is None
+        objs = frozenset(objs)
+        for obj in objs:
+            await self._ensure_cached(obj)
+
+        return (await self._prepare_stream(stream_name, objs))
+
     async def create(self, msg):
-        """Trigger the run engine to start bundling future obj.read() calls for
-         an Event document
+        """
+        Start bundling future obj.read() calls for an Event document.
 
         Expected message object is::
 
@@ -168,6 +234,11 @@ class RunBundler:
                     "Msg('create') now requires a stream name, given as "
                     "Msg('create', name) or Msg('create', name=name)"
                 ) from None
+        if self._strict_pre_declare:
+            if self._bundle_name not in self._descriptors:
+                raise IllegalMessageSequence(
+                    "In strict mode you must pre-declare streams."
+                )
 
     async def read(self, msg, reading):
         """
@@ -185,11 +256,7 @@ class RunBundler:
             # on the same device you make obj.describe() calls multiple times.
             # As this is harmless and not an expected use case, we don't guard
             # against it. Reading multiple devices concurrently works fine.
-            if obj not in self._describe_cache:
-                await self._cache_describe(obj)
-            if obj not in self._config_desc_cache:
-                await self._cache_describe_config(obj)
-                await self._cache_read_config(obj)
+            await self._ensure_cached(obj)
 
             # check that current read collides with nothing else in
             # current event
@@ -274,37 +341,11 @@ class RunBundler:
                 "A 'monitor' message was sent for {}"
                 "which is already monitored".format(obj)
             )
-        descriptor_uid = new_uid()
-        if obj not in self._describe_cache:
-            await self._cache_describe(obj)
-        if obj not in self._config_desc_cache:
-            await self._cache_describe_config(obj)
-            await self._cache_read_config(obj)
-        data_keys = self._describe_cache[obj]
-        config = {obj.name: {
-            "data": self._config_values_cache[obj],
-            "timestamps": self._config_ts_cache[obj],
-            "data_keys": self._config_desc_cache[obj]
-        }}
-        object_keys = {obj.name: list(data_keys)}
-        hints = {}
-        maybe_update_hints(hints, obj)
-        desc_doc = dict(
-            run_start=self._run_start_uid,
-            time=ttime.time(),
-            data_keys=data_keys,
-            uid=descriptor_uid,
-            configuration=config,
-            hints=hints,
-            name=name,
-            object_keys=object_keys,
-        )
-        doc_logger.debug("[descriptor] document is emitted with name %r containing "
-                         "data keys %r (run_uid=%r)", name, data_keys.keys(),
-                         self._run_start_uid,
-                         extra={'doc_name': 'descriptor',
-                                'run_uid': self._run_start_uid,
-                                'data_keys': data_keys.keys()})
+
+        await self._ensure_cached(obj)
+
+        desc_doc = await self._prepare_stream(name, (obj,))
+
         seq_num_counter = count(1)
 
         def emit_event(readings: Dict[str, Reading] = None, *args, **kwargs):
@@ -323,7 +364,7 @@ class RunBundler:
                     "passed to subscribe() was not called with Dict[str, Reading]"
             data, timestamps = _rearrange_into_parallel_dicts(readings)
             doc = dict(
-                descriptor=descriptor_uid,
+                descriptor=desc_doc['uid'],
                 time=ttime.time(),
                 data=data,
                 timestamps=timestamps,
@@ -333,7 +374,6 @@ class RunBundler:
             self.emit_sync(DocumentNames.event, doc)
 
         self._monitor_params[obj] = emit_event, kwargs
-        await self.emit(DocumentNames.descriptor, desc_doc)
         # TODO: deprecate **kwargs when Ophyd.v2 is available
         obj.subscribe(emit_event, **kwargs)
 
@@ -425,56 +465,15 @@ class RunBundler:
         self._bundle_name = None
 
         d_objs, descriptor_doc = self._descriptors.get(desc_key, (None, None))
-        if d_objs is not None and d_objs != objs_read:
+        # we do not have the descriptor cached, make it
+        if descriptor_doc is None:
+            descriptor_doc = await self._prepare_stream(desc_key, objs_read)
+        # do have the descriptor cached
+        elif d_objs != objs_read:
             raise RuntimeError(
                 "Mismatched objects read, expected {!s}, "
                 "got {!s}".format(d_objs, objs_read)
             )
-        if descriptor_doc is None:
-            # We do not have an Event Descriptor for this set
-            # so one must be created.
-            data_keys = {}
-            config = {}
-            object_keys = {}
-            hints = {}
-            for obj in objs_read:
-                dks = self._describe_cache[obj]
-                obj_name = obj.name
-                # dks is an OrderedDict. Record that order as a list.
-                object_keys[obj_name] = list(dks)
-                for field, dk in dks.items():
-                    dk["object_name"] = obj_name
-                data_keys.update(dks)
-                config[obj_name] = {}
-                config[obj_name]["data"] = self._config_values_cache[obj]
-                config[obj_name]["timestamps"] = self._config_ts_cache[obj]
-                config[obj_name]["data_keys"] = self._config_desc_cache[obj]
-                maybe_update_hints(hints, obj)
-            descriptor_uid = new_uid()
-            descriptor_doc = dict(
-                run_start=self._run_start_uid,
-                time=ttime.time(),
-                data_keys=data_keys,
-                uid=descriptor_uid,
-                configuration=config,
-                name=desc_key,
-                hints=hints,
-                object_keys=object_keys,
-            )
-            await self.emit(DocumentNames.descriptor, descriptor_doc)
-            doc_logger.debug(
-                "[descriptor] document emitted with name %r containing "
-                "data keys %r (run_uid=%r)",
-                obj_name,
-                data_keys.keys(),
-                self._run_start_uid,
-                extra={
-                    'doc_name': 'descriptor',
-                    'run_uid': self._run_start_uid,
-                    'data_keys': data_keys.keys()}
-            )
-            self._descriptors[desc_key] = (objs_read, descriptor_doc)
-
         descriptor_uid = descriptor_doc["uid"]
 
         # Resource and Datum documents
@@ -777,4 +776,7 @@ class RunBundler:
             obj_set, _ = self._descriptors[name]
             if obj in obj_set:
                 del self._descriptors[name]
+                await self._prepare_stream(name, obj_set)
+                continue
+
         await self._cache_read_config(obj)
diff --git a/bluesky/examples.py b/bluesky/examples.py
index a095402a..86a43626 100644
--- a/bluesky/examples.py
+++ b/bluesky/examples.py
@@ -137,6 +137,7 @@ def conditional_pause(det, motor, defer, include_checkpoint):
 def simple_scan_saving(det, motor):
     "Set, trigger, read"
     yield Msg('open_run')
+    yield Msg('declare_stream', None, motor, det, name='primary')
     yield Msg('create', name='primary')
     yield Msg('set', motor, 5)
     yield Msg('read', motor)
@@ -148,6 +149,7 @@ def simple_scan_saving(det, motor):
 
 def stepscan(det, motor):
     yield Msg('open_run')
+    yield Msg('declare_stream', None, motor, det, name='primary')
     for i in range(-5, 5):
         yield Msg('create', name='primary')
         yield Msg('set', motor, i)
@@ -160,6 +162,7 @@ def stepscan(det, motor):
 
 def cautious_stepscan(det, motor):
     yield Msg('open_run')
+    yield Msg('declare_stream', None, motor, det, name='primary')
     for i in range(-5, 5):
         yield Msg('checkpoint')
         yield Msg('create', name='primary')
@@ -210,6 +213,7 @@ def multi_sample_temperature_ramp(detector, sample_names, sample_positions,
             detector.center = peak_pos
             detector.sigma = .5 + .25 * idx
             detector.noise_factor = .05 + idx * 0.1
+            yield Msg('declare_stream', None, scan_motor, detector, temp_controller, name='primary')
             for scan_pos in np.arange(start, stop, step):
                 yield Msg('set', scan_motor, scan_pos)
                 # be super paranoid about the temperature. Grab it before and
diff --git a/bluesky/plan_stubs.py b/bluesky/plan_stubs.py
index 54f9b5c2..17a808d6 100644
--- a/bluesky/plan_stubs.py
+++ b/bluesky/plan_stubs.py
@@ -25,6 +25,28 @@ from .utils import (
 )
 
 
+def declare_stream(*objs, name):
+    """
+    Bundle future readings into a new Event document.
+
+    Parameters
+    ----------
+    name : string, optional
+        name given to event stream, used to convenient identification
+        default is 'primary'
+
+    Yields
+    ------
+    msg : Msg
+        Msg('create', name=name)
+
+    See Also
+    --------
+    :func:`bluesky.plan_stubs.save`
+    """
+    return (yield Msg('declare_stream', None, *separate_devices(objs), name=name))
+
+
 def create(name='primary'):
     """
     Bundle future readings into a new Event document.
diff --git a/bluesky/plans.py b/bluesky/plans.py
index 9c060e6e..1aff6203 100644
--- a/bluesky/plans.py
+++ b/bluesky/plans.py
@@ -65,12 +65,16 @@ def count(detectors, num=1, delay=None, *, per_shot=None, md=None):
     _md.update(md or {})
     _md['hints'].setdefault('dimensions', [(('time',), 'primary')])
 
+    # per_shot might define a different stream, so do not predeclare primary
+    predeclare = (per_shot is None)
     if per_shot is None:
         per_shot = bps.one_shot
 
     @bpp.stage_decorator(detectors)
     @bpp.run_decorator(md=_md)
     def inner_count():
+        if predeclare:
+            yield from bps.declare_stream(*detectors, name='primary')
         return (yield from bps.repeat(partial(per_shot, detectors),
                                       num=num, delay=delay))
 
@@ -510,6 +514,7 @@ def log_scan(detectors, motor, start, stop, num, *, per_step=None, md=None):
     else:
         _md['hints'].setdefault('dimensions', dimensions)
 
+    predeclare = (per_step is None)
     if per_step is None:
         per_step = bps.one_1d_step
 
@@ -518,6 +523,8 @@ def log_scan(detectors, motor, start, stop, num, *, per_step=None, md=None):
     @bpp.stage_decorator(list(detectors) + [motor])
     @bpp.run_decorator(md=_md)
     def inner_log_scan():
+        if predeclare:
+            yield from bps.declare_stream(motor, *detectors, name='primary')
         for step in steps:
             yield from per_step(detectors, motor, step)
 
@@ -637,6 +644,8 @@ def adaptive_scan(detectors, target_field, motor, start, stop,
             direction_sign = 1
         else:
             direction_sign = -1
+        devices = tuple(utils.separate_devices(detectors + [motor]))
+        yield from bps.declare_stream(*devices, name='primary')
         while next_pos * direction_sign < stop * direction_sign:
             yield Msg('checkpoint')
             yield from bps.mv(motor, next_pos)
@@ -644,7 +653,7 @@ def adaptive_scan(detectors, target_field, motor, start, stop,
             for det in detectors:
                 yield Msg('trigger', det, group='B')
             yield Msg('wait', None, 'B')
-            for det in utils.separate_devices(detectors + [motor]):
+            for det in devices:
                 cur_det = yield Msg('read', det)
                 if target_field in cur_det:
                     cur_I = cur_det[target_field]['value']
@@ -830,6 +839,7 @@ def tune_centroid(
         sum_I = 0       # for peak centroid calculation, I(x)
         sum_xI = 0
 
+        yield from bps.declare_stream(motor, *detectors, name='primary')
         while abs(step) >= min_step and low_limit <= next_pos <= high_limit:
             yield Msg('checkpoint')
             yield from bps.mv(motor, next_pos)
@@ -921,6 +931,7 @@ def scan_nd(detectors, cycler, *, per_step=None, md=None):
         # change it, else set it to the one generated above
         _md['hints'].setdefault('dimensions', dimensions)
 
+    predeclare = (per_step is None)
     if per_step is None:
         per_step = bps.one_nd_step
     else:
@@ -994,6 +1005,8 @@ def scan_nd(detectors, cycler, *, per_step=None, md=None):
     @bpp.stage_decorator(list(detectors) + motors)
     @bpp.run_decorator(md=_md)
     def inner_scan_nd():
+        if predeclare:
+            yield from bps.declare_stream(*motors, *detectors, name='primary')
         for step in list(cycler):
             yield from per_step(detectors, step, pos_cache)
 
diff --git a/bluesky/preprocessors.py b/bluesky/preprocessors.py
index f3631b44..207f89cc 100644
--- a/bluesky/preprocessors.py
+++ b/bluesky/preprocessors.py
@@ -11,7 +11,7 @@ from .utils import (get_hinted_fields, normalize_subs_input, root_ancestor,
                     short_uid as _short_uid, make_decorator,
                     RunEngineControlException, merge_axis)
 from functools import wraps
-from .plan_stubs import (open_run, close_run, mv, pause, trigger_and_read)
+from .plan_stubs import (open_run, close_run, mv, pause, trigger_and_read, declare_stream)
 
 
 def plan_mutator(plan, msg_proc):
@@ -1155,9 +1155,13 @@ def baseline_wrapper(plan, devices, name='baseline'):
     msg : Msg
         messages from plan, with 'set' messages inserted
     """
+    def head():
+        yield from declare_stream(*devices, name=name)
+        yield from trigger_and_read(devices, name=name)
+
     def insert_baseline(msg):
         if msg.command == 'open_run':
-            return None, trigger_and_read(devices, name=name)
+            return None, head()
 
         elif msg.command == 'close_run':
             def post_baseline():
diff --git a/bluesky/run_engine.py b/bluesky/run_engine.py
index 59095432..e552726a 100644
--- a/bluesky/run_engine.py
+++ b/bluesky/run_engine.py
@@ -330,6 +330,8 @@ class RunEngine:
                              'close_run', 'install_suspender',
                              'remove_suspender', '_start_suspender']
 
+    RunBundler = RunBundler
+
     @property
     def state(self):
         return self._state
@@ -448,7 +450,9 @@ class RunEngine:
         self._task_fut = None  # future proxy to the task above
         self._pardon_failures = None  # will hold an asyncio.Event
         self._plan = None  # the plan instance from __call__
+        self._require_stream_declaration = False
         self._command_registry = {
+            'declare_stream': self._declare_stream,
             'create': self._create,
             'save': self._save,
             'drop': self._drop,
@@ -1768,8 +1772,9 @@ class RunEngine:
         # against users mutating the md with their validator.
         self.md_validator(dict(md))
 
-        current_run = self._run_bundlers[run_key] = RunBundler(
-            md, self.record_interruptions, self.emit, self.emit_sync, self.log
+        current_run = self._run_bundlers[run_key] = type(self).RunBundler(
+            md, self.record_interruptions, self.emit, self.emit_sync, self.log,
+            strict_pre_declare=self._require_stream_declaration
         )
 
         new_uid = await current_run.open_run(msg)
@@ -1822,6 +1827,30 @@ class RunEngine:
                                          "be preceded by 'open_run'.") from ke
         return (await current_run.create(msg))
 
+    async def _declare_stream(self, msg):
+        """Trigger the run engine to start bundling future obj.read() calls for
+         an Event document
+
+        Expected message object is:
+
+            Msg('create', None, name='primary')
+            Msg('create', name='primary')
+
+        Note that the `name` kwarg will be the 'name' field of the resulting
+        descriptor. So descriptor['name'] = msg.kwargs['name'].
+
+        Also note that changing the 'name' of the Event will create a new
+        Descriptor document.
+        """
+        run_key = msg.run
+        try:
+            current_run = self._run_bundlers[run_key]
+        except KeyError as ke:
+            raise IllegalMessageSequence("Cannot bundle readings without "
+                                         "an open run. That is, 'create' must "
+                                         "be preceded by 'open_run'.") from ke
+        return (await current_run.declare_stream(msg))
+
     async def _read(self, msg):
         """
         Add a reading to the open event bundle.
diff --git a/docs/source/conf.py b/docs/source/conf.py
index ec2e9b3c..32ee2884 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -314,10 +314,10 @@ texinfo_documents = [
 
 intersphinx_mapping = {
     'python': ('https://docs.python.org/3/', None),
-    'numpy': ('https://docs.scipy.org/doc/numpy/', None),
-    'scipy': ('https://docs.scipy.org/doc/scipy/reference/', None),
+    'numpy': ('https://numpy.org/doc/stable', None),
+    'scipy': ('https://docs.scipy.org/doc/scipy/', None),
     'pandas': ('https://pandas.pydata.org/pandas-docs/stable', None),
-    'matplotlib': ('https://matplotlib.org', None),
+    'matplotlib': ('https://matplotlib.org/stable/', None),
     'ophyd': ('https://blueskyproject.io/ophyd/', None),
     'event-model': ('https://blueskyproject.io/event-model/', None),
   }
diff --git a/docs/source/examples/multi_run_plans_nested.py b/docs/source/examples/multi_run_plans_nested.py
index 26a14e9b..ffc073ca 100644
--- a/docs/source/examples/multi_run_plans_nested.py
+++ b/docs/source/examples/multi_run_plans_nested.py
@@ -15,25 +15,31 @@ RE = RunEngine({})
 db = Broker.named("temp")
 RE.subscribe(db.insert)
 
+
 def factory(name, doc):
     # Documents from each run is routed to an independent
     #   instance of BestEffortCallback
     bec = BestEffortCallback()
     return [bec], []
 
+
 rr = RunRouter([factory])
 RE.subscribe(rr)
 
+
 @bpp.set_run_key_decorator("run_2")
 @bpp.run_decorator(md={})
 def sim_plan_inner(npts):
+    yield from bps.declare_stream(hw.motor1, hw.motor2, hw.det2, name='primary')
     for j in range(npts):
         yield from bps.mov(hw.motor1, j * 0.1 + 1, hw.motor2, j * 0.2 - 2)
         yield from bps.trigger_and_read([hw.motor1, hw.motor2, hw.det2])
 
+
 @bpp.set_run_key_decorator("run_1")
 @bpp.run_decorator(md={})
 def sim_plan_outer(npts):
+    yield from bps.declare_stream(hw.motor, hw.det, name='primary')
     for j in range(int(npts/2)):
         yield from bps.mov(hw.motor, j * 0.2)
         yield from bps.trigger_and_read([hw.motor, hw.det])
diff --git a/docs/source/examples/multi_run_plans_recursive.py b/docs/source/examples/multi_run_plans_recursive.py
index 7fc75b56..c2f6c495 100644
--- a/docs/source/examples/multi_run_plans_recursive.py
+++ b/docs/source/examples/multi_run_plans_recursive.py
@@ -15,17 +15,20 @@ RE = RunEngine({})
 db = Broker.named("temp")
 RE.subscribe(db.insert)
 
+
 def factory(name, doc):
     # Each run is subscribed to independent instance of BEC
     bec = BestEffortCallback()
     return [bec], []
 
+
 rr = RunRouter([factory])
 RE.subscribe(rr)
 
 # Call counter and the maximum number calls
 n_calls, n_calls_max = 0, 3
 
+
 def sim_plan_recursive(npts):
     global n_calls, n_calls_max
 
@@ -39,7 +42,7 @@ def sim_plan_recursive(npts):
         @bpp.set_run_key_decorator(run_key)
         @bpp.run_decorator(md={})
         def plan(npts):
-
+            yield from bps.declare_stream(hw.motor1, hw.det1, name='primary')
             for j in range(int(npts/2)):
                 yield from bps.mov(hw.motor1, j * 0.2)
                 yield from bps.trigger_and_read([hw.motor1, hw.det1])
diff --git a/docs/source/examples/multi_run_plans_select_cb.py b/docs/source/examples/multi_run_plans_select_cb.py
index fce231c6..59f1ca8e 100644
--- a/docs/source/examples/multi_run_plans_select_cb.py
+++ b/docs/source/examples/multi_run_plans_select_cb.py
@@ -15,6 +15,7 @@ RE = RunEngine({})
 db = Broker.named("temp")
 RE.subscribe(db.insert)
 
+
 def factory(name, doc):
     # Runs may be subscribed to different sets of callbacks. Metadata from start
     #   document may be used to identify, which run is currently being started.
@@ -28,19 +29,24 @@ def factory(name, doc):
         cb_list.append(LiveTable([hw.motor1, hw.motor2, hw.det2]))
     return cb_list, []
 
+
 rr = RunRouter([factory])
 RE.subscribe(rr)
 
+
 @bpp.set_run_key_decorator("run_2")
 @bpp.run_decorator(md={"run_key": "run_2"})
 def sim_plan_inner(npts):
+    yield from bps.declare_stream(hw.motor1, hw.motor2, hw.det2, name='primary')
     for j in range(npts):
         yield from bps.mov(hw.motor1, j * 0.1 + 1, hw.motor2, j * 0.2 - 2)
         yield from bps.trigger_and_read([hw.motor1, hw.motor2, hw.det2])
 
+
 @bpp.set_run_key_decorator("run_1")
 @bpp.run_decorator(md={"run_key": "run_1"})
 def sim_plan_outer(npts):
+    yield from bps.declare_stream(hw.motor1, hw.det1, name='primary')
     for j in range(int(npts/2)):
         yield from bps.mov(hw.motor1, j)
         yield from bps.trigger_and_read([hw.motor1, hw.det1])
diff --git a/docs/source/examples/multi_run_plans_sequential.py b/docs/source/examples/multi_run_plans_sequential.py
index 45205a19..1afa6ca9 100644
--- a/docs/source/examples/multi_run_plans_sequential.py
+++ b/docs/source/examples/multi_run_plans_sequential.py
@@ -16,6 +16,7 @@ RE.subscribe(db.insert)
 bec = BestEffortCallback()
 RE.subscribe(bec)
 
+
 def plan_sequential_runs(npts):
     # Single-run plans may be called consecutively. No special handling is required
     #   as long as the previous scan is closed before the next one is opened
diff --git a/docs/source/tutorial.rst b/docs/source/tutorial.rst
index b882f072..2479dd2b 100644
--- a/docs/source/tutorial.rst
+++ b/docs/source/tutorial.rst
@@ -1438,10 +1438,12 @@ Capture Data
     import bluesky.plan_stubs as bps
     def one_run_one_event(detectors):
         yield from bps.open_run()
+        yield from bps.declare_stream(*detectors, name='primary')
         yield from bps.trigger_and_read(detectors)
         yield from bps.close_run()
     def one_run_multi_events(detectors, num):
         yield from bps.open_run()
+        yield from bps.declare_stream(*detectors, name='primary')
         for i in range(num):
             yield from bps.trigger_and_read(detectors)
         yield from bps.close_run()
@@ -1460,6 +1462,7 @@ into *Events* (i.e. rows in a table) and grouping those Events into *Runs*
     def one_run_one_event(detectors):
         # Declare the beginning of a new run.
         yield from bps.open_run()
+        yield from bps.declare_stream(*detectors, name='primary')
 
         # Trigger each detector and wait for triggering to complete.
         # Then read the detectors and bundle these readings into an Event
@@ -1488,6 +1491,7 @@ moved inside a for loop.
 
     def one_run_multi_events(detectors, num):
         yield from bps.open_run()
+        yield from bps.declare_stream(*detectors, name='primary')
 
         for i in range(num):
             yield from bps.trigger_and_read(detectors)
@@ -1511,6 +1515,8 @@ Finally, add another loop re-using ``one_run_multi_events`` inside that loop.
 .. code-block:: python
 
     def multi_runs_multi_events(detectors, num, num_runs):
+        yield from bps.declare_stream(*detectors, name='primary')
+
         for i in range(num_runs):
             yield from one_run_multi_events(detectors, num)
 
@@ -1550,6 +1556,7 @@ Revising our simplest example above, ``one_run_one_event``,
 
     def one_run_one_event(detectors):
         yield from bps.open_run()
+        yield from bps.declare_stream(*detectors, name='primary')
         yield from bps.trigger_and_read(detectors)
         yield from bps.close_run()
 
@@ -1564,6 +1571,7 @@ we incorporate staging like so:
             yield from bps.stage(det)
 
         yield from bps.open_run()
+        yield from bps.declare_stream(*detectors, name='primary')
         yield from bps.trigger_and_read(detectors)
         yield from bps.close_run()
 
@@ -1585,6 +1593,7 @@ equivalent:
         @bpp.stage_decorator(detectors)
         def inner():
             yield from bps.open_run()
+            yield from bps.declare_stream(*detectors, name='primary')
             yield from bps.trigger_and_read(detectors)
             yield from bps.close_run()
 
@@ -1608,6 +1617,7 @@ The result:
         @bpp.stage_decorator(detectors)
         @bpp.run_decorator()
         def inner():
+            yield from bps.declare_stream(*detectors, name='primary')
             yield from bps.trigger_and_read(detectors)
 
         return (yield from inner())
@@ -1646,6 +1656,7 @@ in the :ref:`the earlier section on metadata <tutorial_metadata>`.
         @bpp.stage_decorator(detectors)
         @bpp.run_decorator(md)
         def inner():
+            yield from bps.declare_stream(*detectors, name='primary')
             yield from bps.trigger_and_read(detectors)
 
         return (yield from inner())
@@ -1678,6 +1689,7 @@ purpose, implemented like so:
         @bpp.stage_decorator(detectors)
         @bpp.run_decorator(_md)
         def inner():
+            yield from bps.declare_stream(*detectors, name='primary')
             yield from bps.trigger_and_read(detectors)
 
         return (yield from inner())
@@ -1805,6 +1817,7 @@ can use it to make an on-the-fly decision about whether to continue or stop.
         @bpp.run_decorator()
         def inner():
             i = 0
+            yield from bps.declare_stream(det, name='primary')
             while True:
                 yield from bps.mv(motor, i)
                 readings = yield from bps.trigger_and_read([det])
@@ -1823,6 +1836,7 @@ can use it to make an on-the-fly decision about whether to continue or stop.
     def conditional_break(threshold):
         def inner():
             i = 0
+            yield from bps.declare_stream(det, name='primary')
             while True:
                 yield from bps.mv(motor, i)
                 readings = yield from bps.trigger_and_read([det])


Test patch:
diff --git a/bluesky/tests/test_bec.py b/bluesky/tests/test_bec.py
index e45f52de..49a818a1 100644
--- a/bluesky/tests/test_bec.py
+++ b/bluesky/tests/test_bec.py
@@ -85,7 +85,8 @@ def test_underhinted_plan(RE, hw):
 
     @bpp.run_decorator()
     def broken_plan(dets):
-        yield from bps.trigger_and_read(dets)
+        yield from bps.declare_stream(*dets, name='primary')
+        yield from bps.trigger_and_read(dets, name='primary')
 
     RE(broken_plan([hw.det]))
 
@@ -154,6 +155,8 @@ def test_multirun_nested_plan(capsys, caplog, RE, hw):
     @bpp.stage_decorator([hw.det1, hw.motor])
     @bpp.run_decorator(md={})
     def plan_outer():
+        yield from bps.declare_stream(hw.det1, name='primary')
+
         yield from sequence()
         # Call inner plan from within the plan
         yield from plan_inner()
diff --git a/bluesky/tests/test_callbacks.py b/bluesky/tests/test_callbacks.py
index 3aeaec02..feb81d8b 100644
--- a/bluesky/tests/test_callbacks.py
+++ b/bluesky/tests/test_callbacks.py
@@ -31,6 +31,7 @@ import time
 # copied from examples.py to avoid import
 def stepscan(det, motor):
     yield Msg('open_run')
+    yield Msg('declare_stream', None, det, motor, name='primary')
     for i in range(-5, 5):
         yield Msg('create', name='primary')
         yield Msg('set', motor, i)
diff --git a/bluesky/tests/test_devices.py b/bluesky/tests/test_devices.py
index 116b3cc4..27be9d5e 100644
--- a/bluesky/tests/test_devices.py
+++ b/bluesky/tests/test_devices.py
@@ -4,7 +4,7 @@ from typing import Dict, List
 from bluesky.protocols import Callback, Descriptor, Reading
 
 from bluesky.utils import ancestry, share_ancestor, separate_devices
-from bluesky.plan_stubs import trigger_and_read
+from bluesky.plan_stubs import trigger_and_read, declare_stream
 from bluesky.preprocessors import run_decorator
 from bluesky import Msg, RunEngineInterrupted
 import pytest
@@ -172,6 +172,7 @@ def test_overlapping_raise(RE, det1, det2):
 
     @run_decorator()
     def test_plan(det1, det2):
+        yield from declare_stream(det1, name='primary')
         yield from trigger_and_read([det1])
         yield from trigger_and_read([det2])
 
@@ -203,7 +204,9 @@ def _make_overlapping_tests_2stream(func):
 def test_keyoverlap_2stream(RE, det1, det2):
     @run_decorator()
     def test_plan(det1, det2):
+        yield from declare_stream(det1, name='primary')
         yield from trigger_and_read([det1])
+        yield from declare_stream(det2, name='other')
         yield from trigger_and_read([det2], name='other')
 
     d = DocCollector()
@@ -241,6 +244,7 @@ def _make_overlapping_tests_stream(func):
 def test_overlapped_but_identical(RE, det1, det_list):
     @run_decorator()
     def test_plan(det1, det_list):
+        yield from declare_stream(det1, name='primary')
         yield from trigger_and_read([det1])
         for p in itertools.permutations(det_list):
             yield from trigger_and_read(det_list)
@@ -262,16 +266,31 @@ def test_read_clash(RE):
     dcm2 = DCM('', name='dcm')
 
     with pytest.raises(ValueError):
-        RE(([Msg('open_run')] +
-            list(trigger_and_read([dcm, dcm2.th])) +
-            [Msg('close_run')]))
+        RE(
+            [
+                Msg('open_run'),
+                Msg('declare_stream', None, dcm, dcm2.th, name='primary'),
+                *trigger_and_read([dcm, dcm2.th]),
+                Msg('close_run')
+            ]
+        )
 
     with pytest.raises(ValueError):
-        RE(([Msg('open_run')] +
-            list(trigger_and_read([dcm, dcm2])) +
-            [Msg('close_run')]))
+        RE(
+            [
+                Msg('open_run'),
+                Msg('declare_stream', None, dcm, dcm2, name='primary'),
+                *trigger_and_read([dcm, dcm2]),
+                Msg('close_run')
+            ]
+        )
 
     with pytest.raises(ValueError):
-        RE(([Msg('open_run')] +
-            list(trigger_and_read([dcm.th, dcm2.th])) +
-            [Msg('close_run')]))
+        RE(
+            [
+                Msg('open_run'),
+                Msg('declare_stream', None, dcm.th, dcm2.th, name='primary'),
+                *trigger_and_read([dcm.th, dcm2.th]),
+                Msg('close_run')
+            ]
+        )
diff --git a/bluesky/tests/test_examples.py b/bluesky/tests/test_examples.py
index 56297c2f..929b96b4 100644
--- a/bluesky/tests/test_examples.py
+++ b/bluesky/tests/test_examples.py
@@ -19,7 +19,7 @@ with pytest.warns(UserWarning):
                                   )
 
 
-def test_msgs(RE, hw):
+def test_msgs(hw):
     m = Msg('set', hw.motor, {'motor': 5})
     assert m.command == 'set'
     assert m.obj is hw.motor
@@ -275,6 +275,7 @@ def test_suspend(RE, hw):
         Msg('sleep', None, .2),
         Msg('set', hw.motor, 5),
         Msg('trigger', hw.det),
+        Msg('declare_stream', None, hw.motor, hw.det, name='primary'),
         Msg('create', name='primary'),
         Msg('read', hw.motor),
         Msg('read', hw.det),
@@ -416,6 +417,7 @@ def test_seqnum_nonrepeated(RE, hw):
 
     def gen():
         yield Msg('open_run')
+        yield Msg('declare_stream', None, hw.motor, name='primary')
         yield Msg('create', name='primary')
         yield Msg('set', hw.motor, 1)
         yield Msg('read', hw.motor)
@@ -452,6 +454,7 @@ def test_duplicate_keys(RE, hw):
 
     def gen():
         yield (Msg('open_run'))
+        yield Msg('declare_stream', None, hw.det, hw.identical_det, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('trigger', hw.det))
         yield (Msg('trigger', hw.identical_det))
@@ -467,6 +470,7 @@ def test_illegal_sequences(RE, hw):
     def gen1():
         # two 'create' msgs in a row
         yield (Msg('open_run'))
+        yield Msg('declare_stream', None, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('create', name='primary'))
         yield (Msg('close_run'))
@@ -477,6 +481,7 @@ def test_illegal_sequences(RE, hw):
     def gen2():
         # two 'save' msgs in a row
         yield (Msg('open_run'))
+        yield Msg('declare_stream', None, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('save'))
         yield (Msg('save'))
@@ -488,6 +493,7 @@ def test_illegal_sequences(RE, hw):
     def gen3():
         # 'configure' after 'create', before 'save'
         yield (Msg('open_run'))
+        yield Msg('declare_stream', None, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('configure', hw.motor, {}))
 
@@ -497,6 +503,7 @@ def test_illegal_sequences(RE, hw):
     def gen4():
         # two 'drop' msgs in a row
         yield (Msg('open_run'))
+        yield Msg('declare_stream', None, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('drop'))
         yield (Msg('drop'))
@@ -515,6 +522,7 @@ def test_new_ev_desc(RE, hw):
     def gen1():
         # configure between two events -> two descs
         yield (Msg('open_run'))
+        yield Msg('declare_stream', None, hw.motor, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('read', hw.motor))
         yield (Msg('save'))
@@ -533,6 +541,7 @@ def test_new_ev_desc(RE, hw):
         # -> two descs
         yield (Msg('open_run'))
         yield (Msg('configure', hw.motor, {}))
+        yield Msg('declare_stream', None, hw.motor, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('read', hw.motor))
         yield (Msg('save'))
@@ -550,6 +559,7 @@ def test_new_ev_desc(RE, hw):
         # configure once before any events -> one desc
         yield (Msg('open_run'))
         yield (Msg('configure', hw.motor, {}))
+        yield Msg('declare_stream', None, hw.motor, name='primary')
         yield (Msg('create', name='primary'))
         yield (Msg('read', hw.motor))
         yield (Msg('save'))
diff --git a/bluesky/tests/test_multi_runs.py b/bluesky/tests/test_multi_runs.py
index 73ae42d7..9ba1a31a 100644
--- a/bluesky/tests/test_multi_runs.py
+++ b/bluesky/tests/test_multi_runs.py
@@ -14,7 +14,7 @@ def test_multirun_smoke(RE, hw):
         run_names = ["run_one", "run_two", "run_three"]
         for rid in run_names:
             yield from srkw(bps.open_run(md={rid: rid}), run=rid)
-
+            yield from srkw(bps.declare_stream(*to_read, name='primary'), run=rid)
         for j in range(5):
             for i, rid in enumerate(run_names):
                 yield from bps.mov(motor, j + 0.1 * i)
@@ -49,17 +49,20 @@ def test_multirun_smoke_nested(RE, hw):
     @bsp.set_run_key_decorator("run_one")
     @bsp.run_decorator(md={})
     def plan_inner():
+        yield from bps.declare_stream(*to_read, name='primary')
         yield from some_plan()
 
     @bsp.set_run_key_decorator("run_two")
     @bsp.run_decorator(md={})
     def plan_middle():
+        yield from bps.declare_stream(*to_read, name='primary')
         yield from some_plan()
         yield from plan_inner()
 
     @bsp.set_run_key_decorator(run="run_three")  # Try kwarg
     @bsp.run_decorator(md={})
     def plan_outer():
+        yield from bps.declare_stream(*to_read, name='primary')
         yield from some_plan()
         yield from plan_middle()
 
diff --git a/bluesky/tests/test_new_examples.py b/bluesky/tests/test_new_examples.py
index e6959de9..efbdc8c3 100644
--- a/bluesky/tests/test_new_examples.py
+++ b/bluesky/tests/test_new_examples.py
@@ -43,6 +43,7 @@ from bluesky.plan_stubs import (
     repeater,
     caching_repeater,
     repeat,
+    one_shot,
     one_1d_step,
     one_nd_step)
 from bluesky.preprocessors import (
@@ -66,7 +67,7 @@ from bluesky.plans import count, scan, rel_scan, inner_product_scan
 import bluesky.plans as bp
 from bluesky.protocols import Descriptor, Locatable, Location, Readable, Reading, Status
 
-from bluesky.utils import all_safe_rewind
+from bluesky.utils import all_safe_rewind, IllegalMessageSequence
 
 import threading
 
@@ -326,7 +327,9 @@ def test_descriptor_layout_from_monitor(RE, hw):
 
     descriptor, = collector
     assert descriptor['object_keys'] == {det.name: list(det.describe().keys())}
-    assert descriptor['data_keys'] == det.describe()
+    assert descriptor['data_keys'] == {
+        k: {**v, 'object_name': det.name} for k, v in det.describe().items()
+    }
     conf = descriptor['configuration'][det.name]
     assert conf['data_keys'] == det.describe_configuration()
     vals = {key: val['value'] for key, val in det.read_configuration().items()}
@@ -900,3 +903,29 @@ def test_old_style_wait(RE):
 
     with pytest.raises(RuntimeError):
         RE(unstage_plan())
+
+
+def test_custom_stream_name(RE, hw):
+
+    def new_trigger_and_read(devices):
+        return (yield from trigger_and_read(devices, name='secondary'))
+
+    def new_per_step(detectors, motor, step):
+        return (yield from one_1d_step(detectors, motor,
+                                       step, take_reading=new_trigger_and_read))
+
+    def new_per_shot(detectors):
+        return (yield from one_shot(detectors, take_reading=new_trigger_and_read))
+
+    RE(scan([hw.det], hw.motor, -1, 1, 3, per_step=new_per_step))
+    RE(count([hw.det], 3, per_shot=new_per_shot))
+
+    RE._require_stream_declaration = True
+    with pytest.raises(IllegalMessageSequence):
+        RE(scan([hw.det], hw.motor, -1, 1, 3, per_step=new_per_step))
+
+    with pytest.raises(IllegalMessageSequence):
+        RE(count([hw.det], 3, per_shot=new_per_shot))
+
+    with pytest.raises(IllegalMessageSequence):
+        RE(count([hw.det], 3, per_shot=one_shot))
diff --git a/bluesky/tests/test_plans.py b/bluesky/tests/test_plans.py
index c67ca6f4..a969c0ea 100644
--- a/bluesky/tests/test_plans.py
+++ b/bluesky/tests/test_plans.py
@@ -204,12 +204,12 @@ def test_reset_wrapper(hw, RE):
     RE(bp.relative_inner_product_scan([], 1,
                                       p.pseudo1, 0, 1,
                                       p.pseudo2, 0, 1))
-    expecte_objs = [p, None, None,
+    expecte_objs = [p, None, None, None,
                     p, None, p,
                     None, None, p,
                     None, None, p,
                     p, None]
-    assert len(m_col.msgs) == 14
+    assert len(m_col.msgs) == 15
     assert [m.obj for m in m_col.msgs] == expecte_objs
 
 
diff --git a/bluesky/tests/test_ramp_plan.py b/bluesky/tests/test_ramp_plan.py
index e033806e..5508f809 100644
--- a/bluesky/tests/test_ramp_plan.py
+++ b/bluesky/tests/test_ramp_plan.py
@@ -1,7 +1,7 @@
 from bluesky.tests import requires_ophyd
 from bluesky.tests.utils import DocCollector
 from bluesky.plans import ramp_plan
-from bluesky.plan_stubs import trigger_and_read
+from bluesky.plan_stubs import trigger_and_read, declare_stream
 from bluesky import Msg
 from bluesky.utils import RampFail
 import numpy as np
@@ -14,6 +14,8 @@ def test_ramp(RE):
     from ophyd.positioner import SoftPositioner
     from ophyd import StatusBase
     from ophyd.sim import SynGauss
+    import bluesky.utils as bsu
+    RE.msg_hook = bsu.ts_msg_hook
 
     tt = SoftPositioner(name='mot')
     tt.set(0)
@@ -26,9 +28,17 @@ def test_ramp(RE):
         for j, v in enumerate(np.linspace(-5, 5, 10)):
             RE.loop.call_later(.1 * j, lambda v=v: tt.set(v))
         RE.loop.call_later(1.2, st._finished)
+        print("YOLO")
         return st
 
+    first = True
+
     def inner_plan():
+        nonlocal first
+        if first:
+            yield from declare_stream(dd, name='primary')
+            first = False
+
         yield from trigger_and_read([dd])
 
     g = ramp_plan(kickoff(), tt, inner_plan, period=0.08)
@@ -73,7 +83,14 @@ def test_timeout(RE):
 
         return StatusBase()
 
+    first = True
+
     def inner_plan():
+        nonlocal first
+        if first:
+            yield from declare_stream(det, name='primary')
+            first = False
+
         yield from trigger_and_read([det])
 
     g = ramp_plan(kickoff(), mot, inner_plan, period=.01, timeout=.1)
diff --git a/bluesky/tests/test_run_engine.py b/bluesky/tests/test_run_engine.py
index 22f6bc1f..41bdd677 100644
--- a/bluesky/tests/test_run_engine.py
+++ b/bluesky/tests/test_run_engine.py
@@ -20,7 +20,7 @@ from bluesky import Msg
 from functools import partial
 from bluesky.tests.utils import MsgCollector, DocCollector
 from bluesky.plans import (count, grid_scan)
-from bluesky.plan_stubs import (abs_set, trigger_and_read, checkpoint)
+from bluesky.plan_stubs import (abs_set, trigger_and_read, checkpoint, declare_stream)
 from bluesky.preprocessors import (finalize_wrapper, run_decorator,
                                    reset_positions_decorator,
                                    run_wrapper, rewindable_wrapper,
@@ -97,9 +97,15 @@ def test_verbose(RE, hw):
     RE.verbose = True
     assert RE.verbose
     # Emit all four kinds of document, exercising the logging.
-    RE([Msg('open_run'), Msg('create', name='primary'), Msg('read', hw.det),
-        Msg('save'),
-        Msg('close_run')])
+    RE(
+        [
+            Msg('open_run'),
+            Msg('declare_stream', None, hw.det, name='primary'),
+            Msg('create', name='primary'), Msg('read', hw.det),
+            Msg('save'),
+            Msg('close_run')
+        ]
+    )
 
 
 def test_reset(RE):
@@ -310,14 +316,27 @@ def test_empty_bundle(RE, hw):
 
     # In this case, an Event should be emitted.
     mutable.clear()
-    RE([Msg('open_run'), Msg('create', name='primary'), Msg('read', hw.det), Msg('save')],
+    RE([
+        Msg('open_run'),
+        Msg('declare_stream', None, hw.det, name='primary'),
+        Msg('create', name='primary'),
+        Msg('read', hw.det),
+        Msg('save')],
        {'event': cb})
     assert 'flag' in mutable
 
     # In this case, an Event should not be emitted because the bundle is
     # emtpy (i.e., there are no readings.)
     mutable.clear()
-    RE([Msg('open_run'), Msg('create', name='primary'), Msg('save')], {'event': cb})
+    RE(
+        [
+            Msg('open_run'),
+            Msg('declare_stream', None, name='primary'),
+            Msg('create', name='primary'),
+            Msg('save')
+        ],
+        {'event': cb}
+    )
     assert 'flag' not in mutable
 
 
@@ -1107,6 +1126,7 @@ def test_rewindable_state_retrival(RE, start_state):
 
 
 @pytest.mark.parametrize('start_state,msg_seq', ((True, ['open_run',
+                                                         'declare_stream',
                                                          'rewindable',
                                                          'rewindable',
                                                          'trigger',
@@ -1119,6 +1139,7 @@ def test_rewindable_state_retrival(RE, start_state):
                                                          'rewindable',
                                                          'close_run']),
                                                  (False, ['open_run',
+                                                          'declare_stream',
                                                           'rewindable',
                                                           'trigger',
                                                           'trigger',
@@ -1139,7 +1160,11 @@ def test_nonrewindable_detector(RE, hw, start_state, msg_seq):
     m_col = MsgCollector()
     RE.msg_hook = m_col
 
-    RE(run_wrapper(trigger_and_read([hw.motor, hw.det])))
+    def test():
+        yield from declare_stream(hw.motor, hw.det, name='primary')
+        return (yield from trigger_and_read([hw.motor, hw.det]))
+
+    RE(run_wrapper(test()))
 
     assert [m.command for m in m_col.msgs] == msg_seq
 
@@ -1265,6 +1290,7 @@ def test_bad_from_idle_transitions(RE, change_func):
 def test_empty_cache_pause(RE):
     RE.rewindable = False
     pln = [Msg('open_run'),
+           Msg('declare_stream', None, name='primary'),
            Msg('create', name='primary'),
            Msg('pause'),
            Msg('save'),
@@ -1642,6 +1668,7 @@ def test_drop(RE, hw):
     # Drop first, drop after saving, save after dropping
     def plan():
         yield Msg('open_run')
+        yield Msg('declare_stream', None, det, name='primary')
         yield from inner('drop')
         yield from inner('save')
         yield from inner('drop')
@@ -1674,10 +1701,12 @@ def test_failing_describe_callback(RE, hw):
     def plan():
         yield Msg('open_run')
         try:
+            yield Msg('declare_stream', None, det,  name='primary')
             yield Msg('create', name='primary')
             yield Msg('read', det)
             yield Msg('save')
         finally:
+            yield Msg('declare_stream', None, det2,  name='primary')
             yield Msg('create', name='primary')
             yield Msg('read', det2)
             yield Msg('save')
diff --git a/bluesky/tests/test_scientific.py b/bluesky/tests/test_scientific.py
index c36f95d1..c69acedd 100644
--- a/bluesky/tests/test_scientific.py
+++ b/bluesky/tests/test_scientific.py
@@ -3,12 +3,14 @@ import numpy as np
 from bluesky.plans import scan
 from ophyd.sim import motor, det, SynGauss
 from bluesky.callbacks.fitting import PeakStats
-from scipy.special import erf
 
 
 def get_ps(x, y, shift=0.5):
     """ peak status calculation from CHX algorithm.
     """
+    pytest.importorskip('scipy')
+    from scipy.special import erf
+
     lmfit = pytest.importorskip('lmfit')
     ps = {}
     x = np.array(x)
diff --git a/bluesky/tests/test_supplemental_data.py b/bluesky/tests/test_supplemental_data.py
index 58072c1b..0093e50c 100644
--- a/bluesky/tests/test_supplemental_data.py
+++ b/bluesky/tests/test_supplemental_data.py
@@ -59,18 +59,18 @@ def test_baseline(hw):
     original = list(count([hw.det]))
     processed = list(D(count([hw.det])))
     # should add 2X (trigger, wait, create, read, save)
-    assert len(processed) == 10 + len(original)
+    assert len(processed) == 11 + len(original)
 
     # two baseline detectors
     D.baseline.append(hw.det3)
     processed = list(D(count([hw.det])))
     # should add 2X (trigger, triger, wait, create, read, read, save)
-    assert len(processed) == 14 + len(original)
+    assert len(processed) == 15 + len(original)
 
     # two baseline detectors applied to a plan with two consecutive runs
     original = list(list(count([hw.det])) + list(count([hw.det])))
     processed = list(D(pchain(count([hw.det]), count([hw.det]))))
-    assert len(processed) == 28 + len(original)
+    assert len(processed) == 30 + len(original)
 
 
 def test_mixture(hw):
@@ -80,7 +80,7 @@ def test_mixture(hw):
 
     original = list(count([hw.det]))
     processed = list(D(count([hw.det])))
-    assert len(processed) == 2 + 5 + 10 + len(original)
+    assert len(processed) == 3 + 5 + 10 + len(original)
 
 
 def test_order(hw):
@@ -96,6 +96,7 @@ def test_order(hw):
     actual = [msg.command for msg in D(null_run())]
     expected = ['open_run',
                 # baseline
+                'declare_stream',
                 'trigger',
                 'wait',
                 'create',