Add StreamingActiveLearner: learner loop driven by streamed data

[andre-merzky]

Summary

Digital-twin architectures need a learner that is retriggered by streamed input (sensor windows, events) rather than one that drives itself with an internal iteration count and its own simulation task. The existing learners run once to convergence — embedded in a DT runtime, data arriving after startup never influences learning.

StreamingActiveLearner makes the loop data-driven:

• Ingestion: items arrive via await learner.feed(item) or attached async-iterator sources (single or list, also via the sources ctor arg; pump tasks are deferred to start(), so construction needs no running event loop).
• Windowing: batch_size items form a window; max_wait flushes a partial window after a quiet period; conflate=True drops backlog and keeps the newest items when iterations are slower than the stream (latest-wins).
• Iteration: each window triggers training → active-learning → criterion; the window is passed as first argument to the training task, with the previous active-learn result as warm-start dependency.
• Publish gate: a met stop criterion does not end the loop — state.should_stop marks the model as publishable and on_model_ready callbacks fire (mapping 1:1 onto a DT runtime's publish_new_model), while consumption continues until stop() or source exhaustion.
• IterationState streaming, LearnerConfig/set_next_config, state registry, and trackers all work as in the existing learners.

A DT-side integration demo (sensor → ZMQ broker → investigator feeding this learner, republishing on criterion-met) lives in the digitaltwin repo under test/rose_streaming/.

Verification

• 7 unit tests: windowing/exhaustion, publish-gate semantics, stop() on an empty queue, max_wait partial flush, conflation, ctor-sources without an event loop, missing-task validation. Full unit suite: 126 passed.
• Benchmarked throughput over batch size, in-process feed() vs ZMQ pub/sub through the DT broker, with stubbed tasks (machinery only) and real no-op tasks through asyncflow + process pool (plots below):

batch size	machinery, in-proc	machinery, ZMQ	e2e, in-proc	e2e, ZMQ
1	88k msg/s	13k msg/s	285 msg/s	299 msg/s
10	224k msg/s	15k msg/s	2.8k msg/s	1.9k msg/s
100	349k msg/s	15k msg/s	30.4k msg/s	5.8k msg/s

The end-to-end numbers use radical-cybertools/radical.asyncflow#91 (event-driven scheduler); on asyncflow 0.5.0 the per-window floor is ~22ms (10ms poll interval × 2 chained tasks), capping batch=1 at ~45 msg/s.

Plots

With event-driven asyncflow scheduler (radical-cybertools/radical.asyncflow#91):

Baseline on released asyncflow 0.5.0:

🤖 Generated with Claude Code

[Copilot]

Pull request overview

Adds a new StreamingActiveLearner to support data-driven (stream-ingested) active learning loops suitable for digital-twin runtimes, where learning iterations are triggered by arriving data windows rather than internal iteration scheduling.

Changes:

• Introduces StreamingActiveLearner with async ingestion (feed / async-iterator sources), windowing (batch_size, max_wait), optional conflation, and “publish gate” criterion semantics via on_model_ready.
• Adds unit tests covering windowing, source exhaustion, stop() behavior, max_wait partial flush, conflation, ctor sources behavior, and missing-task validation.
• Exposes StreamingActiveLearner from rose.al via __init__.py.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
rose/al/streaming_learner.py	Implements the streamed, window-triggered active learner loop and publish-gate callbacks.
tests/unit/test_streaming_learner.py	Adds unit coverage for streaming windowing, stop/exhaustion, publish-gate semantics, and conflation behavior.
rose/al/__init__.py	Exports StreamingActiveLearner as part of the public rose.al API.

Suppressed comments (1)

rose/al/streaming_learner.py:257

• Pump tasks are cancelled in the finally block, but they aren’t awaited. This can leave pending tasks running briefly after start() exits and can produce "Task was destroyed but it is pending!" warnings when the event loop closes. Consider awaiting the cancelled tasks with asyncio.gather(..., return_exceptions=True) and clearing _sources to ensure a clean shutdown.

                task.cancel()

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[andre-merzky]

Fixed in 64330d9 — batch_size and max_wait are validated in init and fail fast with ValueError; covered by test_invalid_params_raise.

[andre-merzky]

Fixed in 64330d9 — conflation now happens at ingestion time: feed() and the source pumps drop the oldest backlog before enqueuing, so the queue stays bounded to ~batch_size items during slow iterations (sentinels are processed, not lost). Covered by test_conflate_bounds_queue_at_ingestion.

[BenCarter44]

Nice! Looks good. I wrote a small example located in examples/active_learn/streaming, and it is working nicely. I like simplicity of the streaming learner.

I have three comments, one small thing missing and two design questions:

1. It doesn't look like simulations are triggered in this. Shouldn't simulations be part of the loop (as active learning's role is to request more sims)

2. Right now, the StreamingActiveLearner requires the criterion function in order to run on_model_ready(). This is ok, but more as a design decision: should the StreamingLearner simply be SequentialActiveLearner + streams (what it essentially is right now), or should it support any generic Learner and then add streams? I do think it is fine leaving it the way it is; just more took notice of it. (A generic learner would pose its own complexities)

3. Right now, the on_model_ready() callback is only able to capture state of the training if the training tasks are function_tasks and emit dictionaries. Design decision: if a user is using executable tasks, should they be required to build out their own implementation for model reference retrieval?

[BenCarter44]

Should it also trigger the simulation function too? (as the active learner selects new data that it wants to rerun sims on)

[BenCarter44]

If a criterion function is not set, the model update callback won't be triggered. More from the design perspective: should we require a StreamingLearner to always have a criterion function set?

(Right now its closely matching SequentialActiveLearner's behavior, which is ok, but I wonder if the StreamingLearner should be more broad and support custom pipelines).

[BenCarter44]

I see that it stores the state in the result from the acl and train. That is good. However, this won't work for executable tasks (and therefore the on_model_ready callback won't get any metadata)

Or: it assumes that if the user is using executable tasks, they add their own logic to pass a reference from the trainer to the callback.

[BenCarter44]

Also, I'm unsure what you mean / where you are referring to in the original post:

A DT-side integration demo (sensor → ZMQ broker → investigator feeding this learner, republishing on criterion-met) lives in the digitaltwin repo under test/rose_streaming/.

[AymenFJA]

@andre-merzky @BenCarter44 thanks for the great work, a few things to mention:

Conceptually, I agree with the streaming behavior, but I would not introduce StreamingActiveLearner as a separate learner hierarchy.

We should keep the existing ROSE learners and make streaming a general behavior/capability that can be applied to them:

SequentialActiveLearner
        +
   Streaming behavior
        ↓
Streaming Sequential Active Learning

and:

ParallelActiveLearner
        +
   Streaming behavior
        ↓
Streaming Parallel Active Learning

This gives us the DT-driven, feed()/source/window/model-publish behavior described above without forcing us to create separate StreamingActiveLearner, StreamingRefinementLearner, etc. The same streaming mechanism can then be reused by future learner types.

I would like to keep the streaming approach consistent with the ROSE design for now. The current learners are essentially batch learners, but the learner abstraction is extensible. Similarly, streaming should be introduced as a general, reusable behavior/capability rather than as a separate learner hierarchy. This allows SequentialActiveLearner and ParallelActiveLearner to support streaming while preserving the existing ROSE design.

Happy to expand further/help if needed.

[andre-merzky]

Thanks @AymenFJA , that makes sense. I'll rework the PR!