diff --git a/.gitignore b/.gitignore index 3113291..a1e7d15 100644 --- a/.gitignore +++ b/.gitignore @@ -229,3 +229,10 @@ __marimo__/ # runtime residue from asyncflow / rhapsody test runs asyncflow.session.*/ telemetry-output/ + +# demo venvs (deploy/install.sh) +ve.demo/ +ve3/ + +# operator-local demo env (broker IP, cert path); see the runbook +/demo.sh diff --git a/README.md b/README.md index 27618ec..a97f245 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,188 @@ no filesystem with the service; `as_executable=False` sends them as cloudpickled function tasks instead (the component warns if it finds executable ones). `test/10-learner/` is a complete worked example. +### Watching it run: the dashboard + +`src/digitaltwin/service/ui/` holds a dependency-free canvas dashboard +(one JS file, no build step) that draws the service as role lanes: the +**client** with a sub-lane per session, the **sensors** below it, the +**broker** with one card per twin -- short uuid, colour-coded state, +namespace, stream-backend badge, and a convergence bar per learner metric +-- and, grouped under an *HPC resources* frame, the **task** and +**ex-situ endpoint** lanes where the twins' simulation tasks appear as +tiles. The lanes are roles, not hosts: a single-endpoint deployment +still gets both endpoint lanes, and the ex-situ one is labelled +`aliases task`. + +The sensors lane is observed, not declared: every `dt_stream` topic names +a twin and a dtype, and only a twin's own persistent components publish, +so one tile per `(twin, dtype)` *is* the set of sensors. It sits outside +the broker frame because that is where a reader looks for where data +comes from, and it says `in the plugin host`, because in v1 that is where +those components run. + +Three ways to open it, in increasing order of what they need: + +```sh +# 1 - offline: replay the recording bundled in the repo, no stack at all +xdg-open src/digitaltwin/service/ui/index.html + +# 2 - live, served by the broker itself (the only way live works -- see below) +xdg-open https:///broker/dt/ui + +# 3 - live, inside the ORBIT Explorer: open https:/// and pick the +# 'Digital Twins' plugin. The plugin ships the page as its `ui_module`; +# nothing to install +``` + +**Live mode has to be same-origin with the broker.** The gateway's CORS +allow-list holds a handful of `localhost` origins, and the +`orbit_broker_token` cookie that the `EventSource` rides is +`SameSite=Strict` -- so a page opened from anywhere else cannot reach a +live broker even with the right token. Served from the broker there is +no cross-origin request at all. The broker's certificate is self-signed: +visit `https:///` once and accept it, which is also where the +token is entered (that mints the cookie the dashboard then reuses). +Everything else -- replaying a recording, loading one by drag-and-drop -- +works from `file://` with no server. + +The data layer treats live and replayed input identically: a stream of +timestamped frames, either an `admin/sessions` poll at 1 Hz plus the +gateway's SSE feed, or the same frames read back from a recording. So +`rec` captures the live stream to a JSON file, `load…` (or a drop on the +canvas) replays one, and the play/pause and speed controls act on the +data rather than on an animation. The schema is documented at the top of +`dt_dash.js` and checked by `test/unit/test_ui_recording.py`. + +One thing the picture makes obvious once it is drawn: **the runtime never +publishes a component's answer**. An inference result goes to the next +component on that dtype over an in-process queue, and is dropped if +nothing is registered there. For anything outside the service to see a +result, a component has to publish it (`EchoSink` in the demos does +exactly that, which is why the sensors lane shows a twin's results as +well as its readings), and the client then subscribes to the twin's +stream with the `PubSubConfig` the twin reports. A client's own +`get_inference` is the other path, and the only one that answers the +caller directly. + +The bundled recording was captured against a **`DT_STREAM_BACKEND=orbit`** +deployment, so it carries the twins' own stream traffic (~300 events, two +dtypes) and the pulses that are drawn from it. **A live dashboard will +not show those pulses yet, and this is an upstream gap, not a bug here**: +`Gateway._sse_frame` in radical.orbit is a bare `json.dumps`, a DT stream +payload is `bytes`, and every one of those events is therefore dropped +with `TypeError: Object of type bytes is not JSON serializable` (the +broker logs one `tap callback failed` per event -- 292 of them in a 45 s +run). Adding a `default=` to that one call is enough; with it the events +flow and the pulses appear, which is how the bundled capture was taken. +Everything else in the dashboard works against an unpatched broker. + +Four things the dashboard reads that nothing else needed. `twin_list` +and `admin/sessions` now carry a per-twin `metrics` dict -- a filtered, +read-only view of a learner's per-window criterion (`value`, `threshold`, +`operator`, `should_stop`, window count and a bounded history), never the +model itself -- a per-twin `calls` count per verb, a per-twin `tasks` list +of the uids that twin most recently submitted, and a per-session +`endpoints` map naming the hardware behind each engine role. + +**Most of what an arc says is inferred**, because in v1 almost nothing on +the wire announces it -- the exception is the task arcs, which are now +joined on a uid the service records -- and the drawing says which is which: + +- a solid arc **client to broker** is a `create`, and back a `destroy`: + a twin that appeared in this poll and was not in the last one, or the + reverse; +- a dim dashed arc **broker to a session sub-lane** is a state + *transition* seen between two polls (`initializing`, `ready`, + `running`, `failed`, `stopped`). Nothing is pushed to a client in v1 -- + the arc stands for the `twin_list` response that would carry the new + state, which is also what the tick on each session card marks, once per + poll; +- a green hop **sensor tile to twin card** is one stream message. Green + is the data plane's colour and only the stream pulses in it: the hop + back out of an endpoint lane is violet, the colour the deck gives the + AsyncFlow engine a task result returns through (red when it failed); +- an arc **session sub-lane to twin card** is one client call that was + answered. `get_inference` is amber, the request a client is actually + waiting on, and so is the answer that comes back to it; the other verbs + are cyan and carry no answer arc, because what they return is a state + nobody waits for. The service counts the verbs it answered per twin + (`calls` in the twin summary) and the arcs are drawn from the difference + between two polls, so what you see is completed round trips, never a + call in flight; +- an arc **into an endpoint lane** is a task, and it leaves the card of the + twin that submitted it -- low on that card's centre line (`CARD_ANCHOR`), + because a point inside the card belongs to exactly one of them while an + edge is shared with whatever sits next to it, and low is where a curve + bowing downward is out from under the card at once. The card grid is + top-aligned in its lane, so those curves bow into the space below it + rather than across the cards between their ends. That is known rather than guessed, and it took + the service to know it: a `task_status` notification carries a uid and an + endpoint and nothing else, so ownership is recorded where the submission + happens. asyncflow assigns each task a uid (`task.NNNNNN`) in the + component description and rhapsody's backend keeps it, which is the same + uid the notification carries -- so the twin remembers the uids it + submitted (`DTRuntime.note_task`, a ring of the newest `TASK_UID_RING`), + `twin_list` carries them as `tasks`, and the dashboard joins on them. + Two paths reach that ring: the runtime records the future it is about to + await, and -- because a real inference task is usually a plain coroutine + that awaits a flow task the runtime never sees -- the engine's own + component registration is wrapped, with the owning twin carried in a + `ContextVar` that asyncio copies into every task underneath. ROSE's + ex-situ tasks are the third case: `Learner._register_task` is wrapped per + instance in `StreamingLearnerInvestigator.main_loop`, where the runtime is + in hand, so training, active learning and the criterion are the twin's too + without ROSE changing. + + A notification beats the 1 Hz poll that explains it, so a task's arcs wait + up to `OWNER_WAIT` for the join and only then leave the broker lane's + edge, which claims nothing. A failed or closed twin keeps the arcs it + really did submit -- its card is on the canvas for a while yet, and the + truth is better than tidiness. The endpoint no longer picks the lane by + itself either: the role is a per-session answer (one endpoint can be one + session's task engine and another's ex-situ engine), and an endpoint no + session of ours declared is another deployment's, so its tasks are drawn + on neither lane. No arc ever leaves the client lane: no task is + submitted from there. + +A twin that leaves `twin_list` keeps its card for nine seconds, dimmed, +with the last state pill the service reported and a `closed` mark, then +fades. The `destroy` arc still fires when it goes; the card is what is +left to read afterwards, and a run that ends by closing its twins used to +erase the evidence a second later. A lingering card yields its grid slot +to a live twin if the lane runs out of room, and it keeps the arcs of the +tasks it submitted while it was alive. + +`get_inference` and the ex-situ lane, since the pairing invites the wrong +conclusion. A probe is answered by running the investigator's inference +task on the **task** engine; the ex-situ engine only ever receives +training windows. In the bundled 56 s capture the service counted 24 +`get_inference` round trips, and every one of the 17 poll windows that +held a probe also held new task-endpoint tasks -- but it held five to +seven of them, because the twin's streaming pipeline is submitting there +continuously, and a `task_status` carries no verb to tell them apart. So +the task lane shows a dim amber `inference` pill for the beat in which a +probe was served: the *when*, on the lane that could have run it, and no +claim about which tile it was. Ex-situ traffic in that capture ran at +2.25 tasks per probe and also in windows with no probe at all, which is +what training windows look like. + +If the dashboard does not look like this, check the version the header +draws next to the stream pill against `VERSION` in `dt_dash.js`. A +browser keeps a `file://` script well past the edit that changed it (hence +the `?v=` on the page's script tags), and the copy the broker serves is +the *installed* one -- as new as the last `pip install .`, no newer. An +older build attributed nothing: every task arc left the broker frame's +right edge, which is the symptom to recognise. + +The Explorer integration is broker-hosted only. ORBIT reads `ui_module` +in `BrokerPluginHost.get_ui_modules()` and nowhere else, so an +endpoint-hosted `dt` plugin gets the declarative `ui_config` tile and no +dashboard page; `{namespace}/ui` still serves it directly. The gateway +also caches a plugin's JS for the life of the broker process, so editing +`dt_explorer.js` needs a restart -- `{namespace}/ui/dt_dash.js`, which +the plugin serves itself, does not. + ### When an endpoint disappears (R8) `OrbitExecutionBackend` does not reconnect and components bind their diff --git a/deploy/install.sh b/deploy/install.sh new file mode 100755 index 0000000..14fcdfe --- /dev/null +++ b/deploy/install.sh @@ -0,0 +1,130 @@ +#!/usr/bin/env bash +# +# Deploy the DTaaS stack for the live demo. +# +# Run this on every host that takes part -- the broker host, any host +# running a rhapsody endpoint, and the client. It installs the same +# pinned commit everywhere, which matters more than it looks: the service +# checks Python and cloudpickle at minor-version granularity and rejects +# skew, and it compares `digitaltwin` by exact version string -- which is +# `0.0.1` on every commit of this branch, so a *commit* mismatch would +# NOT be caught by that gate. Pinning here is the only thing standing +# between us and a confusing unpickle failure mid-demo. +# +# ./install.sh [venv-dir] +# +# role broker | endpoint | client (informational; same install) +# venv-dir default ./ve.demo +# +set -euo pipefail + +ROLE="${1:-}" +VENV="${2:-$PWD/ve.demo}" + +REPO="https://github.com/radical-cybertools/digital.twins" +REF="4b3defd30c9c9d2376fee2e728e4d106eae54447" # feature/dtaas-viz, post devel merge + +# The radical dependencies must NOT come from naive PyPI resolution: PyPI's +# rhapsody-py 0.4.0 lacks `rhapsody.backends.execution.orbit` (the +# OrbitExecutionBackend the whole service runs on), and PyPI's asyncflow +# 0.5.0 lacks the non-main-thread engine fix the broker-hosted plugin +# needs. These pins are the exact commits the verified laptop stack was +# built from -- all pushed to public radical-cybertools repos. +ASYNCFLOW="radical.asyncflow @ git+https://github.com/radical-cybertools/radical.asyncflow@d9f7ca084769a2b72845f069fa141c13177a0800" +# rhapsody's [telemetry] extra is required, not optional, at this commit: +# the ORBIT plugin calls `session.start_telemetry()` whenever it exists, and +# that path hard-imports opentelemetry -- an endpoint without it fails every +# session init with "No module named 'opentelemetry'". (Known upstream gap; +# the proper fix is an ImportError guard in orbit's plugin_rhapsody.) +RHAPSODY="rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2" # f479c75 + participant_name + engine role + +# orbit: the 0.5.0 RELEASE carries the SSE bytes fix (#113) -- PyPI is fine +ORBIT="radical.orbit==0.5.0" + +# ROSE: PyPI's `rose` is an UNRELATED project (a version-string helper) which +# pip will happily install for the `learn` extra -- and the learner then dies +# on `import rose.al`. Pin the real one, same commit the verified stack uses. +ROSE="rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17" + +# Python minor version is part of the wire contract (cloudpickle is not +# portable across minors, and the service rejects skew at the first verb), +# so it is pinned, not discovered. EVERY host must use the same value: +# export the same DT_PYTHON everywhere, or take the default everywhere. +# 3.12 is the demo choice -- radical.3 has it, and it matches the dragonhpc +# constraint should that backend ever join. +PYTHON="${DT_PYTHON:-python3.12}" + +case "$ROLE" in + broker|endpoint|client) ;; + *) echo "usage: $0 [venv-dir]" >&2; exit 2 ;; +esac + +command -v "$PYTHON" >/dev/null || { + echo "ERROR: $PYTHON not found. The service compares Python at minor" >&2 + echo " granularity and rejects skew at the first verb, so every" >&2 + echo " host must run the same minor. Set DT_PYTHON (same value" >&2 + echo " on every host) if it is installed under another name." >&2 + exit 1; } + +echo "==> $ROLE: creating $VENV with $($PYTHON -V)" +"$PYTHON" -m venv "$VENV" +"$VENV/bin/pip" install --quiet --upgrade pip + +# pinned deps first: with these already satisfied, resolving digitaltwin's +# requirements will not reach for the broken PyPI variants +echo "==> $ROLE: installing pinned radical deps (asyncflow, rhapsody, orbit, rose)" +"$VENV/bin/pip" install --quiet "$ASYNCFLOW" "$RHAPSODY" "$ORBIT" "$ROSE" + +# Same extras on every host. The endpoint arguably needs less, but a task +# body that closes over anything from `digitaltwin` would fail to unpickle +# there, and uniformity is cheaper than being clever about it at 2am. +echo "==> $ROLE: installing digitaltwin @ ${REF:0:8} (+ service, learn)" +"$VENV/bin/pip" install --quiet "digitaltwin[service,learn] @ git+$REPO@$REF" + +# soft dependency of the demo driver: highlighted api snippets. The +# driver degrades to plain text without it -- never demo-critical. +"$VENV/bin/pip" install --quiet pygments + +# (the SSE bytes fix that used to be patched in here is upstream now -- +# radical.orbit#113 -- and rides in via the ORBIT pin above) + +# belt and braces: fail HERE, not mid-demo, if pip quietly swapped one out +"$VENV/bin/python" - <<'CHECK' +import rhapsody.backends.execution.orbit # noqa: F401 (PyPI 0.4.0 lacks this) +from radical.orbit import EndpointRuntime # noqa: F401 +import radical.asyncflow # noqa: F401 + +# the SSE tap must survive a bytes payload (radical.orbit#113); without it +# the dashboard shows tiles but no stream pulses +from radical.orbit.gateway import Gateway +Gateway._sse_frame("notification", {"data": b"\x80"}) + +# the real ROSE, not PyPI's homonym (a version-string helper) +from rose.al.streaming_learner import StreamingActiveLearner # noqa: F401 +print("==> dependency sanity: OK") +CHECK + +echo +echo "==> $ROLE: version stamp -- must be IDENTICAL on every host" +"$VENV/bin/python" - <<'PY' +from digitaltwin.service.wire import version_stamp +import json, sys, platform +print(json.dumps(version_stamp(), indent=2)) +print("host:", platform.node()) +PY + +cat < $ROLE: done. Still needed by hand: + + ~/.radical/orbit/ must hold the ORBIT credentials. + + broker host broker_cert.pem, broker_key.pem (mode 0600), broker.token + endpoint host broker_cert.pem, broker.token + client host broker_cert.pem, broker.token + + The cert is *pinned*, not validated against the hostname, so the one + we already use works for a broker on any host -- no regeneration. + The key never leaves the broker host. + +NOTE diff --git a/deploy/run-broker.sh b/deploy/run-broker.sh new file mode 100755 index 0000000..74db1ab --- /dev/null +++ b/deploy/run-broker.sh @@ -0,0 +1,30 @@ +#!/usr/bin/env bash +# +# The DTaaS host: ORBIT broker + the `dt` plugin. Run this on radical.3. +# +# ./run-broker.sh [venv-dir] +# +set -euo pipefail +VENV="${1:-$PWD/ve.demo}" + +# The broker needs its *own* URL in the environment, not just on the CLI: +# the `dt` plugin builds a rhapsody client from it when a twin is created, +# and without it twin creation fails with a misleading +# "twin ... failed to initialize: Broker URL required". Pointing it at +# localhost is right -- the plugin is talking to the broker it lives in. +export RADICAL_ORBIT_BROKER_URL="${RADICAL_ORBIT_BROKER_URL:-wss://localhost:8000}" + +# The twins' data plane. `orbit` puts stream traffic inside the +# token-authenticated ORBIT channel instead of the plugin's embedded ZMQ +# broker -- which is both the better story and the only way the dashboard +# can see the traffic at all, since the pulses are drawn from the +# gateway's event tap. With `zmq` the stream never touches ORBIT and the +# sensors lane stays quiet. +export DT_STREAM_BACKEND="${DT_STREAM_BACKEND:-orbit}" + +echo "broker : 0.0.0.0:8000" +echo "self-url: $RADICAL_ORBIT_BROKER_URL" +echo "dataplane: $DT_STREAM_BACKEND" +echo + +exec "$VENV/bin/radical-orbit-broker.py" --plugins default,dt "${@:2}" diff --git a/deploy/run-endpoint.sh b/deploy/run-endpoint.sh new file mode 100755 index 0000000..efccd13 --- /dev/null +++ b/deploy/run-endpoint.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +# +# A rhapsody endpoint: this is where a twin's compute actually runs. +# The demo wants two, so the dashboard's `task` and `exsitu` lanes are +# distinct hardware rather than one endpoint aliased twice. +# +# ./run-endpoint.sh [venv-dir] +# +# ./run-endpoint.sh dt_task_ep radical.3 +# ./run-endpoint.sh dt_exsitu_ep radical.3 +# +set -euo pipefail +NAME="${1:?usage: $0 [venv-dir]}" +BROKER="${2:?usage: $0 [venv-dir]}" +VENV="${3:-$PWD/ve.demo}" + +export RADICAL_ORBIT_BROKER_URL="wss://$BROKER:8000" + +# Batching a notification window in front of a demo only adds latency +# nobody can see the reason for: 0.25s per round trip was the whole of an +# earlier benchmark surprise. +export RADICAL_ORBIT_RHAPSODY_NOTIFY_WINDOW="${RADICAL_ORBIT_RHAPSODY_NOTIFY_WINDOW:-0}" +export RADICAL_ORBIT_RHAPSODY_BACKEND="${RADICAL_ORBIT_RHAPSODY_BACKEND:-concurrent}" + +# A cloudpickled task body has no other way to find out where it ran, and +# the demo shows in-situ and ex-situ landing on different hardware. +export DT_ENDPOINT_TAG="$NAME" + +echo "endpoint: $NAME -> $RADICAL_ORBIT_BROKER_URL" +echo + +exec "$VENV/bin/radical-orbit-endpoint.py" -n "$NAME" diff --git a/docs/dtaas-architecture.svg b/docs/dtaas-architecture.svg new file mode 100644 index 0000000..cd577fd --- /dev/null +++ b/docs/dtaas-architecture.svg @@ -0,0 +1,166 @@ + + + + + + + + + + + + + + + + + DT-as-a-Service — architecture + digital twins run as long-lived services on a persistent ORBIT broker (v1) + + + + CLIENT HOST + + + DTClient + • create_twin(uuid) · add_* · start + • verbs are synchronous + • twin_create: poll to ready + + + sessions + • attach, detach, reattach by sid + • twin_list reports current state + + + viz dashboard + • Explorer or standalone page + • live + replay (work in progress) + + + + INSTRUMENTS + + sensors · streams + • OPC-UA / MQTT / demo feeds + • continuous input streams + + + + BROKER HOST — DTAAS + persistent sessions; reattach by sid; twins survive client disconnects + + + ORBIT broker + • token-gated WS ingress   • gateway (HTTP, SSE) + + + + PLUGIN: DT + + + DTSession (n twins) + • AsyncFlow engines task | exsitu, via OrbitExecutionBackend (Rhapsody) + + + + TwinInstance — DTRuntime + + + + sensor task + + sci agent + + investigator + + sink + + + + + • StreamingLearnerInvestigator (ROSE): in-situ predict, ex-situ learn + • convergence metrics per window, reported via twin_list + initializing → ready → running → stopped | failed + + + + stream data plane + • ZMQ broker (local, loopback)   • ORBIT eventing (token domain) + + + + HPC RESOURCES + + + endpoint: task + • typically co-located with the broker + • rhapsody plugin + • backend: concurrent · dragon · flux + • in-situ inference tasks + • about 20 ms per prediction + + + + + + + + endpoint: exsitu + • remote HPC allocation or cluster + • rhapsody plugin + • backend: dragon · flux (HPC scale) + • ROSE training + active-learning tasks + • aliases task if unconfigured + + + + + + + + + + verbs + + + state + + + + sensor streams + + + + inference + + + + train / AL + + + + publish model + + + + + L5 DT framework + + L4 ROSE + + L3 AsyncFlow + + L2.5 ORBIT + + L2 Rhapsody + + L1 resources + + layer stack per amsc/architecture/dt-framework.md; Rhapsody appears twice: control-side engines on the broker, compute-side backends on the endpoints + diff --git a/pyproject.toml b/pyproject.toml index 4e4c58f..21b20b4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,6 +48,11 @@ dt = "digitaltwin.service.plugin:PluginDT" [tool.setuptools.packages.find] where = ["src"] +# the dashboard: the Explorer reads `ui_module` off the installed package, +# and the plugin serves the rest from `{namespace}/ui` +[tool.setuptools.package-data] +"digitaltwin.service" = ["ui/*.html", "ui/*.js"] + [tool.pytest.ini_options] testpaths = ["test/unit", "test/integration"] asyncio_mode = "auto" diff --git a/src/digitaltwin/learn.py b/src/digitaltwin/learn.py index 7a2870b..76707e3 100644 --- a/src/digitaltwin/learn.py +++ b/src/digitaltwin/learn.py @@ -51,6 +51,7 @@ async def predict(in_data, slope=0.0): import asyncio import contextlib import logging +import math from typing import Any, Callable, Optional @@ -58,7 +59,7 @@ async def predict(in_data, slope=0.0): from rose.al.streaming_learner import StreamingActiveLearner # type: ignore from .components import ModelInvestigator, TypedData -from .runtime import RuntimeAPI +from .runtime import RuntimeAPI, hook_engine, note_flow_task logger = logging.getLogger(__name__) @@ -72,6 +73,30 @@ async def predict(in_data, slope=0.0): # ROSE's own per-window bookkeeping -- state, but not model parameters _ROSE_STATE_KEYS = ("window_size",) +# how much of the criterion's metric history travels in `metrics`. A +# days-long twin accumulates one value per window forever; the dashboard +# only ever draws a sparkline of the recent tail, and it keeps 24 points +# (`SPARK_MAX` in `service/ui/dt_dash.js`) -- so sending more is paying +# for something nothing reads. +METRIC_HISTORY = 24 + + +def _number(value: Any) -> Optional[float]: + """A JSON-safe float, or `None` -- a criterion may not have run yet. + + Six *significant* figures, not six decimal places: a criterion + threshold of 1e-8 is an ordinary target, and rounding it by decimals + would put 0.0 on the wire. Infinities and NaN are dropped rather + than emitted, because neither survives JSON. + """ + + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + + number = float(f"{value:.6g}") + + return number if math.isfinite(number) else None + class StreamingLearnerInvestigator(ModelInvestigator): """A `ModelInvestigator` with a ROSE `StreamingActiveLearner` inside. @@ -113,6 +138,13 @@ def __init__( # set by the subclass; the in-situ half of the pair self.inference_task: Optional[Callable] = None + # Read-only observation surface, refreshed once per window and + # carried by `twin_list` (see `_record_metrics`). Nothing in the + # framework reads these -- they exist so an operator can see a + # learner converging without a second channel. + self.metrics: dict = {} + self.windows: int = 0 + self._started = False self._finished = asyncio.Event() @@ -169,6 +201,7 @@ async def main_loop(self, runtime: RuntimeAPI): ) self._warn_local_learner_tasks() + self._own_learner_tasks(runtime) runtime.set_inference_task(self.inference_task) runtime.subscribe_to_topic(RuntimeAPI.ON_INPUT, self._feed) @@ -184,6 +217,7 @@ async def main_loop(self, runtime: RuntimeAPI): try: async for state in self.learner.start(): + self._record_metrics(state) self.on_window(state) finally: @@ -191,6 +225,75 @@ async def main_loop(self, runtime: RuntimeAPI): self.learner.stop() self._finished.set() + def _own_learner_tasks(self, runtime: RuntimeAPI) -> None: + """Record the ex-situ tasks as this twin's, as ROSE submits them. + + Every training / active-learning / criterion task goes out through + `Learner._register_task`, which returns the asyncflow future -- so an + instance-attribute wrapper here sees all three without ROSE knowing. + The engine is hooked as well, because the uid is minted a tick after + the future is handed back; the wrapper's own reading catches the case + where the hook cannot (an engine ROSE replaced), and both write to the + same bounded ring, which ignores a uid it already has. + + Best-effort by construction: a ROSE that stops routing through + `_register_task` loses the attribution, not the learning. + """ + + owner = getattr(runtime, "_runtime", None) + if owner is None: + return + + hook_engine(self.learn_flow, owner) + + inner = getattr(self.learner, "_register_task", None) + if inner is None or getattr(self.learner, "_dt_owned", False): + return + + def registered(*args, **kwargs): + future = inner(*args, **kwargs) + try: + note_flow_task(future, owner) + except Exception as exc: + logger.debug("learner uid capture failed: %s", exc) + + return future + + self.learner._register_task = registered + self.learner._dt_owned = True + + def _record_metrics(self, state: Any) -> None: + """Mirror the window's criterion state into `self.metrics`. + + A filtered, JSON-safe view -- the value, the target it is compared + against, the operator doing the comparing and whether this window + met it -- and never the model, which can be megabytes. This is + what `twin_list` and `admin/sessions` carry per twin. + """ + + self.windows = state.iteration + 1 + name = state.metric_name + if not name: + return + + window = (state.metric_history or [])[-METRIC_HISTORY:] + history = [value for value in map(_number, window) + if value is not None] + + self.metrics = { + name: { + "value": _number(state.metric_value), + "threshold": _number(state.metric_threshold), + # '' for a standard metric, whose operator ROSE knows + # itself; the consumer then falls back on `should_stop` + "operator": (self.learner.criterion_function + or {}).get("operator") or None, + "should_stop": bool(state.should_stop), + "windows": self.windows, + "history": history, + } + } + async def _feed(self, in_data: TypedData) -> None: """`ON_INPUT`: everything the twin sees also feeds the learner.""" diff --git a/src/digitaltwin/runtime.py b/src/digitaltwin/runtime.py index e059974..ccb1577 100644 --- a/src/digitaltwin/runtime.py +++ b/src/digitaltwin/runtime.py @@ -10,7 +10,8 @@ import asyncio import logging -from collections import defaultdict +from collections import defaultdict, deque +from contextvars import ContextVar from dataclasses import dataclass, field from typing import cast @@ -46,6 +47,84 @@ # bounded wait for in-flight tasks to settle on stop() STOP_TIMEOUT = 10.0 +# How many recently submitted task uids a twin remembers. This is an +# observation surface, not a ledger: `twin_list` carries it so an observer +# can say which twin a `task_status` notification belongs to, and one poll +# period's worth of submissions is all that is needed for that. A uid the +# ring has dropped is simply unattributed. asyncflow's uid counter is +# process-global and only `reset_uid_counter()` rewinds it (at backend +# shutdown), so within one host a uid never names two different twins. +TASK_UID_RING = 24 + +# The twin whose work the current asyncio task is doing. asyncio copies the +# context into every task it creates, and asyncflow registers a component +# from a task of its own, so a runtime that stamps this once owns every uid +# assigned underneath it -- including the ones a user component submits from +# inside its own coroutine, which the runtime never sees. +_OWNER: ContextVar[Optional["DTRuntime"]] = ContextVar("dt_task_owner", + default=None) + + +def note_flow_task(fut: Any, owner: Optional["DTRuntime"] = None) -> Optional[str]: + """Record the uid asyncflow assigned to `fut` against its owning twin. + + The uid is the one rhapsody publishes in `task_status`: asyncflow puts + `task.NNNNNN` in the component description (`_assign_uid`) and the + execution backend keeps it (`setdefault`), so joining on it is exact. + + Returns the uid when there was one. A plain coroutine -- an inference + task that is not a flow task -- has none and is skipped, and so is a + future asyncflow has not stamped yet (`hook_engine` catches those). + """ + + who = owner or _OWNER.get() + desc = getattr(fut, "task", None) + uid = desc.get("uid") if isinstance(desc, dict) else None + + if who is None or not uid: + return None + + who.note_task(uid) + + return uid + + +def hook_engine(flow: Any, owner: Optional["DTRuntime"] = None) -> None: + """Capture every uid this engine assigns, at the moment it assigns it. + + asyncflow stamps the future from a task of its own, so the uid is not on + it when a submitting call returns -- which is exactly when a twin would + like to record it. `_register_component` is where the uid is minted, so + that is what this wraps, once per engine object; ownership comes from the + context (see `_OWNER`), because one engine serves every twin of a + session. `owner` pins it for an engine driven from outside the runtime's + own tasks (the ex-situ one, whose loop is ROSE's). + + In-package coupling to one asyncflow internal, deliberately: the + alternative is a uid the service cannot know, and every honest + alternative was tried first (see the README). + """ + + inner = getattr(flow, "_register_component", None) + if inner is None or getattr(flow, "_dt_uid_hook", False): + return + + def hooked(comp_fut, comp_type, comp_desc, *args, **kwargs): + result = inner(comp_fut, comp_type, comp_desc, *args, **kwargs) + try: + uid = comp_desc.get("uid") + who = _OWNER.get() or owner + # blocks are not tasks and never appear in `task_status` + if who is not None and isinstance(uid, str) and uid.startswith("task."): + who.note_task(uid) + except Exception as exc: # never break a submit + logger.debug("uid capture failed: %s", exc) + + return result + + flow._register_component = hooked + flow._dt_uid_hook = True + @dataclass(frozen=True) class _InputBinding: @@ -618,9 +697,36 @@ def __init__(self, flow: WorkflowEngine, streamer: PubSubClient) -> None: # Its presence is also what closes the twin for new work. self._stop_task: Optional[asyncio.Task] = None + # Task ownership, for anything watching from outside: the uids this + # twin most recently submitted (see `TASK_UID_RING`). A submission + # overwrites the oldest, which is also what makes it self-healing -- + # nothing has to be cleaned up, and a twin that stops submitting + # stops appearing in new notifications. + self._task_uids: deque[str] = deque(maxlen=TASK_UID_RING) + self._task_seen: set[str] = set() + + hook_engine(flow) + # a stalled stream is a twin failure, not a log line streamer.on_error = self._record_error + def note_task(self, uid: str) -> None: + """Record a task uid as this twin's. Idempotent and bounded.""" + + if uid in self._task_seen: + return + + if len(self._task_uids) == self._task_uids.maxlen: + self._task_seen.discard(self._task_uids[0]) + + self._task_uids.append(uid) + self._task_seen.add(uid) + + def task_uids(self) -> list[str]: + """The uids this twin submitted most recently, oldest first.""" + + return list(self._task_uids) + @property def stream_config(self) -> PubSubConfig: """This twin's stream endpoint as plain data (see `PubSubConfig`). @@ -780,6 +886,18 @@ def _annotated(self): yield ant yield from ant.investigators.values() + async def _owned(self, func, *args, **kwargs): + """Run a component's coroutine as this twin's work. + + The stamp lives in the task's own context, so it reaches every task + created underneath -- asyncflow's registration among them -- and no + sibling twin's. + """ + + _OWNER.set(self) + + return await func(*args, **kwargs) + def _to_asyncio_task(self, func, *args, **kwargs) -> Optional[asyncio.Task]: """Schedule a coroutine as an :class:`asyncio.Task` and track its completion. @@ -799,7 +917,7 @@ def _to_asyncio_task(self, func, *args, **kwargs) -> Optional[asyncio.Task]: logger.debug("twin is %s - not running %s", self.state, func) return None - task = asyncio.create_task(func(*args, **kwargs)) + task = asyncio.create_task(self._owned(func, *args, **kwargs)) self.running_tasks.add(task) task.add_done_callback(self._task_done) @@ -1220,6 +1338,12 @@ async def _run_component( joins and persistent utility tasks.) """ + # Every task submitted under this call is this twin's, including the + # ones a component submits from inside its own coroutine. Stamped + # here as well as in `_owned` because a client's `get_inference` + # arrives on a request handler's context, not on one of ours. + _OWNER.set(self) + await self.is_start.wait() logger.info(f"Online run: {type(ant.component).__name__}.") @@ -1328,9 +1452,11 @@ async def _run_component( assert ant.model_select_task is not None logger.debug(f"Run {type(ant.component).__name__} selection task") - answer_ms = await ant.model_select_task( + selecting = ant.model_select_task( in_data, *ant.model_select_args, **ant.model_select_kwargs ) + note_flow_task(selecting) + answer_ms = await selecting # answer is an investigator id. if isinstance(answer_ms, tuple) and len(answer_ms) == 2: @@ -1393,7 +1519,13 @@ async def _infer( """ try: - return await ant.inference_task(in_data, **model_kwargs) + # the future first, so the task it stands for is recorded as this + # twin's before anyone can hear about it (`note_flow_task`); a + # plain coroutine has no uid and is simply skipped + pending = ant.inference_task(in_data, **model_kwargs) + note_flow_task(pending) + + return await pending except TypeError as exc: traceback = exc.__traceback__ @@ -1555,6 +1687,32 @@ def described(ant: _AnnotatedComponent) -> dict: }, } + def metrics(self) -> dict: + """Convergence metrics the graph's components report, by name. + + Duck-typed on purpose: a `StreamingLearnerInvestigator` refreshes a + filtered `metrics` dict once per learning window, and this collects + whatever component carries one -- so the runtime never has to know + about ROSE. A metric name two components both track is qualified + with the second one's class. + """ + + collected: dict = {} + + for ant in self._annotated(): + reported = getattr(ant.component, "metrics", None) + if not isinstance(reported, dict): + continue + + component = type(ant.component).__name__ + for name, entry in reported.items(): + if not isinstance(entry, dict): + continue + key = name if name not in collected else f"{component}.{name}" + collected[key] = {**entry, "component": component} + + return collected + def print_graph(self) -> str: """Human-readable rendering of `describe()`.""" diff --git a/src/digitaltwin/service/plugin.py b/src/digitaltwin/service/plugin.py index c5c31c7..f5cae00 100644 --- a/src/digitaltwin/service/plugin.py +++ b/src/digitaltwin/service/plugin.py @@ -29,12 +29,14 @@ import os import time +from pathlib import Path from typing import Any, Optional from fastapi import FastAPI from radical.orbit.errors import http_exception from radical.orbit.plugin_base import Plugin from starlette.requests import Request +from starlette.responses import Response from ..config import ( BACKEND_ORBIT, @@ -59,6 +61,22 @@ ROUTE_TWIN_CLOSE = "twin_close/{sid}/{twin_id}" ROUTE_TWIN_CALL = "twin_call/{sid}/{twin_id}" ROUTE_ADMIN_SESSIONS = "admin/sessions" +ROUTE_UI = "ui" +ROUTE_UI_ASSET = "ui/{asset}" + +# The dashboard. `dt_explorer.js` is the ORBIT Explorer's UI module (see +# `ui_module`); `index.html` is the standalone host, which the plugin +# serves so that a browser can reach a live broker *same-origin* -- the +# gateway's CORS allow-list and the `SameSite=Strict` auth cookie rule out +# every other origin. An allow-list, not a directory walk: `{asset}` is a +# client-supplied path segment. +UI_DIR = Path(__file__).parent / "ui" +UI_ASSETS = { + "index.html": "text/html; charset=utf-8", + "dt_dash.js": "application/javascript", + "dt_sample.js": "application/javascript", + "dt_explorer.js": "application/javascript", +} # how often the supervisor checks that the stream broker is still alive BROKER_WATCH_INTERVAL = 5.0 @@ -77,6 +95,7 @@ class PluginDT(Plugin): - POST `/dt/twin_close/{sid}/{twin_id}` -- stop and forget one twin - POST `/dt/twin_call/{sid}/{twin_id}` -- exactly one graph verb - GET `/dt/admin/sessions` -- every session, twin and error + - GET `/dt/ui`, `/dt/ui/{asset}` -- the live dashboard Every call is short except `get_inference`. No notifications, no request ids: `twin_list` polling is the observation mechanism. @@ -93,6 +112,14 @@ class PluginDT(Plugin): "description": "Host long-running digital twins (in-situ inference).", } + # The Explorer's per-plugin JS module, served by the gateway at + # `/plugins/dt.js`. Honoured for broker-hosted plugins only + # (`BrokerPluginHost.get_ui_modules` is its single reader), which is + # this plugin's default deployment; endpoint-hosted, the Explorer falls + # back on `ui_config` above and the dashboard is reached at + # `{namespace}/ui` instead. + ui_module = str(UI_DIR / "dt_explorer.js") + def __init__(self, app: FastAPI, instance_name: str = "dt"): super().__init__(app, instance_name) @@ -115,6 +142,16 @@ def __init__(self, app: FastAPI, instance_name: str = "dt"): self.add_route_post(ROUTE_TWIN_CLOSE, self.twin_close) self.add_route_post(ROUTE_TWIN_CALL, self.twin_call) self.add_route_get(ROUTE_ADMIN_SESSIONS, self.admin_sessions) + self.add_route_get(ROUTE_UI, self.ui_index) + self.add_route_get(ROUTE_UI_ASSET, self.ui_asset) + + # the page references its script relative to itself, and a browser + # at `{namespace}/ui` (no trailing slash) resolves that against the + # *parent* -- so the assets answer there as well, and both spellings + # of the page work + for asset in UI_ASSETS: + if asset != "index.html": + self.add_route_get(asset, self._root_asset(asset)) # -- session policy ----------------------------------------------------- @@ -277,6 +314,58 @@ async def admin_sessions(self, request: Request) -> dict: return {"sessions": sessions, "stream_broker": self.stream_summary()} + # -- the dashboard ------------------------------------------------------ + + async def ui_index(self, request: Request) -> Response: + """The standalone dashboard page, same-origin with the broker.""" + + return self._ui_asset("index.html") + + async def ui_asset(self, request: Request) -> Response: + """One dashboard asset, from the allow-list.""" + + return self._ui_asset(request.path_params["asset"]) + + def _root_asset(self, asset: str): + """The same assets, next to `ui` instead of under it (see routes). + + The name is bound per route: the broker-hosted dispatch hands a + request shim, so the handler must not introspect the request. + """ + + async def handler(request: Request) -> Response: + return self._ui_asset(asset) + + return handler + + @staticmethod + def _ui_asset(asset: str) -> Response: + """A `Response`, not a dict -- which every dispatch path handles. + + Checked, because it is the only route in this plugin that does not + return JSON: all three normalize on `status_code` and forward the + raw body. `Plugin._wrap_handler` for the ASGI/Explorer path, + `BrokerPluginHost.handle_request` for a broker-hosted call + (`Broker._dispatch_to_host` then packs `bytes(result.body)` into + the wire response), and `EndpointRuntime._dispatch_served` for an + endpoint-hosted one. + """ + + media = UI_ASSETS.get(asset) + if media is None: + raise http_exception(FileNotFoundError(f"no such asset: {asset}")) + + try: + body = (UI_DIR / asset).read_bytes() + except OSError as exc: + raise http_exception(FileNotFoundError(str(exc))) from exc + + # read per request rather than cached: these are a handful of KiB, + # asked for once per page load, and editing one should not need a + # broker restart (the gateway's own `ui_module` cache does) + return Response(body, media_type=media, + headers={"cache-control": "no-store"}) + # -- observability ------------------------------------------------------ async def on_topology_change(self, participants: dict) -> None: diff --git a/src/digitaltwin/service/session.py b/src/digitaltwin/service/session.py index bf29d4f..34e1c50 100644 --- a/src/digitaltwin/service/session.py +++ b/src/digitaltwin/service/session.py @@ -9,6 +9,7 @@ """ import asyncio +import inspect import contextlib import logging import time @@ -134,6 +135,12 @@ def __init__(self, twin_id: str, config: Optional[dict] = None): # explicitly or a closing twin would leave a caller hanging self._inflight: set[asyncio.Task] = set() + # verb -> how many of them this twin has served. The only record + # that a client ever called: the verbs are synchronous and leave + # nothing else behind, so without this an observer cannot tell a + # twin being driven from one merely sitting in `running`. + self.calls: dict[str, int] = {} + @property def state(self) -> str: return self._state if self.runtime is None else str(self.runtime.state) @@ -145,7 +152,13 @@ def last_error(self) -> Optional[str]: return self._last_error def summary(self) -> dict: - """The twin's entry in `twin_list` / `admin/sessions`.""" + """The twin's entry in `twin_list` / `admin/sessions`. + + `metrics` is the graph's convergence criteria (empty for a twin + with no learner in it): `twin_list` polling is the only + observation mechanism in v1, so anything an operator has to watch + rides here. + """ return { "twin_id": self.twin_id, @@ -153,6 +166,14 @@ def summary(self) -> dict: "last_error": self.last_error, "age": round(time.time() - self.created, 3), "config": self.config, + "metrics": {} if self.runtime is None else self.runtime.metrics(), + "calls": dict(self.calls), + # The uids this twin most recently submitted (`TASK_UID_RING`). + # A `task_status` notification carries a uid and an endpoint and + # nothing else, so this is what lets an observer say which twin a + # task belongs to; bounded, newest last, and a uid that has aged + # out is unattributed rather than wrong. + "tasks": [] if self.runtime is None else self.runtime.task_uids(), } def ready(self, runtime: DTRuntime, stream: PubSubClient) -> None: @@ -326,6 +347,11 @@ async def twin_call( detail=f"twin {twin_id}: {verb}: {type(exc).__name__}: {exc}", ) from exc + # counted on the way out, so what it records is a round trip a + # client really completed -- a call that failed or is still in + # flight has not been answered and is not one + twin.calls[verb] = twin.calls.get(verb, 0) + 1 + return {**self._twin_state(twin), **(extra or {})} def _decode_call(self, verb: str, payload: Optional[str], @@ -544,13 +570,25 @@ async def _create_engine(self, name: str) -> WorkflowEngine: cfg.get("endpoint_name") or "", ) - backend = await OrbitExecutionBackend( + kwargs: dict = dict( broker_url=self.broker_url, endpoint_name=cfg.get("endpoint_name"), backends=cfg.get("backends") or DEFAULT_BACKENDS, batch_window=0, # per-call latency beats batching for in-situ ) + # name the backend's broker participant after what it is for, so a + # topology view shows `rhapsody..` instead of an + # anonymous uuid. Unique by construction: one engine per role per + # session (`engine` caches, `_lost` forbids rebuilds). Guarded so + # a rhapsody without the parameter keeps working. + if "participant_name" in inspect.signature( + OrbitExecutionBackend.__init__).parameters: + kwargs["participant_name"] = ( + f"rhapsody.{self.sid.split('.')[-1]}.{name}") + + backend = await OrbitExecutionBackend(**kwargs) + # the endpoint the backend *settled on* (it auto-selects when the # config named none) -- what a topology change is matched against self._endpoints[name] = ( @@ -635,13 +673,26 @@ async def close(self) -> dict: return await super().close() def summary(self) -> dict: - """This session's entry in the `admin/sessions` listing.""" + """This session's entry in the `admin/sessions` listing. + + `endpoints` names the hardware behind each engine role -- the + endpoint the backend settled on, the configured one before that, + `None` for an engine that is not configured (`'exsitu'` then + aliases `'task'`, and an observer can say so). + """ return { "sid": self.sid, "active": self.is_active, "age": round(time.time() - self.created, 3), "engines": sorted(self._engines), + "endpoints": { + TASK_ENGINE: self._engine_endpoint(TASK_ENGINE), + EXSITU_ENGINE: ( + self._engine_endpoint(EXSITU_ENGINE) + if self.configured(EXSITU_ENGINE) else None + ), + }, "twins": [twin.summary() for twin in self.twins.values()], } diff --git a/src/digitaltwin/service/ui/dt_dash.js b/src/digitaltwin/service/ui/dt_dash.js new file mode 100644 index 0000000..a3a7109 --- /dev/null +++ b/src/digitaltwin/service/ui/dt_dash.js @@ -0,0 +1,2352 @@ +/* ========================================================================== + * dt_dash.js -- a live / replayable dashboard for the DTaaS `dt` plugin + * + * One implementation, two hosts: + * + * - standalone: `index.html` loads this file with a plain + + + + diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py index b453405..48f98f8 100644 --- a/src/digitaltwin/streaming.py +++ b/src/digitaltwin/streaming.py @@ -28,6 +28,7 @@ import json import logging import multiprocessing +import uuid from abc import ABC, abstractmethod from dataclasses import dataclass @@ -868,7 +869,16 @@ async def connect_backend(self, timeout: Optional[float] = None) -> PubSubBacken # extra, and a plain ZMQ install must not need it from .streaming_orbit import OrbitPubSubBackend - backend = OrbitPubSubBackend(self.broker_url) + # named after the twin it serves, so a topology view can match + # the participant to a dashboard card. A short random suffix + # keeps two clients on one namespace apart (the twin's own, + # plus any consumer which opened the twin's config). + name = None + if self.namespace: + name = (f"dt_stream.{self.namespace.split('-')[0]}" + f".{uuid.uuid4().hex[:4]}") + + backend = OrbitPubSubBackend(self.broker_url, name=name) elif self.kind == ZMQ_PS_Client.kind: backend = ZMQ_PS_Client(self.pub_addr, self.sub_addr) diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py index 58283f4..64a47aa 100644 --- a/src/digitaltwin/streaming_orbit.py +++ b/src/digitaltwin/streaming_orbit.py @@ -159,7 +159,10 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime: logger.info("connecting stream participant %s", self.name) - runtime = EndpointRuntime(broker_url=self.broker_url, name=self.name) + # role: the default 'consumer' says nothing in a topology view. + # This participant is a twin's data plane. + runtime = EndpointRuntime(broker_url=self.broker_url, name=self.name, + role="stream") try: runtime.start(wait=True, timeout=timeout) diff --git a/test/12-dtaas-live/RUNBOOK.md b/test/12-dtaas-live/RUNBOOK.md new file mode 100644 index 0000000..3b7520a --- /dev/null +++ b/test/12-dtaas-live/RUNBOOK.md @@ -0,0 +1,94 @@ +# DTaaS live demo — operator runbook + +Roles: radical.3 hosts the broker (+ `dt` plugin) and both rhapsody +endpoints; the laptop runs the client and the browser. Four terminals +on radical.3 are overkill — broker and endpoints can live in one tmux. + +## Once, before demo day + + # radical.3 + ~/.radical/orbit/ broker_cert.pem broker_key.pem (0600) broker.token + ./deploy/install.sh broker # venv: ./ve.demo + + # laptop (already true for ve3, listed for completeness) + ~/.radical/orbit/ broker_cert.pem broker.token + +The install script prints the version stamp — compare it across hosts +BEFORE demo day. The digitaltwin version reads `0.0.1` on every commit, +so only the pinned install guarantees the trees match. + +## Demo day, radical.3 (tmux, three panes) + + # pane 1 -- broker + dt plugin + ./deploy/run-broker.sh + + # pane 2 + 3 -- the two endpoints (order after the broker) + ./deploy/run-endpoint.sh dt_task_ep localhost + ./deploy/run-endpoint.sh dt_exsitu_ep localhost + +Sanity: pane 1 shows `registered as 'dt_task_ep'` and `'dt_exsitu_ep'`. + +## Demo day, laptop -- BEFORE the audience arrives + +1. Browser: open `https://95.217.193.116:8000/`, accept the self-signed + cert, enter the token. This mints the cookie the dashboard rides; + skipping it means a 401 at the worst possible moment. +2. Open `https://95.217.193.116:8000/broker/dt/ui?live=1` full-window + (without `live=1` the page plays the bundled sample recording -- + the `sample` toolbar button gets it back as the fallback) — the + standalone dashboard. Check the version badge next to the stream + pill reads 0.5.0. +3. Second tab, loaded and paused: the bundled recording + (`src/digitaltwin/service/ui/index.html`) — the fallback. One + keystroke away, never mentioned unless needed. +4. Terminal -- NOTE: the client venv (`./deploy/install.sh client`, + python 3.12 like the service; NOT the 3.13 dev venv `ve3`, which the + service would reject for version skew): + + . demo.sh # from the repo root + + (`demo.sh` is operator-local, untracked: activates ve.demo, cds into + test/12-dtaas-live, and exports the broker URL, the broker's pinned + cert and the two endpoint names.) + + ('radical.3' is only an ssh alias -- the client and browser use the IP. + The CERT is the BROKER's cert, fetched once via + `scp radical.3:.radical/orbit/broker_cert.pem + ~/.radical/orbit/broker_cert.radical3.pem`: + each host generated its own self-signed pair, and the client pins the + broker's, not its own.) + +## The show + + ./run_me.py # steps 1-8, Enter-paced; EXITS at step 8 + ./run_me.py --attach session.XXXXXXXX # steps 9-11 + +Step 8 prints the exact --attach line — leave the terminal visible so +the "client is gone, twins are not" beat lands, give it ~30s while the +dashboard keeps moving, then reattach. + +Talking anchors per step live in run_me.py itself; the two on-screen +proofs worth pointing at explicitly: + + - EchoSink line `served_by: dt_task_ep, trained_on: dt_exsitu_ep` + -- in-situ and ex-situ on different hardware, printed by the twin. + - The convergence bar on twin B's card -- ROSE's fit_error criterion, + updated per training window (~15s cadence). + +## If it goes sideways + + - Dashboard 401 -> the cookie step was skipped; do step 1 above. + - Client dies with "TLS verification failed ... self-signed + certificate" -> the pinned cert is the laptop's own, not the + broker's; re-fetch it (see the scp line above). + - Twin fails with "No module named 'opentelemetry'" -> the endpoint + venv predates the rhapsody[telemetry] pin; `pip install + 'opentelemetry-sdk>=1.20.0' nvidia-ml-py` into it (no restart + needed -- the next session init retries the import). + - Twin stuck `initializing`, error mentions "Broker URL required" + -> broker was started without RADICAL_ORBIT_BROKER_URL; restart + pane 1 via run-broker.sh (it sets it). + - No pulses in the sensors lane, tiles present -> the venv predates + the orbit SSE fix (radical.orbit#113); rerun `deploy/install.sh + broker` (fresh venv, pinned deps) and restart the broker. + - Anything else -> tab 2, resume the recording, keep narrating. diff --git a/test/12-dtaas-live/components.py b/test/12-dtaas-live/components.py new file mode 100644 index 0000000..f716961 --- /dev/null +++ b/test/12-dtaas-live/components.py @@ -0,0 +1,79 @@ +"""The plain twin: sensor -> model -> sink. + +Paced for narration rather than for throughput. Everything here is +cloudpickled by value and instantiated inside the plugin host, so this +module must stay importable on its own. +""" + +import asyncio + +from digitaltwin.components import ModelInvestigator, TypedData, UtilityTask + +from dtypes import INFERENCE_DTYPE, SENSOR_DTYPE + +# slow enough to talk over, long enough to outlast the whole narration +TICK = 2.5 +READINGS = 400 + + +class PacedSensor(UtilityTask): + """A persistent component, running inline on the service loop. + + It publishes through `runtime.stream` -- the twin's own client, which + the runtime injected. Not `stream_config`: that is for code running + somewhere else, and opening a second client from in here would leak + one per twin inside a broker that runs for days. + """ + + async def main_loop(self, runtime, in_data): + for value in range(READINGS): + await runtime.stream.publish(SENSOR_DTYPE, value) + await asyncio.sleep(TICK) + + +class RampModel(ModelInvestigator): + """No learning -- inference whose answer moves as the model is + republished, so the sink's output visibly changes mid-demo. + + `compute` runs on the rhapsody endpoint; the `TypedData` wrapping + happens here, in the service. Task return values have to be + JSON-safe or bytes to survive ORBIT's rhapsody plugin, so the task + returns a plain number. + """ + + def __init__(self, flow, *args, **kwargs): + super().__init__(flow) + self.flow = flow + + @self.flow.function_task + async def compute(in_data: TypedData, gain=1): + return gain * in_data.data + + self.compute = compute + + async def main_loop(self, runtime): + async def do_inference(in_data: TypedData, gain=1): + answer = await self.compute(in_data, gain=gain) + return TypedData(INFERENCE_DTYPE, answer) + + runtime.set_inference_task(do_inference) + + gain = 2 + while True: + runtime.publish_new_model({"gain": gain}) + gain += 1 + await asyncio.sleep(20.0) + + +class EchoSink(UtilityTask): + """Prints, and *publishes*. + + The runtime hands a component's answer to the next component on that + dtype over an in-process queue and drops it if nobody is registered. + Anything outside the service -- the dashboard included -- only sees a + result because a component chose to publish it. + """ + + async def main_loop(self, runtime, in_data): + print(f" [twin] inference -> {in_data.data}", flush=True) + await runtime.stream.publish(INFERENCE_DTYPE, in_data.data) diff --git a/test/12-dtaas-live/dtypes.py b/test/12-dtaas-live/dtypes.py new file mode 100644 index 0000000..b018d47 --- /dev/null +++ b/test/12-dtaas-live/dtypes.py @@ -0,0 +1,15 @@ +"""The dtypes this demo's twins are wired with. + +Shipped to the service by value along with the components -- the service +has no copy of this directory. +""" + +from digitaltwin.components import DataType + +SENSOR_DTYPE = DataType("sensor") +INFERENCE_DTYPE = DataType("inference") + +# What a twin's own results are published under, so the dashboard's +# sensors lane shows answers as well as readings. The runtime never +# publishes a component's return value by itself -- a component has to. +RESULT_CHANNEL = "results" diff --git a/test/12-dtaas-live/learner.py b/test/12-dtaas-live/learner.py new file mode 100644 index 0000000..65b41f1 --- /dev/null +++ b/test/12-dtaas-live/learner.py @@ -0,0 +1,96 @@ +"""The second twin: a ROSE streaming learner, retraining ex-situ. + +Two engines, two endpoints. The learner's training windows go to the +`exsitu` engine; the inference it serves runs on `task`. Both halves +report which endpoint they ran on, so the claim is checkable on screen +rather than asserted in prose. + +Paced so the convergence bar in the dashboard moves during the +narration instead of snapping to converged on the first window. +""" + +import os + +from digitaltwin.components import DataType, TypedData +from digitaltwin.learn import StreamingLearnerInvestigator + +from dtypes import INFERENCE_DTYPE, SENSOR_DTYPE + +# the calibration the learner has to recover from the sensor stream +SLOPE = 10.0 + +# a window every BATCH_SIZE readings; at the sensor's 2.5s tick that is +# one training round roughly every 15s -- slow enough to point at +BATCH_SIZE = 6 +MAX_WAIT = 30.0 + + +def _tag() -> str: + """Which endpoint is this task running on? `os.environ` is the only + channel a cloudpickled function body has for finding out.""" + + return os.environ.get("DT_ENDPOINT_TAG", "?") + + +class DriftingLearner(StreamingLearnerInvestigator): + def __init__(self, flow, learn_flow=None): + super().__init__(flow, learn_flow, batch_size=BATCH_SIZE, + max_wait=MAX_WAIT) + + # the criterion task takes no dependency, so the model it scores + # has to travel with it: this mirror is filled service-side from + # each training result and cloudpickled on every submission + latest: dict = {} + self.learner.on_state_update(latest.__setitem__) + + # -- ex-situ, on `learn_flow` --------------------------------------- + + @self.learner.training_task(as_executable=False) + async def training(window, *args): + # labelling the window stands in for the simulation a real + # ex-situ learner would run out here + xs = [float(x) for x in window] + ys = [SLOPE * x for x in xs] + den = sum(x * x for x in xs) or 1.0 + + return { + "slope": sum(x * y for x, y in zip(xs, ys)) / den, + "trained_on": _tag(), + } + + @self.learner.active_learn_task(as_executable=False) + async def active_learn(model, *args): + return len(model) + + @self.learner.as_stop_criterion( + metric_name="fit_error", + threshold=1e-6, + operator="<", + as_executable=False, + ) + async def criterion(*args, model=latest): + error = model.get("slope", 0.0) - SLOPE + + return error * error + + # -- in-situ, on `flow` --------------------------------------------- + + @flow.function_task + async def predict(in_data: TypedData, slope=0.0, trained_on=""): + return { + "value": slope * in_data.data, + "served_by": _tag(), + "trained_on": trained_on, + } + + async def infer(in_data: TypedData, slope=0.0, trained_on=""): + return TypedData(INFERENCE_DTYPE, + await predict(in_data, slope=slope, + trained_on=trained_on)) + + self.inference_task = infer + + def bootstrap_model(self) -> tuple: + """Nothing learned yet: every reading predicts zero.""" + + return {"slope": 0.0}, {} diff --git a/test/12-dtaas-live/run_me.py b/test/12-dtaas-live/run_me.py new file mode 100755 index 0000000..b7d2d53 --- /dev/null +++ b/test/12-dtaas-live/run_me.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python3 +"""The live DTaaS demo, paced for narration. + +Two phases, because the point of phase two is that phase one's process is +gone: + + python run_me.py # build and start two twins, then exit + python run_me.py --attach # come back to them, then tear down + +Each step waits for Enter so the pacing is yours. Set DEMO_STEP to a +number of seconds to make it run itself instead. + +Environment: + + RADICAL_ORBIT_BROKER_URL wss://radical.3:8000 + DT_SERVICE_HOST participant hosting the `dt` plugin (default 'broker') + DT_TASK_ENDPOINT endpoint for in-situ compute + DT_EXSITU_ENDPOINT endpoint for the learner's training windows +""" + +import argparse +import json +import logging +import os +import sys +import textwrap +import time + +from radical.orbit import EndpointRuntime + +from digitaltwin.components import NULL_DTYPE, TRUTHY, TypedData +from digitaltwin.service import register_user_modules + +from dtypes import INFERENCE_DTYPE, SENSOR_DTYPE +from components import EchoSink, PacedSensor, RampModel +from learner import DriftingLearner + +# the service has no copy of this directory -- ship it by value +import components +import dtypes +import learner + +register_user_modules([dtypes, components, learner]) + +DT_HOST = os.environ.get("DT_SERVICE_HOST", "broker") +TASK_EP = os.environ.get("DT_TASK_ENDPOINT") or None +EXSITU_EP = os.environ.get("DT_EXSITU_ENDPOINT") or TASK_EP + +# Two engines, named by role and pinned to hardware. 'exsitu' is what +# makes the learner's training a separate lane in the dashboard rather +# than an alias of 'task'. +ENGINES = { + "engines": { + "task": {"endpoint_name": TASK_EP, "backends": ["concurrent"]}, + "exsitu": {"endpoint_name": EXSITU_EP, "backends": ["concurrent"]}, + } +} + +PACE = os.environ.get("DEMO_STEP", "manual") + +# syntax highlighting is a soft dependency: without pygments (or with +# NO_COLOR set) the api snippets print plain +try: + from pygments import highlight + from pygments.formatters import Terminal256Formatter + from pygments.lexers import PythonLexer + _LEXER = PythonLexer() + _FORMATTER = Terminal256Formatter(style="monokai") +except ImportError: + _LEXER = None + + +def _render_code(code: str) -> None: + """The step's API surface, as a block the audience can read.""" + + text = textwrap.dedent(code).strip("\n") + if _LEXER is not None and not os.environ.get("NO_COLOR"): + text = highlight(text, _LEXER, _FORMATTER).rstrip("\n") + + print(" \033[2mapi:\033[0m") + for line in text.splitlines(): + print(f" {line}") + print() + + +def step(title: str, note: str = "", code: str = "") -> None: + """Announce the next beat and hold until the narrator is ready.""" + + print(f"\n\033[1m{'=' * 70}\n{title}\033[0m") + if note: + print(f"{note}\n") + if code: + _render_code(code) + + if PACE == "manual": + try: + input(" [Enter] ") + except EOFError: + pass + else: + time.sleep(float(PACE)) + + +def show(label: str, obj) -> None: + print(f" {label}:") + for line in json.dumps(obj, indent=2).splitlines(): + print(f" {line}") + + +def build(dt): + """Phase one: two twins on one session, then walk away.""" + + step("1. Session", + f" - sid: {dt.sid}\n" + " - the sid is a bearer capability: it is the only client state\n" + " - twins belong to the session, not to this process", + code=""" + ENGINES = {'engines': {'task': {'endpoint_name': 'dt_task_ep'}, + 'exsitu': {'endpoint_name': 'dt_exsitu_ep'}}} + + runtime = EndpointRuntime() # the ORBIT client runtime + dt = runtime.get_plugin('broker', 'dt', config=ENGINES) + """) + + # -- twin A: plain in-situ inference ----------------------------------- + + step("2. Create a twin", + " - create_twin returns on registration; the helper polls to ready\n" + " - dashboard: card in the broker lane, initializing -> ready", + code=""" + twin = dt.create_twin() + """) + twin_a = dt.create_twin() + print(f" twin A: {twin_a}") + + step("3. Ship the graph", + " - the service has none of this code\n" + " - component classes go over the wire (cloudpickle, by value)\n" + " - the service instantiates them, injecting the session engine\n" + " - graph: sensor -> model -> sink", + code=""" + dt.add_task (twin, dt.package(PacedSensor), + TRUTHY, SENSOR_DTYPE, is_persistent=True) + dt.add_investigator(twin, dt.package(RampModel), + SENSOR_DTYPE, INFERENCE_DTYPE) + dt.add_task (twin, dt.package(EchoSink), + INFERENCE_DTYPE, NULL_DTYPE) + dt.describe(twin) + """) + dt.add_task(twin_a, dt.package(PacedSensor), TRUTHY, SENSOR_DTYPE, + is_persistent=True) + dt.add_investigator(twin_a, dt.package(RampModel), SENSOR_DTYPE, + INFERENCE_DTYPE) + dt.add_task(twin_a, dt.package(EchoSink), INFERENCE_DTYPE, NULL_DTYPE) + show("graph", dt.describe(twin_a)) + + step("4. Start", + " - sensor publishes a reading every 2.5s\n" + " - inference runs on the rhapsody endpoint, not in the broker\n" + " - dashboard: sensor tile, client -> broker arcs, task tiles on\n" + " the HPC lane", + code=""" + dt.start(twin) + """) + dt.start(twin_a) + + # -- twin B: the dual-engine learner ----------------------------------- + + step("5. Second twin: ex-situ learning", + " - same session, same stream shape\n" + " - retrains on input windows, on a second engine / endpoint\n" + " - inference serves from the task endpoint while training runs\n" + " - dashboard: convergence bar = ROSE stop criterion (fit_error\n" + " vs threshold, updated per window)", + code=""" + dt.add_investigator(twin_b, dt.package(DriftingLearner), + SENSOR_DTYPE, INFERENCE_DTYPE) + + # inside DriftingLearner (ROSE): + @learner.training_task + async def training(window): ... + + @learner.as_stop_criterion(metric_name='fit_error', + threshold=1e-6, operator='<') + async def criterion(): ... + """) + twin_b = dt.create_twin() + dt.add_task(twin_b, dt.package(PacedSensor), TRUTHY, SENSOR_DTYPE, + is_persistent=True) + dt.add_investigator(twin_b, dt.package(DriftingLearner), SENSOR_DTYPE, + INFERENCE_DTYPE) + dt.add_task(twin_b, dt.package(EchoSink), INFERENCE_DTYPE, NULL_DTYPE) + dt.start(twin_b) + print(f" twin B: {twin_b}") + + # -- the client asks directly ------------------------------------------ + + step("6. Query twin A", + " - get_inference is the one call that answers the caller\n" + " - all other results flow component to component in the twin\n" + " - 10 calls, 2s apart; dashboard: one arc per call\n" + " - meanwhile twin B trains: exsitu tiles, convergence bar\n" + " per window (~15s)", + code=""" + for value in range(10): + answer = dt.get_inference(twin, TypedData(SENSOR_DTYPE, value), + INFERENCE_DTYPE) + """) + for value in range(10): + answer = dt.get_inference(twin_a, TypedData(SENSOR_DTYPE, value), + INFERENCE_DTYPE) + print(f" {value} -> {answer.data}", flush=True) + time.sleep(2) + + + + step("7. Operator view", + " - admin_sessions: owner, age, twins, states, last errors\n" + " - names the endpoint behind each engine role\n" + " - this is how orphaned sessions are found", + code=""" + dt.admin_sessions() + """) + show("sessions", dt.admin_sessions()) + + step("8. Client exits", + " - this process ends; the twins do not\n" + " - no timeout: twins run for days, clients come and go\n" + " - the dashboard keeps updating", + code=""" + # no teardown call + sys.exit(0) + """) + + print("\n reattach with:\n") + print(f" python run_me.py --attach {dt.sid}\n") + + return dt.sid + + +def attach(dt): + """Phase two: the client is a different process now.""" + + step("9. Reattach", + " - a new process; its only input is the sid", + code=""" + dt = runtime.get_plugin('broker', 'dt', sid=sid) + dt.twin_list() + """) + for entry in dt.twin_list(): + print(f" {entry['twin_id'][:8]} {entry['state']:<10}" + f" metrics={list((entry.get('metrics') or {}).keys())}") + + step("10. Close the twins", + " - twin_close, per twin; same route an operator uses", + code=""" + for twin in dt.twin_list(): + dt.twin_close(twin['twin_id']) + """) + for entry in dt.twin_list(): + print(f" closing {entry['twin_id'][:8]}") + dt.twin_close(entry["twin_id"]) + + show("sessions", dt.admin_sessions()) + + step("11. Close the session", + " - sessions are persistent: twins gone != session gone\n" + " - unregister_session shuts the engines down\n" + " - explorer: the rhapsody.. participants leave", + code=""" + dt.unregister_session() + """) + sid = dt.sid + dt.unregister_session() + print(f" session {sid} unregistered; the service holds no state of ours") + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--attach", metavar="SID", default=None, + help="reattach to an existing session") + args = parser.parse_args() + + logging.basicConfig(level=logging.WARNING) + + runtime = EndpointRuntime() + runtime.start(wait=True) + + try: + kwargs = {"sid": args.attach} if args.attach else {"config": ENGINES} + dt = runtime.get_plugin(DT_HOST, "dt", **kwargs) + + if args.attach: + attach(dt) + else: + build(dt) + + finally: + runtime.stop() + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test/unit/test_learn.py b/test/unit/test_learn.py index a79fa71..3f422f4 100644 --- a/test/unit/test_learn.py +++ b/test/unit/test_learn.py @@ -8,6 +8,7 @@ import logging from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace import pytest @@ -24,7 +25,11 @@ TypedData, UtilityTask, ) -from digitaltwin.learn import StreamingLearnerInvestigator # noqa: E402 +from digitaltwin.learn import ( # noqa: E402 + METRIC_HISTORY, + StreamingLearnerInvestigator, + _number, +) X = DataType("x") Y = DataType("y") @@ -170,6 +175,111 @@ async def test_the_learner_uses_the_engine_it_was_given(flow, engines): assert learner.flow is flow +async def test_the_criterion_state_shows_up_in_the_twins_metrics( + flow, stream_clients): + """Per window, the learner mirrors its criterion into `metrics`, and + the runtime collects it for `twin_list` -- the only observation + mechanism v1 has, so convergence has to ride on it.""" + + runtime, learner = await _twin(flow, await stream_clients("twin-metrics")) + + try: + await _await_learned(runtime) + + assert learner.windows >= 1 + metric = learner.metrics["fit_error"] + + assert metric["threshold"] == 1e-6 + assert metric["operator"] == "<" + assert metric["should_stop"] is True + assert metric["windows"] == learner.windows + assert metric["history"][-1] == metric["value"] + # filtered: the model itself never travels in a metric + assert set(metric) == {"value", "threshold", "operator", + "should_stop", "windows", "history"} + + collected = runtime.metrics() + assert collected["fit_error"]["component"] == "LinearLearner" + assert collected["fit_error"]["value"] == metric["value"] + + finally: + await runtime.stop() + + +@pytest.mark.parametrize("value, expected", [ + # significant figures, not decimal places: a 1e-8 criterion threshold + # is an ordinary target, and decimal rounding would wire it as 0.0 + (1e-8, 1e-8), + (2.5e-11, 2.5e-11), + (0.12345678901, 0.123457), + (1234567.0, 1234570.0), + (0.0, 0.0), + (7, 7.0), + # not numbers, or not JSON-safe + (True, None), + (False, None), + (None, None), + ("0.5", None), + (float("inf"), None), + (float("nan"), None), +]) +def test_a_metric_number_keeps_its_magnitude(value, expected): + assert _number(value) == expected + + +def test_a_tiny_threshold_survives_the_wire(flow): + """The whole point of `metrics`: what the dashboard draws as the target + tick has to be the target the learner is comparing against.""" + + class Tiny(LinearLearner): + def __init__(self, flow, learn_flow=None): + super().__init__(flow, learn_flow) + + @self.learner.as_stop_criterion( + metric_name="fit_error", threshold=1e-8, operator="<", + as_executable=False) + async def criterion(*args): + return 3e-9 + + learner = Tiny(flow) + learner._record_metrics(SimpleNamespace( + iteration=0, metric_name="fit_error", metric_value=3.14159e-9, + metric_threshold=1e-8, metric_history=[2e-8, 3.14159e-9], + should_stop=True)) + + metric = learner.metrics["fit_error"] + + assert metric["threshold"] == 1e-8 + assert metric["value"] == 3.14159e-9 + assert metric["history"] == [2e-8, 3.14159e-9] + + +def test_the_metric_history_is_bounded(flow): + """A days-long twin accumulates one value per window forever; only the + tail the sparkline draws travels.""" + + learner = LinearLearner(flow) + learner._record_metrics(SimpleNamespace( + iteration=99, metric_name="fit_error", metric_value=1.0, + metric_threshold=1.0, metric_history=list(range(1, 200)), + should_stop=False)) + + history = learner.metrics["fit_error"]["history"] + + assert len(history) == METRIC_HISTORY + assert history[-1] == 199.0 + + +async def test_a_twin_without_a_learner_reports_no_metrics(flow, + stream_clients): + runtime = DTRuntime(flow, await stream_clients("twin-nometrics")) + runtime.add_task(Counter(flow), TRUTHY, X, is_persistent=True) + + assert runtime.metrics() == {} + + await runtime.stop() + + async def test_an_absent_exsitu_engine_falls_back_to_the_twins(flow): learner = LinearLearner(flow) diff --git a/test/unit/test_service.py b/test/unit/test_service.py index 9487d43..c82364a 100644 --- a/test/unit/test_service.py +++ b/test/unit/test_service.py @@ -16,9 +16,9 @@ from fastapi import FastAPI, HTTPException # noqa: E402 from starlette.testclient import TestClient # noqa: E402 -from digitaltwin.components import UtilityTask # noqa: E402 +from digitaltwin.components import TRUTHY, DataType, UtilityTask # noqa: E402 from digitaltwin.runtime import DTRuntime # noqa: E402 -from digitaltwin.service.plugin import PluginDT # noqa: E402 +from digitaltwin.service.plugin import UI_ASSETS, PluginDT # noqa: E402 from digitaltwin.service.session import DTSession, TwinInstance # noqa: E402 from digitaltwin.service.wire import ( # noqa: E402 MAX_PAYLOAD, @@ -642,3 +642,141 @@ async def test_shutdown_stops_the_broker_and_the_supervisor(plugin): assert not broker.is_alive() assert supervisor.done() assert plugin._stream_broker is None + + +# --------------------------------------------------------------------------- +# the observation surface the dashboard reads +# --------------------------------------------------------------------------- + +class _Metered(UtilityTask): + """A component reporting a convergence metric, the way a + `StreamingLearnerInvestigator` does -- duck-typed, so this needs no + ROSE.""" + + metrics = {"rmse": {"value": 0.4, "threshold": 0.25, "operator": "<", + "should_stop": False, "windows": 3, + "history": [0.9, 0.6, 0.4]}} + + +async def test_a_twin_summary_carries_its_metrics(): + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.add_task(_Metered(_FakeFlow()), TRUTHY, DataType("x")) + + summary = twin.summary() + + assert summary["metrics"]["rmse"]["value"] == 0.4 + assert summary["metrics"]["rmse"]["component"] == "_Metered" + + await twin.close() + # a closed twin has no graph left to ask + assert twin.summary()["metrics"] == {} + + +async def test_a_twin_counts_the_verbs_it_answered(): + """The only trace a synchronous verb leaves: what the dashboard's + client-ward arcs are inferred from.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + + assert twin.summary()["calls"] == {} + + for _ in range(3): + await session.twin_call("t1", "describe", stamp=version_stamp()) + + assert twin.summary()["calls"] == {"describe": 3} + + await twin.close() + + +async def test_a_verb_that_failed_is_not_counted(): + """A round trip that was never answered is not one.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + await twin.runtime.stop() + + with pytest.raises(HTTPException) as raised: + await session.twin_call("t1", "start", stamp=version_stamp()) + + assert raised.value.status_code == 409 + assert "start" not in twin.summary()["calls"] + + await twin.close() + + +async def test_the_session_summary_names_the_engine_endpoints(): + """The dashboard draws one lane per engine *role*; `None` for + `'exsitu'` is the documented alias of `'task'`, not an omission.""" + + single = DTSession("s1", _dual(task="ep1")) + assert single.summary()["endpoints"] == {"task": "ep1", "exsitu": None} + + dual = DTSession("s2", _dual(task="ep1", exsitu="hpc1")) + assert dual.summary()["endpoints"] == {"task": "ep1", "exsitu": "hpc1"} + + +def test_admin_sessions_carries_endpoints_and_metrics(client): + sid = client.post("/dt/register_session", + json={"config": _dual(task="ep1")}).json()["sid"] + entry = next(s for s in client.get("/dt/admin/sessions").json()["sessions"] + if s["sid"] == sid) + + assert entry["endpoints"] == {"task": "ep1", "exsitu": None} + + +# --------------------------------------------------------------------------- +# the dashboard's assets +# --------------------------------------------------------------------------- + +def test_the_ui_route_serves_the_standalone_page(client): + """Served by the plugin so a browser can reach a live broker + same-origin: the gateway's CORS allow-list and the SameSite=Strict + auth cookie rule out every other origin.""" + + resp = client.get("/dt/ui") + + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("text/html") + assert "dt_dash.js" in resp.text + + +@pytest.mark.parametrize("asset", ["dt_dash.js", "dt_sample.js", + "dt_explorer.js"]) +def test_the_ui_assets_are_served_as_javascript(client, asset): + resp = client.get(f"/dt/ui/{asset}") + + assert resp.status_code == 200 + assert "javascript" in resp.headers["content-type"] + + +@pytest.mark.parametrize("asset", [ + "passwd", "plugin.py", ".env", + # a percent-encoded separator survives the route's `[^/]+` segment, so + # the allow-list is what has to refuse it -- not the router + "..%2f..%2fplugin.py", "%2e%2e%2f%2e%2e%2fplugin.py", + "dt_dash.js%00.png", +]) +def test_an_unlisted_ui_asset_is_404(client, asset): + """An allow-list, not a directory walk: the asset name comes from the + client.""" + + resp = client.get(f"/dt/ui/{asset}") + + assert resp.status_code == 404 + assert "plugin" not in resp.text or "no such asset" in resp.text + + +def test_the_explorer_module_is_the_one_the_plugin_declares(): + """ORBIT reads `ui_module` off the class and serves its content at + `/plugins/dt.js` -- so the path has to exist in the installed + package.""" + + from pathlib import Path + + module = Path(PluginDT.ui_module) + + assert module.is_file() + assert module.name in UI_ASSETS + assert "window.DTDash.mount" in module.read_text() diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py index 1781bd0..f385a74 100644 --- a/test/unit/test_streaming_orbit.py +++ b/test/unit/test_streaming_orbit.py @@ -154,7 +154,7 @@ async def orbit_backends(loopback, monkeypatch): monkeypatch.setattr( "digitaltwin.streaming_orbit.EndpointRuntime", - lambda broker_url=None, name=None: loopback.runtime(name), + lambda broker_url=None, name=None, **kw: loopback.runtime(name), ) backends = [] @@ -464,7 +464,7 @@ async def test_a_runtime_which_cannot_register_is_not_left_behind( made = [] - def build(broker_url=None, name=None): + def build(broker_url=None, name=None, **kw): runtime = loopback.runtime(name, registers=False) made.append(runtime) return runtime diff --git a/test/unit/test_task_owners.py b/test/unit/test_task_owners.py new file mode 100644 index 0000000..9844f0c --- /dev/null +++ b/test/unit/test_task_owners.py @@ -0,0 +1,228 @@ +"""Task ownership: which twin submitted the task a notification names. + +A `task_status` notification carries a uid and an endpoint and nothing else, +so the service records what it submitted (`DTRuntime.note_task`) and +`twin_list` carries it. The uid is asyncflow's own -- the execution backend +keeps the one in the component description -- so the join is exact rather +than inferred. + +Two paths reach the ring, and both are tested here against a real (local, +thread-backed) engine: the runtime's own submissions, and ROSE's. +""" + +import asyncio + +from concurrent.futures import ThreadPoolExecutor + +import pytest + +pytest.importorskip("rose") + +from radical.asyncflow import WorkflowEngine # noqa: E402 +from rhapsody.backends import ConcurrentExecutionBackend # noqa: E402 + +from digitaltwin import ( # noqa: E402 + TRUTHY, + DTRuntime, + DataType, + ModelInvestigator, + TypedData, + UtilityTask, +) +from digitaltwin.learn import StreamingLearnerInvestigator # noqa: E402 +from digitaltwin.runtime import TASK_UID_RING # noqa: E402 + +X = DataType("x") +Y = DataType("y") + + +@pytest.fixture +async def engines(): + made = [] + + async def make(): + backend = await ConcurrentExecutionBackend(ThreadPoolExecutor()) + made.append(await WorkflowEngine.create(backend=backend)) + return made[-1] + + try: + yield make + finally: + for engine in made: + await engine.shutdown() + + +class Direct(ModelInvestigator): + """The inference task *is* the flow task.""" + + def __init__(self, flow): + super().__init__(flow) + + @flow.function_task + async def infer(in_data, k=1.0): + return TypedData(Y, k * in_data.data) + + self._infer = infer + + async def main_loop(self, runtime): + runtime.set_inference_task(self._infer) + runtime.publish_new_model({"k": 2.0}, {}) + + +class Wrapped(ModelInvestigator): + """The inference task is a plain coroutine that awaits a flow task. + + The shape every real learner has, and the one the runtime cannot see: + what it awaits is a coroutine, and the future with the uid on it never + passes through the runtime at all. + """ + + def __init__(self, flow): + super().__init__(flow) + + @flow.function_task + async def predict(in_data, k=1.0): + return k * in_data.data + + async def infer(in_data, k=1.0): + return TypedData(Y, await predict(in_data, k=k)) + + self._infer = infer + + async def main_loop(self, runtime): + runtime.set_inference_task(self._infer) + runtime.publish_new_model({"k": 3.0}, {}) + + +class Learner(StreamingLearnerInvestigator): + """Windows of one, so a single item drives a whole ROSE iteration.""" + + def __init__(self, flow, learn_flow=None): + super().__init__(flow, learn_flow, batch_size=1, max_wait=1.0) + + @self.learner.training_task(as_executable=False) + async def training(window, *args): + return {"k": 2.0} + + @self.learner.active_learn_task(as_executable=False) + async def active_learn(model, *args): + return model + + @self.learner.as_stop_criterion( + metric_name="err", threshold=1e-9, operator="<", + as_executable=False) + async def criterion(*args): + return 1.0 + + @flow.function_task + async def predict(in_data, k=0.0): + return k * in_data.data + + async def infer(in_data, k=0.0): + return TypedData(Y, await predict(in_data, k=k)) + + self.inference_task = infer + + def bootstrap_model(self): + return {"k": 0.0}, {} + + +class Feeder(UtilityTask): + """One item per tick, into the twin's stream.""" + + async def main_loop(self, runtime, in_data): + for value in range(1000): + await runtime.stream.publish(X, float(value)) + await asyncio.sleep(0.1) + + +# --------------------------------------------------------------------------- +# the runtime's own submissions +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("component", [Direct, Wrapped]) +async def test_an_inference_task_is_recorded_against_its_twin( + component, engines, stream_clients): + """One inference, one uid -- whichever shape the inference task has.""" + + flow = await engines() + runtime = DTRuntime(flow, await stream_clients(f"own-{component.__name__}")) + runtime.add_investigator(component(flow), X, Y) + runtime.start() + + assert runtime.task_uids() == [] + + answer = await runtime.get_inference(TypedData(X, 21.0), Y) + await asyncio.sleep(0.1) + + assert answer.data + uids = runtime.task_uids() + assert len(uids) == 1, uids + assert uids[0].startswith("task."), uids + + # and a second call is a second task, in order + await runtime.get_inference(TypedData(X, 1.0), Y) + await asyncio.sleep(0.1) + assert runtime.task_uids()[:1] == uids + assert len(runtime.task_uids()) == 2 + + await runtime.stop() + + +async def test_the_ring_is_bounded_and_keeps_the_newest(engines, + stream_clients): + """A twin that submits forever must not remember forever.""" + + flow = await engines() + runtime = DTRuntime(flow, await stream_clients("own-bounded")) + runtime.start() + + for i in range(TASK_UID_RING + 5): + runtime.note_task(f"task.{i:06d}") + + uids = runtime.task_uids() + + assert len(uids) == TASK_UID_RING + assert uids[-1] == f"task.{TASK_UID_RING + 4:06d}" + assert uids[0] == f"task.{5:06d}" + # a uid it already has is not a new submission + runtime.note_task(uids[-1]) + assert runtime.task_uids() == uids + + await runtime.stop() + + +# --------------------------------------------------------------------------- +# ROSE's submissions +# --------------------------------------------------------------------------- + +async def test_the_learners_own_tasks_are_recorded_too(engines, + stream_clients): + """Training / active learning / criterion run on the ex-situ engine and + never pass through the runtime; the wrapper in `main_loop` is what makes + them the twin's (`_own_learner_tasks`).""" + + flow = await engines() + learn_flow = await engines() + + runtime = DTRuntime(flow, await stream_clients("own-learner")) + learner = Learner(flow, learn_flow) + + runtime.add_task(Feeder(flow), TRUTHY, X, is_persistent=True) + runtime.add_investigator(learner, X, Y) + runtime.start() + + # the learner is wrapped as soon as its main loop runs + deadline = asyncio.get_running_loop().time() + 30.0 + while len(runtime.task_uids()) < 3: + if asyncio.get_running_loop().time() > deadline: + pytest.fail(f"only {runtime.task_uids()} recorded" + f" ({runtime.state} {runtime.last_error})") + await asyncio.sleep(0.25) + + assert getattr(learner.learner, "_dt_owned", False) + assert all(uid.startswith("task.") for uid in runtime.task_uids()) + # every window is three ex-situ tasks, so a handful arrive quickly + assert len(set(runtime.task_uids())) == len(runtime.task_uids()) + + await runtime.stop() diff --git a/test/unit/test_ui_recording.py b/test/unit/test_ui_recording.py new file mode 100644 index 0000000..96d26d2 --- /dev/null +++ b/test/unit/test_ui_recording.py @@ -0,0 +1,335 @@ +"""The bundled dashboard recording, against the schema its reader expects. + +`dt_dash.js` documents the recording format and consumes it; this is the +other half of that contract -- a Python check that the sample shipped in +the package still parses, still carries both frame kinds, and still uses +the field names the renderer reads. It is what keeps a change on one side +from silently breaking the other. + +The sample is a `.js` file assigning one JSON object, not a `.json` one: +a classic `