DTaaS M1: ORBIT service plugin (PluginDT/DTSession/DTClient), stacked on M0 - #4
Closed
andre-merzky wants to merge 19 commits into
Closed
DTaaS M1: ORBIT service plugin (PluginDT/DTSession/DTClient), stacked on M0#4andre-merzky wants to merge 19 commits into
andre-merzky wants to merge 19 commits into
Conversation
M0.2/M0.4 plus the transport half of M0.1/M0.5. - PubSubClient takes a mandatory namespace; topics become `dt/<namespace>/dtypes/<label>` so twins which use identical dtype labels no longer cross-subscribe on a shared broker. A topic terminator keeps a label from prefix-matching a longer one (ZMQ SUBSCRIBE is a prefix match; hygiene, not correctness). - Teardown: ZMQ_PS_Client.close() cancels the receive loop, closes the sockets and terminates the context; PubSubClient.unsubscribe_dtype() and close() drop subscriptions and the backend with them. The connect monitor is now attached before connecting (it only reports later events) and is explicitly closed -- disable_monitor() only detaches it, and the leftover socket blocked ctx.term() forever. - ZMQ_Broker binds in the process that runs the proxy (a context does not survive spawn), defaults to a random loopback port and reports what it bound. ZMQ_BrokerProcess embeds it as a spawn-context subprocess with async start/stop; the blocking spawn/join calls run off the event loop. - New `config` module: the one place that decides addresses. Loopback-only defaults (R7), env overridable. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
M0.1/M0.3/M0.5 in the runtime. - DTRuntime.stop(): terminal, idempotent, per-twin. Cancels every task the runtime owns (component main loops, callbacks, dtype consumers, barrier loops), waits for them with a bound, abandons what does not settle, then drops the twin's subscriptions and closes its stream client. The shared engine is never touched. In-flight backend tasks are cancelled best-effort through the awaiting task and abandoned after the timeout. start() after stop() raises. - All done-callbacks are now the one cancellation-safe _task_done, which routes component exceptions into runtime state (failed + last error) instead of the loop's exception handler. RuntimeAPI's own background-task set is gone: it uses the runtime's task plumbing. - RuntimeAPI exposes the injected, namespaced, connected stream client as `runtime.stream` -- persistent components publish through it and never build transport clients or see addresses. - describe() returns a serializable graph/state summary; print_graph() is now just a rendering of it. - Ride-alongs: awaited the coroutine in _internal_agent_inference, fixed WindowDataType.__eq__ comparing a field to itself, removed the dead _to_block machinery and the redundant truthy_list. Barrier's loop is owned by the runtime (Barrier.run) instead of an untracked task with a result-raising callback. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
M0.5/M0.6. pyproject declared no dependencies at all; the versions listed are the ones the framework is developed against (dist name `rhapsody-py`). `digitaltwin` now exports its public API, and pytest (asyncio auto mode, tests under test/unit) plus tox are configured. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
M0.6. Covers stop being terminal/idempotent, component failure -> failed state + last error, bounded abandon of tasks which ignore cancellation, and a leak assertion (no lingering tasks, subscriptions, sockets or contexts after teardown); namespace isolation of identical dtype labels on one broker; persistent components publishing through the injected stream client end to end, including two twins side by side; and broker start/stop cycling with random ports. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
M0.3/M0.6. Persistent components are plain async main_loops which publish through `runtime.stream`: the `@flow.function_task` wrapper and the hand-built `ZMQ_PS_Client` inside every sensor are gone, and with them the last address literals in demo code (a function task would have been cloudpickled to a backend slot for the lifetime of the twin). Every demo now opens one namespaced stream client via `connect_stream_client()` and tears its twin down with `runtime.stop()` before shutting the shared engine down. `local_broker.py` and the demos resolve the broker addresses through `digitaltwin.config`, so the two-terminal model still works and the broker is configurable; the READMEs document the resolution and the loopback binding policy. Not migrated: test/rose_streaming (untracked in this repo). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Guard against stopping a process whose start() failed, and reformat. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Fable review of M0. Stability fixes, no new surface: - The receive loop no longer dies on a bad payload or a raising subscriber: malformed messages are dropped and logged, and a failing subscriber does not starve its siblings. Any other exit of that loop is reported through a new `PubSubBackend.on_error` hook which the runtime routes into `_record_error` -- a silently stalled stream is now a `failed` twin with a last error, not silence. - `stop()` memoizes its teardown: concurrent and repeated callers await the same task and thus only return once the twin is really down (M1's `twin_close` idempotency for free). The caller's cancellation does not abort it -- stop is terminal. - `start()` on a failed twin raises, like `start()` after `stop()`. - Graph mutation after stop raises instead of registering components whose tasks are then silently dropped. - `ZMQ_BrokerProcess` serializes start/stop with a lock: concurrent starts can no longer spawn a second broker or observe (None, None). - `PubSubClient` rejects namespaces containing the topic separator or the terminator -- the aliasing the namespace exists to prevent. - `ZMQ_PS_Client.close()` closes sockets and terminates the context in a `finally`: a cancelled close can no longer leak them permanently behind the one-shot `_closed` guard. - Dropped the instance assignment in `PubSubBackend.__init__` which shadowed each subclass's `label`, and the unexplained sleep in `connect()` (the monitor already confirms the connection). - Broker cycling test no longer asserts distinct ephemeral ports (nondeterministic); it asserts valid ports and that a broker from a fresh cycle actually proxies. `local_broker` prints with flush. New tests: malformed payload / failing subscriber survival, receive loop exit reporting, stream failure -> failed twin, concurrent stop, stopped-graph mutation, namespace validation. 21 passing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The reviewed plan documents the milestone series (M0-M3) references; PR descriptions point at docs/dtaas-v1-plan.md as the authoritative spec. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
andre-merzky
force-pushed
the
feature/dtaas-m1
branch
from
August 16, 2026 09:07
8191f0a to
e085824
Compare
Upstream review (Ben): handing components the live streamer is wrong for anything that does not run in the host process -- the client owns sockets, a receive loop and subscriber queues, none of which a remote machine can reach. This is the stream-endpoint descriptor the plan cut in review round five, back in its minimal form. - `PubSubConfig` is a frozen dataclass of plain fields (namespace, publish/subscribe address, backend kind) and therefore travels as pickle or as a dict. `connect(timeout=None)` opens the backend for its kind, connects it bounded, and wraps it in a namespaced PubSubClient; a client which fails to connect is closed before the error propagates. `resolve()` builds the config for the configured broker, and `connect_stream_client()` is now that plus connect. - A backend declares which kind reopens it (`PubSubBackend.kind`, `zmq` here) and exposes the addresses a remote client would dial; a config for a foreign kind is refused rather than mis-opened. The field is a plain string, so a second backend needs no change here. - `PubSubClient.config` describes the client's own endpoint; `DTRuntime.stream_config` derives the twin's from the injected client (so it cannot go stale) and `RuntimeAPI` exposes both: `stream` stays exactly as it was -- the in-process convenience -- and `stream_config` is what a task ships when its code runs elsewhere. Where shipping it off-host is documented, so is what it implies: the broker has to be reachable from there, which for zmq means a non-loopback bind on a firewalled network, not the loopback default. API compatible: demos and tests are untouched except for new tests. 25 passing; demo 01 re-run live on non-default ports (env-resolved). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
andre-merzky
force-pushed
the
feature/dtaas-m1
branch
from
August 17, 2026 14:37
e085824 to
e72d8a3
Compare
`failed` meant "something broke and I recorded it": the twin's other tasks kept running, its stream client kept its subscriptions, and the whole lot stayed up until a client got around to closing the twin. A component failure is a twin failure -- the others have lost the graph they were part of -- so it now runs the same teardown `stop()` runs and ends in `failed` with `last_error`, not in `stopped`. The traps this had to avoid: - teardown cannot be awaited where the failure is seen. `_record_error` is reached from synchronous done-callbacks (a component task, the stream backend's receive loop via `on_error`) and from inside the teardown itself, so it *schedules* one and keeps the handle. - the teardown task must not be registered in `running_tasks`: teardown cancels that set, so it would cancel itself on its first await. `_start_teardown` creates it outside the set and consumes its exception, since nothing joins the teardown a failure started. - one teardown only. `_stop_task` is the single handle for both doors, so `stop()` on a failed twin joins the teardown already in flight (bounded by the same `STOP_TIMEOUT` the service would have granted) instead of starting a second one, and the state it finds is the state it leaves: `failed` sticks, which is the mirror image of the rule that a stopped twin stays stopped. - the first error is the reported one. Everything after it is fallout -- teardown cancelling the failure's siblings, a stop hook tripping over a half-dead component -- and must not clobber the cause. - `_to_asyncio_task` now refuses on a twin whose teardown is scheduled, not just on a stopped one: a task registered after teardown swapped `running_tasks` out would never be cancelled by anyone. `test_component_failure_sets_failed_state` asserted the old contract (`stopped` after an explicit stop of a failed twin) and is replaced by `test_component_failure_tears_the_twin_down`, which makes the same leak assertions the explicit-stop tests make. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Sensors are external entities. A sensor is not a twin component: it is a process with a lifetime of its own, it precedes and outlives any twin, and one instrument's data is wanted by however many twins care about it. Persistent components can produce data, but they cannot be how data enters a graph. `DTRuntime.add_input(dtype, channel, codec='json')` binds a shared external channel to an input dtype. The channel topic goes on the backend verbatim, with no `dt/<namespace>` prefix, which is exactly what makes it shareable: n twins may bind the same channel and the pubsub fan-out gives each of them every message. A channel claiming the internal prefix is refused, as is an unknown codec, and both are refused at registration rather than on the first message. From the dtype queue onwards, external data is indistinguishable from internal traffic, so consumers, barriers and investigators need no notion of where it came from. `stop()` drops the bindings with every other subscription, and `describe()` reports them. What goes on the wire is a deployment decision, not ours: `json` for the plain scripts and instruments which are the normal producers, `raw` for bytes, `cloudpickle` only for producers inside the same trust domain. An undecodable payload costs its own message and is logged, the same contract malformed internal traffic already had. The codecs and the verbatim subscribe live at `PubSubClient` level, above the backend seam. A backend learns one new thing, through the `backend_params` it already accepted: `raw` payloads are opaque bytes which it hands over untouched, because something above owns their format. Nothing zmq-specific moved upwards, and nothing above knows a topic is a zmq subscription. `ChannelPublisher` is the other half, for code which is not part of the framework at all: a broker, a channel, a codec, no namespace and no runtime. `PubSubConfig` grew `connect_backend()` for it and its namespace became optional, since a channel belongs to no twin. Demos 01 and 04 now run their sensor as a separate process publishing JSON on a shared channel, and their twins bind it with `add_input`. The timer-driven demos (06, 07) stay as they were and are documented as the internal-producer examples. The README explains why the sensor gets a terminal of its own. Tests: two twins on one channel both receive every message, raw delivery, an undecodable payload dropped without stalling the stream, channel and codec validation, bindings in `describe()`, and codec round trips. 40 passing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Records the revision agreed with the user in §1, adds the input-binding item to M0 (item 7) and the verb to the §3 route list, and defers a channel registry in §5. Persistent tasks keep their place as internal producers. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
M0 follow-ups the service depends on: - `ZMQ_PS_Client.connect(timeout=)` (and `connect_stream_client`) is bounded and closes the half-connected client on failure. The twin's background initialization must fail into `failed` + a last error, never park in `initializing` because the stream broker is unreachable. - `config.embedded_stream_addresses()`: an unconfigured *embedded* broker takes a random loopback port (it reports what it bound), where the standalone demo broker keeps the fixed default ports. - `DTRuntime.get_inference()`: public entry to the inference path the service exposes as a verb, so it stops reaching into a private method. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`src/digitaltwin/service/` implements the plugin of the DTaaS plan's section 3, registered out-of-tree via the `radical.orbit.plugins` entry point (`--plugins default,dt`). - `PluginDT`: forces `persistent` sessions and relaxes the owner check so the sid is a bearer capability (reattach by sid after a disconnect); owns the embedded DT stream broker -- one supervised subprocess shared plugin-wide, loopback by default, respawned on the addresses it reported; `admin/sessions` is the single admin route. - `DTSession`: n twins plus the session-shared engines they run on (M1 builds exactly one, `'task'`, an `OrbitExecutionBackend` with `batch_window=0` and `backends=['concurrent']` by default). Teardown stops the twins, then bounds `flow.shutdown()` with `wait_for` -- a bare await would hang on asyncflow's unbounded gather. - `TwinInstance`: `DTRuntime` + its own twin-id-namespaced stream client, with the `initializing -> ready -> running -> stopped | failed` state machine and a last error. - API model: `twin_create` is the only asynchronous verb (client-supplied uuid, background init, returns `initializing`); everything else is a short request/response carrying exactly one graph verb. `start` / `stop` / `twin_close` are idempotent no-ops on a twin already in that state; `twin_list` is the only observation mechanism. - `wire`: cloudpickle-base64 payloads with a client-side size check against the 4 MiB frame cap and a Python/cloudpickle version stamp the service rejects on skew. - The persistent-component guard warns when a component registered `function_task`s at instantiation and is then added as persistent -- the one place where both facts are visible. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`test/09-remote` becomes `test/09-service` and drives the twin through `DTClient` over ORBIT instead of the hand-rolled ZMQ REQ/REP prototype, which is deleted along with its host script (`src/digitaltwin/remote/`, `test/remote_service.py`). `model.py` also splits its inference in two: the engine task returns a plain value and the component wraps it in `TypedData`. ORBIT's rhapsody plugin cloudpickles task *arguments* but JSON-encodes return values, stringifying anything that is neither JSON-safe nor bytes (`plugin_rhapsody.py:547-556`) -- so a DT-typed object cannot come back out of a task. Noted in the demo and its README. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Unit tests (no broker): forced-persistent sessions, sid-as-bearer reattach, twin id and verb validation, the wire format's size and version-skew guards, the persistent-`function_task` warning, and the embedded stream broker -- including the supervisor respawning it on the addresses it reported. Integration tests spin a real stack as fixtures (ORBIT broker on 8031 hosting `dt`, co-located rhapsody endpoint with `backends=['concurrent']` and the notify window at 0) and cover the plan's M1 item 9: inference round trip, two concurrent twins whose identical dtype labels do not cross-subscribe, independent teardown, twin churn with a leak assertion, `twin_close` with an inference in flight, client disconnect and reattach by sid, idempotent retries, start-after-stop and graph-verb-after-stop errors, component crash surfacing as `failed` with a last error, the admin listing, and an endpoint-hosted smoke test. They skip when no broker can be started. Also: a closed twin now reports `closed` (keeping its last error) and drops its runtime, and a rejected verb logs its traceback service-side. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`perf/bench_insitu.py` is the section 6 benchmark, parameterized so it runs against the integration stack (`--broker` / `--endpoint`, or the usual ORBIT resolution). Re-measured here: ~11 ms p50 in-process versus ~19 ms p50 through an ORBIT endpoint with the notify window at 0, and ~334 vs ~35 tasks/s under 50-way concurrency -- the placement decision holds. The README gains a service section: how to host the plugin (`--plugins default,dt` plus a rhapsody endpoint), a client snippet, the two contract notes users trip over (task return values must be JSON-safe or bytes; persistent bodies are not function tasks), and the R7 binding policy -- the embedded stream broker is loopback-by-default and a non-loopback bind needs an explicit decision and a firewalled network. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- `test/unit/test_service.py` importorskips ORBIT and the integration conftest sets `collect_ignore_glob` when it is missing, so an install without the `service` extra still runs the framework suite. - `twin_call` only converts a `CancelledError` into "twin was closed" when the handler itself is not being cancelled -- a host shutdown must keep propagating. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Nine findings from the M1 review, no behaviour the API promises changes. Wire and validation: - `twin_call` hands the payload to the session as an opaque blob and unpickles it only after `_forward` has established that the sid names a live session -- decoding is arbitrary code execution and must not happen for an unknown or expired one. - The version stamp carries `digitaltwin` too, compared exactly: a shipped component class pickles the framework by reference, so the two sides have to be running the same code, not merely compatible pickle machinery. - A hand-built payload with the wrong arity (`TypeError`) is a 409, and a call on a closed session is a 410 -- both were 500s. - `get_inference` clamps a missing or nonsensical timeout to the default, so the service-side wait is literally always bounded. Engines: - The build is now a session-owned task that callers `shield`-await. A twin whose initialization is cancelled halfway used to cancel the build with it and could strand a live `OrbitExecutionBackend` that `close()` never saw; a build landing after the session closed now shuts itself down. `close()` raises the inactive flag first and drains outstanding builds briefly rather than waiting out a 150 s initialization. Tests: - The in-flight `twin_close` case runs twice: once waiting in the service, once on a real task on the endpoint, so the best-effort backend-cancel path is exercised. That task's delay is deliberately short -- the cancel does not reach the endpoint, so a long one would hold a slot for the rest of the suite. - The fd-leak assertion counts descriptors of *the* broker under test (pid from the fixture) and fails rather than skips when it cannot; the guard-warning assertion is scoped to its own twin id; `_spawn`'s log handles are closed on teardown. - Unit coverage for all of the above, including engine-build cancellation and self-disposal. Also: the twin-id uniqueness scan and the insertion it guards are race-free only because nothing awaits between them -- pinned with a comment at both halves. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
andre-merzky
force-pushed
the
feature/dtaas-m1
branch
from
August 17, 2026 22:00
e72d8a3 to
471cf75
Compare
Owner
|
Done. Merged in |
Owner
|
See branch |
Owner
|
Actually, it's been merged to main. Vanilla framework moved to a release branch. |
This was referenced Aug 19, 2026
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 M1 of the DTaaS v1 plan (
docs/dtaas-v1-plan.md): the DT framework exposed as an ORBIT plugin — Digital Twins as a Service.Stacked on #3 (M0): this PR's diff against
mainincludes the M0 commits; review only the commits from949da6donward (or merge #3 first — this PR then shows the M1 delta alone). Merge order: #3 → this.What's in here
digitaltwin.service—PluginDT(routestwin_create/twin_list/twin_close/twin_call/{sid}+admin/sessions; forced-persistent sessions; sid-as-bearer-capability reattach; supervised embedded stream broker, loopback random ports, respawn on the same reported addresses),DTSession(n twins per session; one shared'task'engine via rhapsody'sOrbitExecutionBackend,batch_window=0;wait_for-bounded shutdown; shielded session-owned engine builds),TwinInstance(initializing → ready → running → stopped | failedwith last-error),DTClient(sync; public API mirrors the old 09-remote shape),wire(cloudpickle-base64 packaging with 4 MiB size check and a python/cloudpickle/digitaltwin version stamp, rejected server-side on skew, validated only after session resolution).twin_create(client-supplied uuid → retry-idempotent; background init;twin_listpolling as ground truth), all other verbs short sync,get_inferenceserver-side bounded,stopterminal, idempotent no-ops for repeated lifecycle verbs, unknown-idtwin_close→closed.test/09-service/(sensor → agent → investigator-on-endpoint → sink, end to end); the old ZMQ REQ/REP prototype (src/digitaltwin/remote/,test/remote_service.py) is retired.dtsmoke test; churn/fd-leak assertions, reattach-by-sid, crash→failed, close-with-in-flight-backend-task).perf/bench_insitu.pychecked in; re-measured 19.4 ms p50 / 334 tasks/s through the orbit endpoint (vs 11.6 ms / 35 tasks/s in-process pool) — the plan's §6 placement conclusion holds.Requires: radical.orbit ≥ devel@dc26036 (P2 notify-window knob) and the radical.asyncflow non-main-thread fix (radical-cybertools/radical.asyncflow#92). Both are open/merged upstream PRs; the
serviceextra is optional — the package works without ORBIT installed.Internally reviewed (approve); all nine review findings applied.
🤖 Generated with Claude Code