DTaaS M2: ROSE ex-situ learning (StreamingLearnerInvestigator, dual-engine twins, endpoint-loss visibility) - #1
Merged
Merged
Conversation
`StreamingLearnerInvestigator` packages what `test/rose_streaming` spells out by hand: a `StreamingActiveLearner` fed from `ON_INPUT`, a bootstrap model published before the first input (inference gates on a published model, and the stream reaches the learner through that same input -- without it the twin deadlocks), `on_model_ready -> publish_new_model`, and hooks for the two things a subclass really shapes: what to bootstrap with and what a criterion-met window publishes. The class is also the marker for dual-engine injection: it takes the ex-situ engine as `learn_flow` and runs its learner tasks there, while inference stays on the twin's own engine. Unset, one engine serves both, so it works locally and against a single-endpoint deployment. Cancelling the consumer task alone is not a clean shutdown -- it abandons ROSE's async generator mid-window along with the source pumps it owns. So `_TwinComponent` grows an internal `_on_stop` hook that `DTRuntime.stop()` calls, under one shared budget, before it cancels anything; the learner's implementation lets the window collector unblock and the loop unwind on its own. `DTRuntime.fail()` is the matching door for failures only the host can see. ROSE is an optional `learn` extra, not part of `service`: the learner runs fine against a local engine with no ORBIT in sight, and a service host serving only in-situ twins should not carry ROSE's dependencies. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Engine configuration was already name-keyed, so `'exsitu'` is a
config-only addition: `{"engines": {"exsitu": {"endpoint_name": ...}}}`.
Unconfigured, it aliases `'task'` -- a single-endpoint deployment keeps
working and no demo has to change. Build tasks and locks are now both
keyed by engine name, so a two-minute remote backend init cannot
serialize ahead of a `'task'` build another twin is waiting on, and a
configured `'exsitu'` is built in `twin_create`'s background phase
alongside it: `add_investigator` stays a short verb.
Injection is by subclass check, as planned -- there is no user-facing
engine selector in v1. A `StreamingLearnerInvestigator` is
instantiated with the ex-situ engine as `learn_flow` on top of the
usual `flow`, and the twin records which engines it actually bound to.
That record is what makes R8 detection precise. `on_topology_change`
maps lost participants onto each session's resolved engine endpoints
and fails exactly the twins that used them, with `engine endpoint
lost: <endpoint>` in `twin_list`; twins on surviving engines are
untouched. Detection only -- the backend does not reconnect and
components bind their engine at construction, so recovery stays the
client's job. What this removes is the silent version: inference
calls on a days-long twin that simply never return.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`test/10-learner` is the M2 service demo: a sensor stream that both feeds a ROSE streaming learner and is served by an inference task, with the two halves on two different endpoints. Synthetic regression rather than MNIST -- the point is the wiring, and a demo should not pull in a deep-learning stack to make it. The client asks the twin for the same reading every few seconds and the answer walks from 0.0 (the uncalibrated bootstrap model) up to the true calibration. Nothing about the request changes; only the model behind it does. The README states the dual-engine wiring rather than leaving it to a default, and explains the two things that follow from a remote endpoint: learner tasks registered `as_executable=False` so they travel as cloudpickled function tasks, and a criterion that carries the model it scores by value instead of reading the `model.json` a training task would have left on a filesystem the endpoint does not share. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Unit level, against a local thread-backed engine (ROSE typechecks its `WorkflowEngine`, so there is no useful fake): the bootstrap model that keeps the first input from deadlocking, a published model changing the next prediction, dual-engine construction, a stop that winds the learner down and leaves no task, and a twin stopped before its learner ever ran not sitting out the stop hook's timeout. Plus the service side without a broker: the `'exsitu'` alias, a slow build not holding up a sibling engine, injection by subclass check, and `endpoints_lost` failing exactly the twins that bound the lost engine. Integration level, against a live stack with two *distinct* rhapsody endpoints: the M2 acceptance test asserts that a model learned ex-situ is what the next in-situ prediction answers with, and -- since every endpoint now stamps its name into the environment and both tasks report it -- that the two halves really did run on different endpoints. The inference task returns a dict, so it also covers rich return values round-tripping through ORBIT. For R8, its own disposable endpoint: killing it fails the learner twin with `engine endpoint lost: <endpoint>` while its task-only sibling in the same session keeps serving. Endpoint startup is now one context manager instead of three copies, and `dt_client` is a factory so a test can pick its engine configuration; `dt` is the single-engine case of it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Ex-situ learning, the dual-engine config, why learner tasks have to be function tasks, and the R8 failure mode a `twin_list` now shows. The M1 note about JSON-safe return values keeps its advice but records that the upstream fix has landed on radical.orbit `devel`. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
F1 was the real one. ORBIT announces a lost participant exactly once, but the consequence does not expire: `endpoints_lost` failed the twins that had bound the dead engine and then left it sitting in `_engines`, so the *next* twin in that session bound it, reached `ready`, and stalled in silence -- reintroducing precisely the failure mode R8 detection exists to remove. The session now remembers its lost endpoints and `engine()` refuses to hand one out again, so a later `twin_create` fails immediately with `engine '<name>' endpoint was lost; recreate the session`. Which is also the correct remediation per the plan, and the wording is now that everywhere it appears (F2): engines are session-shared, so the session is what has to go, not just the twins. Smaller things: - the executable-task warning stayed quiet when both halves run on one engine -- a local learner's shell command with local paths is a perfectly good task (F3); - the demo now omits the `'exsitu'` key rather than configuring it with a `None` endpoint, so a single-endpoint run really does take the documented alias path (F4); - a failed `learn` import is logged: a service built with the extra and a broken ROSE looked exactly like one built without it (F5); - `_on_stop` hooks are gathered concurrently, so two learners at five seconds each no longer eat a ten-second budget end to end (F6). The optional deduction from the cancellation wait is deliberately not taken: a quiesce that spends the budget would leave zero for cancellation and report tasks as ignoring it that were simply never given a tick; - one session's bookkeeping can no longer cost the others their only notification of an endpoint loss (F7); - the criterion-carry pattern is documented with its cost -- the state mirror is re-cloudpickled every window and keeps every key (F8); - the `_on_stop` docstring no longer claims cancellation abandons the generator. It usually does not; what the hook actually buys is a window-boundary exit instead of killing an in-flight training task on a shared engine, a correct tracker stop_reason (ROSE catches `Exception`, not `CancelledError`), and no reliance on asyncgen GC after days of running (F9); - a published model whose keys the inference task cannot take is named as such instead of surfacing as a bare `TypeError` from a call the user never wrote. Only the call is rewritten: an error from inside the task body has its own traceback frame and is left alone (F10). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
BenCarter44
approved these changes
Aug 19, 2026
Collaborator
There was a problem hiding this comment.
Looks great!
An architectural change before merge though:
Could ORBIT instead simply capture tags from asyncflow tasks and then route it based on that? (Then, the DT plugin would reference an Rhapsody plugin for RaaS). That way the user can do their own exsitu / insitu picks. A follow up to this would be to use the task_dispatcher plugin to support resource pools. The "insitu" is a resource pool, the "exsitu" is a resource pool. This would remove the dependency and need for StreamingActiveLearner / StreamingInvestigator
User code --> DT Plugin on Broker --> Task Dispatcher Plugin on Broker --> Rhapsody Plugin on endpoint
EX:
User: "exsitu" label on asyncflow task --> DT: Passthrough --> T.Dispatcher: resolve "exsitu" pool --> RH: Call Rhapsody
Also, I'm unsure if you want to merge #5 first before this one.
| extra = {} | ||
|
|
||
| if _is_learner(package.cls): | ||
| learn_flow = self._engines.get(EXSITU_ENGINE) |
Collaborator
There was a problem hiding this comment.
Architecturally: Would it be simpler to have asyncflow just tag the backend?
In user's code:
@self.flow.function_task(pool="exsitu", backend="rhapsody")
def do_function():
# ....Then, the task would be routed through ORBIT to the appropriate backend?
Collaborator
There was a problem hiding this comment.
That way it supports more than just StreamingLearnerInvestigator
Collaborator
|
Also another note for the architectural change: Task Dispatcher and Rhapsody are global to the DT plugin. All DT sessions share the same Task Dispatcher. |
BenCarter44
self-requested a review
August 19, 2026 15:08
This was referenced Aug 19, 2026
Member
Author
|
Agree on the direction. Full design notes and TODO sequence are in #7 (comment there); the short version:
(#5 ordering resolved — merged, this PR sits on a devel containing it.) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Milestone M2 of the DTaaS v1 plan (
docs/dtaas-v1-plan.md): ROSE-driven ex-situ learning inside service-hosted twins.Stacked on #4 (M1), which stacks on #3 (M0) — review the commits from
a54dbaaonward, or merge in order #3 → #4 → this.What's in here
digitaltwin.learn.StreamingLearnerInvestigator— ROSE'sStreamingActiveLearnerembedded in aModelInvestigator: fed per-item fromON_INPUT, bootstrap model published up front (no inference deadlock),on_model_ready → publish_new_model, and a clean shutdown contract (learner.stop()+ bounded wait via an internal component stop hook — window-boundary exit instead of killing in-flight training on a shared engine, correct tracker telemetry).'exsitu'engine (config-only addition; aliases'task'when unconfigured; per-name build locks so a slow HPC engine init can't block task-engine builds), inference stays on'task'. No user-facing engine parameter.PluginDT.on_topology_changemarks twinsfailedwithengine endpoint lost: <name>when the endpoint backing one of their engines drops — including twins created after the loss (the session remembers lost endpoints and fails fast with "recreate the session"). Detection only; recovery stays out of v1 per the plan.test/10-learner/— synthetic sensor-calibration regression, no ML-framework dependency; output shows the bootstrap model visibly converging to the true calibration. README documents the dual-engine wiring and thelearnoptional extra (ROSE pinned to PR Add StreamingActiveLearner: learner loop driven by streamed data ROSE#98 / commit 64330d9 until merged).Internally reviewed (request-changes round applied in full: post-loss engine fail-fast, recovery wording, alias-path demo fix, concurrent stop hooks, readable published-model/inference signature mismatch errors).
🤖 Generated with Claude Code