DTaaS viz: live + replay dashboard (Explorer and standalone) and architecture figure - #7
Closed
andre-merzky wants to merge 40 commits into
Closed
DTaaS viz: live + replay dashboard (Explorer and standalone) and architecture figure#7andre-merzky wants to merge 40 commits into
andre-merzky wants to merge 40 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>
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-viz
branch
from
August 17, 2026 14:37
ecd82cd to
ea8e329
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>
`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>
The DT streams carry cloudpickled payloads over ZMQ ports that nobody authenticates: reaching them is code execution in every subscriber (plan risk R7). This puts the same payloads inside ORBIT's token-authenticated star instead, behind the backend seam M0 built, so nothing above it notices. `OrbitPubSubBackend` publishes with `EndpointRuntime.send_notification` and subscribes with `register_callback`, one registration per DT topic under a single `dt_stream` plugin namespace -- ORBIT matches topics by exact equality, which is what DT topics already are. It owns one participant connection per twin (rhapsody's `OrbitExecutionBackend` pattern, loopback wrinkle and all), or rides an injected one. Frames cross from the runtime's callback thread to the host loop through a bounded drop-oldest inbox -- the broker's own discipline, and the conflation semantics the data plane is specified with. Oversized payloads raise at publish, because ORBIT would drop them with nothing but a log line. The two backends had grown the same receive loop, subscriber registry and closed/running state, so those move up into `PubSubBackend`. `DT_STREAM_BACKEND` selects the transport at deployment time. Under 'orbit' the plugin's embedded ZMQ broker is never started at all -- the `connect_stream` seam is the only caller of `stream_addresses`, so that is structural rather than remembered. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The stand-in reproduces the three ORBIT behaviours the backend is written against: exact-match subscriptions, fan-out that does *not* exclude the sender, and callbacks arriving on a foreign thread. The first of those is load-bearing for DT -- a twin's runtime consumes the dtypes its own persistent components publish, so a backend that never got its own events back would deliver nothing at all. Covers namespacing isolation on the second backend (same dtype label, two twins, one broker), unsubscribe and re-subscribe, one wire subscription per topic however many local subscribers, the payload ceiling, drop- oldest under a stalled loop, and teardown: the participant is stopped, an injected one is not, and a runtime that cannot register is not left behind. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A second deployment on the next port, started with DT_STREAM_BACKEND=orbit -- separate because the transport is a deployment-time choice and no client can ask for it. The tests run whole twins on it: sensor to investigator to sink, the M2 learner retraining off the twin's input stream, two twins with the same dtype label staying apart, inference unaffected, churn leaking no participants. The subscriber they attach is itself an ORBIT participant, so even the observation path opens no port -- where the ZMQ suite has to go and ask the service for addresses. The R7 assertion is the direct one: the embedded stream broker is a spawned subprocess, so "none was started" is "the plugin host has no children", checked after a twin has actually streamed. Plus the oversized-payload refusal, which is the one place the 4 MiB frame cap is visible to a user. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`bench_streams.py` times publish -> deliver through the same `PubSubClient` on both backends, so the difference is the transport and nothing else. Loopback: 0.84 ms p50 on ZMQ, 1.94 ms on ORBIT, and about a quarter of the burst throughput. Roughly a millisecond per hop for the security property -- noise against the ~20 ms of a single in-situ prediction. Numbers in the perf README, informational, not a gate. Neither backend acknowledges a subscription, so the harness opens with a round trip it retries until one comes back; without that barrier the first publishes are simply lost and the measurement never starts. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The framework's own pubsub ports authenticate nobody, so the data plane was weaker than the control plane wrapped around it: reaching those ports was code execution in every subscriber, token-free (risk R7). `DT_STREAM_BACKEND=orbit` puts the same payloads on the token- authenticated star instead, and starts no ZMQ broker at all -- there is then nothing left to firewall. Says so plainly, including the two things it does not do (per-tenant auth is still post-v1; the 4 MiB frame cap is new and the ZMQ backend has none), how a reviewer checks the guarantee, and that the ZMQ mitigations remain in force for the deployments that keep it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`_await_running` parks a publish or subscribe that arrives before connect() finished. If the client is closed while somebody waits there, the wait ends on a backend that has already let its transport go -- so re-check, and raise the ordinary closed-client error instead of failing somewhere further in. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- The docstring claimed two bounded queues on the receive path; there are three. The middle one is ORBIT's own `CallbackDispatcher`, and it is the odd one out -- it drops the *newest* frame, not the oldest. Name it, and log its counter alongside ours at close, so a twin that lost samples does not have to be inferred from `seq` gaps in somebody else's log. - `_await_running` parked a caller that arrived before connect. If the connect was then abandoned, `close()` merely cleared the flag and the waiter kept waiting for a connection that would never come. Closing now sets the event before clearing it: the waiters wake, hit `_check_open`, and get the ordinary closed-client error. Tested for both verbs that can park. - The zmq branch of `connect_stream` re-read the environment per twin, which could contradict the choice the plugin resolved once at construction. Name the backend. - `frame_cap()` reaches into a private `EndpointRuntime` attribute; say so, and that the fix belongs upstream. - The benchmark's barrier drained with an `empty()` check, which leaves a barrier message still in flight to arrive mid-measurement and pair with the wrong publish. Wait for quiet instead. Numbers re-measured. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The M3 data plane added broker_url to PubSubConfig; the M0 round-trip test compares the dict shape exactly and follows. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A channel's payload is encoded by the codec its binding names, so the transport has to hand those bytes over untouched -- the `raw` flag the seam already carries. The ZMQ backend honoured it; the ORBIT one did not, so a twin binding a channel on that data plane failed at the first subscribe. The set of raw topics is the same notion in both backends, so it moves up to PubSubBackend where the rest of the shared receive machinery already lives. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Four role lanes on one canvas: the client with a sub-lane per session, the broker with one card per twin, and the task and ex-situ endpoints grouped under an HPC-resources frame. The lanes are roles rather than hosts, so all four are drawn even when a deployment collapses them onto one endpoint. The data layer takes a stream of timestamped frames and folds them into one world -- snapshots of `admin/sessions` and the gateway's SSE notifications -- so it cannot tell a recording from a live stack. This commit brings only the recording half: `index.html` plays the bundled `dt_sample.js` on load and loops it, and play/pause and the speed slider act on the frames rather than on an animation drawn from them. The sample is a real capture of a live stack (one broker hosting `dt`, two rhapsody endpoints): a ROSE learner twin converging on its `rmse` criterion, an in-situ twin created and closed mid-recording, and a twin whose component crashes. It is a .js file assigning one JSON object for exactly one reason -- a classic <script src> loads from `file://` where `fetch()` does not, which is what makes the page self-demo with nothing running and no server. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The data layer gains its second source: a 1 Hz `admin/sessions` poll and
the gateway's SSE feed, turned into the frames the replay half already
consumes -- so `rec` captures the live stream to a recording that plays
back through exactly the same model, and none of the renderer knows which
it is watching.
Live has to be same-origin with the broker, and that is not a choice.
The gateway's CORS allow-list holds a handful of localhost origins, and
the `orbit_broker_token` cookie an EventSource rides is SameSite=Strict:
a page opened from anywhere else cannot reach a live broker even holding
the right token. So the plugin serves the page itself, at
`{namespace}/ui`, from an allow-list of assets -- `{asset}` is a
client-supplied path segment.
Two things the dashboard needs that nothing else did. A learner now
mirrors its per-window criterion into a filtered, read-only `metrics`
dict -- value, threshold, the operator between them, whether the window
met it, the window count and a bounded history, and never the model,
which can be megabytes -- which `DTRuntime.metrics()` collects off the
graph by duck typing (the runtime must not have to know about ROSE) and
`twin_list` / `admin/sessions` carry per twin. And a session summary now
names the endpoint behind each engine role, so the two HPC lanes can say
which hardware they are, and the ex-situ one can say when it aliases the
task one.
The create / destroy / state-change verbs on the arcs stay *inferred*
from the delta between two polls: in v1 nothing on the wire announces
them, and `twin_list` polling is the documented observation mechanism.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
ORBIT's plugin-UI machinery turns out to support arbitrary plugin JS: a
`Plugin.ui_module` path is read by `BrokerPluginHost.get_ui_modules()` and
served by the gateway at `/plugins/<plugin_name>.js`, which the Explorer
imports by convention and drops into a page of its own -- same origin,
same scope, no iframe and no CSP, so a canvas is entirely fine. So the
`dt` plugin declares one.
`dt_explorer.js` is an adapter, not a second dashboard: it hands the
Explorer a template and then dynamically imports the very file the
standalone page loads with a <script src> (it has no imports and no
exports, so it is valid either way) and calls the same `mount()`, in
compact mode and live -- the Explorer page is already same-origin with
the broker, so the auth cookie is simply there. It does not use the
Explorer's `onNotification` hook: that hook only delivers *this* plugin's
events, and the simulation tiles are rhapsody's, so the dashboard keeps
its own EventSource in both hosts.
One limitation to know about, since ORBIT's own docs present `ui_config`
and `ui_module` as interchangeable: `ui_module` is read on the broker
plugin host only. Endpoint-hosted, the Explorer falls back to the
declarative `ui_config` tile and the page has to be opened at
`{namespace}/ui` instead. The gateway also caches a plugin's JS for the
life of the broker process, which the plugin's own `ui/` route does not.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
`round(x, 6)` counts decimal places, so a criterion threshold of 1e-8 -- an ordinary target -- reached the wire as 0.0, and the dashboard drew its target tick at the bottom of the track. Six *significant* figures instead, in the value and the history alike, and infinities and NaN drop out rather than becoming invalid JSON. The history now stops at 24 to match what the sparkline actually keeps. Three things that had no ceiling: - `rec` grew until the tab died. Under the orbit data plane every stream message is an event on the same feed, so a recording left running on a chatty twin is unbounded by construction. It now stops at a frame cap and hands over what it captured, and a stream frame keeps its topic but not its payload -- the topic is what draws the pulse; the payload is a cloudpickled blob no replay can read. - A task whose terminal event fell into an SSE reconnect gap pulsed RUNNING forever, holding its tile and its place in the tally. Tasks are now aged out on silence, which trades a task running longer than the TTL going undrawn for a slot that always comes back. - The Explorer removes a page's node on disconnect without telling the module that drew it, so every reconnect cycle left another poll, EventSource and RAF loop running against a canvas nobody can see. The frame loop notices it has been detached and tears itself down. Checked against the real thing: one dashboard polls 6 times in 6 s, and so does one that has been through a disconnect/reconnect cycle. The poll was a `setInterval` with nothing stopping two from overlapping; out-of-order snapshots read as twins vanishing and coming straight back, complete with inferred arcs. It chains from completion now, carries a request deadline, and drops a response that lands after its source was replaced. A CLOSED EventSource is re-opened with backoff -- the browser only reconnects one that is still CONNECTING, and a dashboard that has silently stopped seeing tasks looks like an idle service. Also: the stream topic's terminator is a NUL, not the pipe the pulse parser was stripping, so a dtype label kept a stray character; one synthetic `dt_stream` frame in the sample (labelled as such -- the recorded stack ran the zmq plane) now pins that path; percent-encoded traversal cases on the asset route; a note against putting a broker token in a URL; and a comment recording that a `Response` really is handled by all three dispatch paths, since this is the one route here that does not return JSON. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Deck-native dark idiom, level colors per the AmSC layer stack (L5 DT / L4 ROSE / L3 AsyncFlow / L2.5 ORBIT / L2 Rhapsody / L1 resources), roles drawn as separate hosts. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The dashboard used to guess which twin a task belonged to: the events carry a uid, an endpoint and nothing else, so attribution went through candidate narrowing, a session block and finally a drawn brace. It could also lie -- a task whose twin stopped before its event arrived was confidently credited to a sibling. Ownership is now recorded where the task is submitted. The runtime hooks its engine once and keeps a bounded ring of task uids per twin, a ContextVar carries the owning twin into every child task, and an instance wrapper on the learner's task registration covers the ROSE side. twin_list reports the ring, and the dashboard joins events against it: 99.6% of task arcs in the bundled trace, 100% in the multi-session one, where every arc used to start at the broker frame. That deletes the guessing subsystem entirely, and the dashboard shrinks by about 550 lines. Arcs anchor two thirds down a card and bow into the empty lane below the top-aligned grid, so a trail belongs visibly to one twin. Lanes now resolve per session, which stops a second session's ex-situ traffic from drawing on the task lane. A stale copy of the dashboard cost a debugging round, so the header carries its version and the page busts its own cache, with a test keeping the two in step. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
andre-merzky
force-pushed
the
feature/dtaas-viz
branch
from
August 17, 2026 22:00
ea8e329 to
733321d
Compare
Contributor
Author
|
Migrated to radical-cybertools#3 — the project now works from the org fork; same branch and commits, stacked on the M3 branch. Review continues there. |
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
1 participant
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.
Live + replay visualization dashboard for the DT service, plus a slide-ready architecture figure.
Stacked on #6 (M3) → #5 → #4 → #3 — review from commit
58d4be4onward, or merge in order.What's in here
src/digitaltwin/service/ui/dt_dash.js(vanilla, no build step, canvas idiom after the AmSC deck's flow diagrams) mounted by a standalone page (works fromfile://, self-demos with a bundled real 61 s recording) and by the ORBIT Explorer viaPluginDT.ui_module(verified end-to-end in headless Chrome against a live broker).admin/sessionspoll (chained, abort-bounded, ordered) + SSE with reconnect backoff and task TTLs; record button captures a bounded replay file (stream payloads stripped); replay uses the play/pause/speed controls.metricsdict fed from ROSE'sIterationState(significant-figure rounding — a1e-8threshold survives the wire), per-sessionendpointsmap, pluginuiroutes with an allow-list (encoded-traversal test-pinned).docs/dtaas-architecture.svg— high-level architecture in the deck's idiom with consistent per-layer colors.Internally reviewed (request-changes round applied in full; security review of the new routes clean). Upstream notes recorded in the README: task→twin attribution needs a rhapsody payload field;
ui_moduleis broker-hosted-only and ORBIT'sui_config/ui_moduledocs conflate the two; gateway sessions carry nox-orbit-srcowner.🤖 Generated with Claude Code