ADR 0002: Gate workflow execution at the JobManager on context-stream readiness#

  • Status: accepted

  • Deciders: Simon

  • Date: 2026-05-21

Context#

Sciline-based workflows parametrise their reduction graph on context values — motor positions, sample geometry, choppers — delivered as their own streams alongside detector data. StreamProcessor holds a hidden None seed for every such context binding. If a workflow runs before its context stream has produced a value, the providers downstream of that input receive None and either crash (bifrost’s process_chunk_workflow reads rotation inside accumulate, raising 'NoneType' object has no attribute 'ndim') or — worse — silently produce detector data attributed to a default geometry. Both failure modes have surfaced in production.

The system needs a gate that prevents workflow invocation until every context input that has no safe default has a value. The design questions are where this gate lives and how it learns which inputs to gate on.

What needs gating, and what does not#

A job consumes several kinds of input beyond its primary data stream. They differ in whether the workflow tolerates the input being absent:

  • Dynamic aux — inputs that accumulate over time inside the workflow (e.g. LOKI’s incident/transmission monitors). They flow through StreamProcessor.accumulate and tolerate late arrival: events processed against an empty accumulator still produce a useful (uncorrected) result the consumer can ignore. Gating these would break accumulation — events arriving before monitors would be dropped, and no result would ever appear even after the monitors do.

  • Context with a safe default — inputs that parametrise the graph but map a missing value to a well-defined result. ROI is the example: the Sciline providers treat “no request” and “empty request” identically as no ROI selected (detector_view/roi.py). There is no failure mode to prevent, so these are not gated either; ROI is routed as an ordinary auxiliary source.

  • Context with no safe default — inputs that parametrise the graph and whose absence is unrepresentable: motion/geometry (bifrost rotation, LOKI carriage). None crashes or yields silently-wrong geometry. Only this category is gated.

The gating axis is therefore “does this input tolerate absence?”, not “set_context vs accumulate” — ROI flows through set_context yet is not gated. The set of gated streams for a job is derived from ContextBinding declarations at job-creation time; the declaration model is settled in ADR 0003.

Decision#

Gate at JobManager, not at the workflow execution unit.

Job carries gating_streams: set[str] — the wire stream names whose values the workflow needs before it can run. Job.missing_context(available) returns the gating streams not present in available. The set is populated at JobFactory.create from the ContextBinding declarations that resolve for the specific (spec, source) (ADR 0003).

A gated job occupies a dedicated stage between scheduled and active, held in its own _pending_context_jobs bucket. When should_start fires, _start_job routes the job by gating_streams: empty → it becomes active at once; non-empty → it enters pending_context with a pending_context_warning. Only active jobs are processed — a pending job is never fed primary data.

The preprocessor’s context cache is the single source of truth for which context streams have a value. JobManager holds a context_reader onto it; each tick, _open_context_gates refills the batch from the cache for every gating stream of a pending job still absent, then checks job.missing_context(available) against the post-refill batch:

  • non-empty → the job stays pending; its pending_context_warning(missing) is refreshed to the still-missing streams.

  • empty → the job graduates to active (_activate_pending_job), dropping its warning; its own processing outcome drives its state from there.

Because gate opening and value delivery read the same dict, the gate-opening batch structurally carries every context value the job needs — a gate cannot open without its values being deliverable to set_context. Context values arrive intermittently — a motor position is published once and then is quiet — but the cache holds every value ever received, so availability is sticky across ticks.

Graduation is one-way: gating_streams is fixed and the cache only grows, so once a job is active it has left the gate permanently and is never re-checked, and a job with no context never enters the stage at all. The per-tick gate therefore scans only the small, transient pending bucket — not every active job — and refills only gating streams of pending jobs, never the cached context of steady-state active jobs (which would re-fire set_context and force eager downstream recompute on unchanged values).

Time-based activation is preserved: should_start(current_time) drives the job out of scheduled on time, independent of stream-arrival timing. The readiness gate governs only the subsequent pending → active transition, so a job blocked on a dead producer stays visible as pending_context rather than silently rewriting its schedule.

StreamProcessorWorkflow remains a pure “given dynamic + context, produce result” unit. It does not consult the gate, does not track which keys have arrived, and does not drop dynamic data.

There is no seeding mechanism. The gate stays closed until the producer publishes. If the producer is down, the workflow does not run — which is the correct behaviour for a context with no safe default.

Alternatives considered#

Option

Notes

Gate at JobManager on context streams with no safe default (chosen)

The gate respects the workflow’s own tolerates-absence distinction. The set is derived from declarations, so adding a gated workflow needs no JobManager change.

Gate at JobManager on all aux

Conflates routing with execution; gates dynamic aux (monitors) that legitimately accumulate without context. Forces tests and production to pre-publish dynamic aux before events. Rejected.

Gate inside StreamProcessorWorkflow with per-workflow opt-in

Couples scheduling to workflow execution; forces StreamProcessorWorkflow to carry a “drop dynamic while pending” branch and spreads gate configuration across factories. Rejected.

Defer job construction until context ready

Conflicts with time-based scheduling and moves the Sciline-pipeline build cost from config-load to first-data-arrival, introducing user-visible latency. Rejected.

Hold gated jobs in scheduled rather than a distinct stage

Entangles the schedule contract with producer liveness — a dead motion producer would keep a job “scheduled” past its start, and the dashboard could not tell “not due yet” from “started but blocked”. A dedicated pending_context stage keeps the two axes separate. Rejected.

Key design choices#

The gate set is derived, not declared per workflow#

Job.gating_streams is populated at JobFactory.create from the ContextBinding records resolved for the (spec, source). Declarations are the single source of truth for which inputs gate; adding a workflow that needs gating requires only the standard declaration (ADR 0003), no JobManager change.

The context cache is the gate’s only readiness source#

Gating compares against the streams present in the batch after refilling it from the preprocessor’s context cache (context_reader), not against the streams that happened to arrive this tick. The cache holds every context value ever received, so a value seen once stays available and the gate can open on a tick where the last missing stream finally arrives, with all earlier values refilled alongside it. Tracking readiness separately (e.g. a per-job set of seen stream names) was rejected: a second sticky store can disagree with the cache, opening a gate whose values are not deliverable — the workflow would then finalize against the hidden None seed, the exact failure this ADR exists to prevent. With a single source, gate opening and value delivery cannot diverge.

A distinct stage, not a per-tick skip#

A gated job is held in its own _pending_context_jobs bucket, not left among the active jobs with a per-tick “skip if gated” branch. Because gating_streams is fixed and the cache only grows, readiness is monotonic: a job graduates once and never re-gates. Modelling that as a one-way stage move means the processing loop iterates only runnable jobs, the per-tick gate scans only the (small) pending bucket, and a job that never needed context never participates in gating at all.

Primary data arriving before context is dropped#

A pending job is not among the active jobs a batch is dispatched to, so primary data arriving while a gating stream is still missing simply is not delivered to it and nothing is buffered. It processes the first primary batch on or after the tick its gate opens. This is conceptually correct: detector events whose geometry context is unknown cannot be meaningfully reduced, and silent attribution to a default geometry is the exact failure mode the gate exists to prevent. Dynamic aux is unaffected — it is not in the gate set and accumulates normally.

Dedicated JobState.pending_context#

A job held in the pending stage is conceptually distinct both from one not yet due (scheduled) and from one whose last processing attempt errored (warning). JobState.pending_context is healthy and waiting on a producer; surfacing it distinctly lets an operator tell a dead motion producer from a job that simply has not started.

Consequences#

  • StreamProcessorWorkflow stays focused on workflow execution and carries no gate state.

  • Any workflow that declares motion or geometry context gets correct startup behaviour automatically — the only requirement is the ContextBinding declaration (ADR 0003). No per-instrument gate opt-in.

  • Dynamic aux (LOKI monitors) flows through unchanged — the gate does not see it.

  • ROI is routed as an auxiliary source, not gated: “no ROI” is a safe default, so there is no readiness to wait on and no seed to inject.

  • Warning emission for context-pending jobs is consolidated inside JobManager. There is no warning channel between the workflow execution unit and the job layer.

  • JobState.pending_context is added to the job state machine, backed by a _pending_context_jobs bucket between _scheduled_jobs and _active_jobs. Job lookup, control, reset, status, and finish handling all account for the new bucket.

  • JobManager takes a context_reader onto the preprocessor’s context cache; the gate’s refill-then-check runs entirely inside JobManager, so gate correctness does not depend on callers enriching the batch.

  • Tests at JobManager level cover: dynamic aux not gated; context (motion) blocks then unblocks once its stream has a value; a value cached on an earlier tick still opens the gate (and is delivered) when the last stream arrives later; primary dropped while gated; gates independent across jobs.

  • The declaration model that feeds gating_streams and context_keys, and the routing that ensures the gated streams are actually subscribed, are settled in ADR 0003.