ADR 0004: Key workflow inputs uniformly by canonical stream name#

  • Status: accepted

  • Deciders: Simon

  • Date: 2026-06-02

Context#

A message’s identifier passes through four naming schemes between Kafka and a Sciline provider:

  • (topic, source_name) — raw Kafka identity.

  • role — the logical input a workflow author and a user reason in for auxiliary data (“the incident monitor used for normalisation”), which may be filled by any of several physical streams.

  • canonical stream_name — the name a stream is given in the instrument stream config once StreamLUT resolves its Kafka identity on ingress. It is a NeXus name: for f144 streams the NeXus path generated from the geometry file; for detector/monitor data the NeXus group name, which the workflow spec calls source_name and which need not equal the Kafka source_name it resolved from.

  • sciline.Key — the graph’s type-addressed input (e.g. NeXusData[Incident, SampleRun]).

The three kinds of workflow input (primary detector data, dynamic auxiliary data, context — ADR 0002/0003) all arrive on the data path and all end up inside the wrapped StreamProcessor, differing only in how their stream binding is chosen and how they are fed into the graph.

Dynamic aux used to be keyed in dynamic_keys by its stable role (e.g. incident_monitor), while the primary input and context inputs were keyed by canonical stream name. That asymmetry leaked the role scheme onto the data path: the role→stream map had to be threaded into wire_dynamic_transforms (to match bindings against canonical dependent_sources) and into the Job’s runtime routing (a _stream_to_fields remap turning incoming stream-named data back into roles before handing it to the workflow). Two layers spoke a scheme that only mattered at construction.

Decision#

Key every dynamic input — primary, aux, context — uniformly by its canonical stream_name on the data path. The role scheme is resolved to a stream name exactly once, at workflow construction, and never enters the running workflow object.

Translation between schemes happens at three points, only two of which are on the data path:

#

Boundary

From → To

Performed by

Path

1

Kafka ingress

(topic, source_name) → stream_name

MessageAdapter.get_stream_id + StreamLUT

data

2

Workflow construction

role → stream_name (building dynamic_keys)

instrument factory, inline aux_source_names[role]

build-time

3

Workflow runtime

stream_name → sciline.Key

StreamProcessorWorkflow.accumulate / set_context (dict lookup)

data

JobFactory.create computes the role→stream map (aux_source_names) from AuxSources.render — instrument defaults overlaid with the user’s selection — and passes it to the factory. The factory keys dynamic_keys by canonical stream name directly, resolving each role inline (aux_source_names['incident_monitor']: NeXusData[Incident, SampleRun]). An unknown role raises KeyError at job creation, visible in factory code. The primary input is the degenerate case: its binding is the job’s identity (source_name), so translation 2 is the identity function and the factory keys it directly without aux_source_names. Context wire names equal their stream names (ADR 0003), so translation 2 does not touch them either — it concerns auxiliary inputs only.

Consequently the Job carries no mapping: input_streams and gating_streams are plain set[str] of stream names, and Job.add forwards {**primary_data, **aux_data} to the workflow unchanged. The only place a stream_name becomes a sciline.Key is the dict lookup inside StreamProcessorWorkflow.accumulate, where which dict the name is found in (dynamic_keys vs context_keys) decides accumulate vs set_context.

Because the routing layer must inject per-job context and f144-driven transforms after the factory has produced the workflow but before the wrapped StreamProcessor is built, StreamProcessorWorkflow defers construction behind SupportsContext.build(context_keys, chain_patch_bindings). The factory passes only its own internal context (e.g. ROI) to the constructor; build merges the routing-layer context_keys, wires chain_patch_bindings, and bakes the graph.

One-stream-to-many-roles multiplexing is dropped. It was unused and is incompatible with canonical-name keying: a single incoming stream_name maps to exactly one dynamic_keys entry.

Alternatives considered#

Option

Notes

Uniform canonical-name keying, role resolved at construction (chosen)

One scheme on the data path; Job is a plain name set; role lives only in factory code where an unknown role fails loudly at job creation.

Keep aux keyed by role; remap stream_name → role in Job

The prior state. Forces the role map into wire_dynamic_transforms and a Job._stream_to_fields remap on every batch; two data-path layers carry a scheme that only matters at build time. Rejected.

Resolve roles at runtime in Job/JobManager

Moves a construction-time concern onto the hot path and re-introduces the remap. Rejected.

Preserve one-stream-to-many-roles multiplexing

Unused, and irreconcilable with a one-to-one stream_name → dynamic_keys mapping. Rejected.

Key design choices#

Only one naming scheme is on the data path#

Translation 2 happens at construction. Once messages flow, the entire stack from WorkflowData.data down to the StreamProcessorWorkflow dict lookup speaks a single language — stream_name. Only the innermost shell (StreamProcessor / Sciline) speaks sciline.Key. This is the invariant that lets Job hold plain name sets and forward data untouched.

Role resolution is a build-time concern, co-located with the factory#

The role scheme is the one a workflow author and a user reason in. Resolving it in the factory that constructs the workflow — the same module that declares the dynamic_keys — keeps role knowledge local and makes a typo a loud KeyError at job creation rather than a silent mis-route.

Two independent sources must agree on the canonical name#

The StreamLUT drives translation 1 on ingress; the render / aux_source_names map drives translation 2 in the factory. Nothing cross-checks them: an aux source whose rendered stream_name never appears on ingress simply never delivers data. This is the price of decoupling routing from construction; it is acceptable because both sides derive from the same instrument stream definitions.

Consequences#

  • Job.input_streams / gating_streams are set[str]; Job.add forwards primary and aux data to the workflow unchanged, with no remap.

  • dynamic_keys is keyed by canonical stream name for every input; the role scheme never enters the workflow object.

  • wire_dynamic_transforms takes only the workflow and the resolved ChainPatchBinding records — no aux_source_names — because it reads component types off dynamic_keys, which is already stream-name-keyed (ADR 0003).

  • StreamProcessorWorkflow defers its StreamProcessor construction behind SupportsContext.build, so the routing layer can inject per-job context and chain-patch bindings before the graph is baked.

  • One-stream-to-many-roles multiplexing is no longer supported.

  • The naming model and its translation points are illustrated in

    doc:

    /developer/design/stream-keying.