diff --git a/.gitignore b/.gitignore index 0e48c0e..d2d421a 100644 --- a/.gitignore +++ b/.gitignore @@ -223,3 +223,7 @@ __marimo__/ # Streamlit .streamlit/secrets.toml + +# runtime residue from asyncflow / rhapsody test runs +asyncflow.session.*/ +telemetry-output/ diff --git a/README.md b/README.md index 5f51efc..9ffdc1f 100644 --- a/README.md +++ b/README.md @@ -23,11 +23,136 @@ Not yet implemented: - - Docs - - Type Annotation -## Running tests: +## Running the unit tests: -1. `pip install -e .` -2. `cd tests/` -3. In one terminal, run `local_broker.py` -4. In second terminal, cd into the test and run `run_me.py` +1. `pip install .[test,service]` +2. `pytest` (or `tox` for all supported interpreters) -**When running tests: be sure to start the ZMQ PubSub broker!** +The unit tests start their own stream broker on a random port; no setup. +The integration tests under `test/integration` bring up a real ORBIT +broker and rhapsody endpoint and skip themselves when they cannot. + +## Running the demos: + +1. `pip install .` +2. `cd test/` +3. In one terminal, run `local_broker.py` -- it prints the addresses it + bound +4. In a second terminal, cd into the demo and run `run_me.py` +5. In a third terminal, in the same demo, run its `sensor.py` if it has + one (`01-start-inference-stop` and `04-start-agent-stop` do) + +**When running a demo: be sure to start the ZMQ PubSub broker!** + +The third terminal is the point, not an inconvenience: a sensor is an +external entity. It is a process of its own with a lifetime of its own, +it publishes JSON on a shared channel, and it knows nothing about twins. +The twin binds that channel with `runtime.add_input(dtype, channel)`, and +a second twin binding the same channel receives the same messages -- which +is how one instrument feeds many twins. Start and stop the sensor +independently of the twin; neither cares. + +Demos without a `sensor.py` produce their input inside the twin, which is +what persistent components are still for: `06-agent-pi` drives itself off +a timer, and `07-barrier` off several. + +Every side resolves the broker addresses the same way: `DT_STREAM_PUB_ADDR` +and `DT_STREAM_SUB_ADDR`, defaulting to `tcp://127.0.0.1:5000` and `:5001` +(see `digitaltwin.config`). Set them in every terminal to move the broker. + +**Binding policy**: the broker binds to loopback by default, and it must +stay that way unless you know what you are doing -- twin-internal payloads +are cloudpickled, so anyone who can reach the broker ports can execute code +in every subscriber. A non-loopback bind needs an explicit configuration +and a private/firewalled network. External channels are decoded with the +codec their binding names: `json` (the default) and `raw` are safe to +accept from a producer you do not control, `cloudpickle` is not. + +## Running it as a service (the `dt` ORBIT plugin) + +`digitaltwin.service` exposes the framework as a long-running ORBIT +plugin: one session per client, many independent twins per session, and +twins that keep running while their client is away. Installing the +package registers the plugin through the `radical.orbit.plugins` entry +point, so a broker or endpoint only has to be told to host it. + +```sh +pip install .[service] + +# 1 - the broker, hosting the dt plugin +radical-orbit-broker.py --plugins default,dt + +# 2 - a rhapsody endpoint: where the twins' tasks execute. The notify +# window costs 250 ms on every sequential prediction at its default +radical-orbit-endpoint.py -n dt_task_ep +# ... started with: +# RADICAL_ORBIT_RHAPSODY_NOTIFY_WINDOW=0 +# RADICAL_ORBIT_RHAPSODY_BACKEND=concurrent +``` + +The client: + +```python +from radical.orbit import EndpointRuntime +from digitaltwin.components import NULL_DTYPE, TRUTHY +from digitaltwin.service import register_user_modules + +import my_components # not installed on the service +register_user_modules([my_components]) + +rt = EndpointRuntime() +rt.start(wait=True) + +# 'broker' is the participant hosting dt; engine wiring is explicit +dt = rt.get_plugin('broker', 'dt', config={ + 'engines': {'task': {'endpoint_name': 'dt_task_ep', + 'backends': ['concurrent']}}}) + +twin = dt.create_twin() # polls until the twin is ready +dt.add_task(twin, dt.package(MySensor), TRUTHY, SENSOR, is_persistent=True) +dt.add_investigator(twin, dt.package(MyModel), SENSOR, PREDICTION) +dt.start(twin) + +print(dt.twin_list()) # the observation mechanism +answer = dt.get_inference(twin, TypedData(SENSOR, 5), PREDICTION) + +dt.twin_close(twin) +``` + +The session outlives the client: reattach with +`rt.get_plugin('broker', 'dt', sid=)` and the twins are still +there. `dt.admin_sessions()` lists every session, twin, state and last +error on the service -- which is how orphans are found and torn down. +`test/09-service/` is a complete worked example. + +Three contract notes: + +- The client and the service must run **the same `digitaltwin` version** + (and compatible Python / cloudpickle): shipped component classes + pickle the framework by reference. Every call carries those versions + and the service rejects skew with a clear error rather than failing + somewhere inside an unpickle. +- A task's *arguments* are cloudpickled, but its **return value must be + JSON-safe or `bytes`** -- ORBIT's rhapsody plugin JSON-encodes results + and stringifies anything else. Return plain values from + `@flow.function_task` bodies and wrap them in `TypedData` in the + component. +- Persistent components run inline on the service's event loop. Their + bodies must be thin async glue publishing through + `runtime.stream`, never `@flow.function_task`s (the service warns when + it sees one). + +### Binding policy for the service (R7) + +The plugin runs its own DT stream broker, embedded, one per plugin and +shared by every twin. **It binds to loopback on a random port by +default, and that default is the safe one**: its payloads are +cloudpickled, so anyone who can reach the XSUB/XPUB ports gets code +execution in every subscriber -- weaker than the token-authenticated +ORBIT channel around it. + +A non-loopback bind is possible (`DT_STREAM_PUB_ADDR` / +`DT_STREAM_SUB_ADDR` on the service host) but requires a deliberate +decision *and* a firewalled or private network. Until the data plane +moves inside ORBIT's authenticated channel, do not expose those ports -- +including in demos. diff --git a/docs/dtaas-v1-plan.md b/docs/dtaas-v1-plan.md new file mode 100644 index 0000000..213b70f --- /dev/null +++ b/docs/dtaas-v1-plan.md @@ -0,0 +1,592 @@ +# DT-as-a-Service (DTaaS) — v1 implementation plan + +Status: five independent Fable review rounds (2026-08-14/15), all +findings folded in; compute-placement decision benchmarked; API model +settled: short synchronous verbs + async `twin_create` (rhapsody +pattern). Target repo: +`radical/digital_twins` (package `digitaltwin`). Related plan: +`orbit-p0-liveness-scoped-calls-plan.md` (P0) — **decoupled**: DTaaS +v1 does not depend on it; interim limitations while it is unmerged +are marked [P0-interim] below. + +## 1. Goal and target semantics + +Expose the experimental Digital Twin framework (this repo) as a +long-running service: an ORBIT plugin hosted on a standalone, +persistent ORBIT broker (driving use cases: AmSC / Matey fine-tuning, +xGFabric). In production the broker runs on capable dedicated +hardware. Endpoint-hosted deployment must remain possible +([P0-interim]: until P0 lands that mode sits under ORBIT's 30 s +relay backstop — harmless for the short verbs, limiting only +`get_inference`). + +Agreed semantics (decisions, not open questions): + +- **DTaaS is long-running; twins come and go.** Twins are defined and + managed *programmatically* by clients (no declarative twin spec in + v1). A serializable graph/twin description is welcome as + introspection but must not constrain semantics. +- **n twin instances per plugin session.** A session belongs to one + client; it hosts many independent twins. Twin teardown must not + disturb sibling twins or the session. +- **Twins survive disappearing clients, without timeout.** Twins may + run for days; clients attach/detach opportunistically. Sessions are + therefore forced `persistent` server-side; reattach uses the sid as + a bearer capability (ownership check relaxed within the + single-token trust domain); orphans are discoverable via the admin + listing and killable via the ordinary teardown routes. Explicit + lifecycle only — no idle expiry. +- **Recovery of twins across broker restart is out of scope for v1** + (candidate for v2/v3). Design state to be serializable where cheap. +- **Streams (pubsub)**: two stages. Stage 1 (v1): the DT framework's + own ZMQ pubsub broker, run by the plugin, separate from ORBIT + messaging. Stage 2: an ORBIT-pubsub backend behind the same + abstraction — promoted to stretch milestone M3 for security + reasons (risk R7), required before production. The pubsub + abstraction is a deployment-time backend choice at the same + architectural altitude as RHAPSODY (compute backends); the + backend *interface* is the seam — nothing above it may depend on + ZMQ specifics. +- **All user compute goes through the Rhapsody abstraction** (decided + after benchmarking, §6): asyncflow engines are constructed with + rhapsody's `OrbitExecutionBackend` targeting registered endpoints; + a co-located endpoint (same node as the broker) is the "local" + deployment. The broker process runs only the DT control plane. No + in-process `ProcessPoolExecutor` for user tasks. +- **Sensors are external entities.** The graph opens at its input + edge. A producer runs outside the framework and publishes to a + shared channel; a twin binds that channel to an input dtype with + `DTRuntime.add_input(dtype, channel, codec)` (M0.7). Channel topics + carry no twin namespace, so n twins may consume one channel and the + pubsub fan-out does the sharing. Producers precede and outlive any + twin, and neither side manages the other. Payload codecs are a + deployment choice: `json` for the plain scripts and instruments + which are the normal producers, `raw` for bytes, `cloudpickle` only + inside one trust domain (risk R7). +- **Persistent DT tasks (internal producers, in-situ loops) run in + the plugin host process in v1** — as plain async `main_loop` code + on the host loop, using an injected, namespaced stream client + (M0.3). Internal producers are timers, agent loops and other + sources a twin owns. They are no longer how data enters a graph, + which is `add_input`. + Persistent bodies are NOT `@flow.function_task`s: under an + Orbit-backed engine a function task would be cloudpickled to the + endpoint and occupy a backend slot for the twin's lifetime. This + simplifies the user API overall (today's hand-built task wrapping + and ZMQ clients disappear; `RuntimeAPI` gains its first publish + path). User code on the host loop (main_loops, callbacks, + selectors) is contractually thin async glue — documented (risk + R2), with one cheap guard: warn at instantiation if a persistent + component registered `function_task`s (catches the actual + migration mistake). Remote persistent components (psij + child-endpoint story) are post-v1. +- **Ex-situ learning uses ROSE in v1**, as a *plugin-local module* + (the ROSE "raas" service plugin is abandoned; do not depend on it). + ROSE's learner engine is `OrbitExecutionBackend`-backed like + everything else, typically targeting a remote HPC endpoint. ROSE + `StreamingActiveLearner` (PR #98, commit 64330d9) is an accepted + dependency. +- **Trust model**: clients ship code (cloudpickle) that executes in + the service. Accepted inside ORBIT's single-token trust domain + (same stance as rhapsody function tasks). Per-tenant auth is + post-v1. The DT *data plane* is currently weaker than that domain — + see risk R7 and milestone M3. + +## 2. Codebase facts the plan builds on + +Verified 2026-08-14/15 against: `digital_twins` @ `main` (df3b664), +`radical.orbit` @ `devel` (8f1d18c), `rose` @ +`feature/streaming_learner` (64330d9 — branch, not merged), +`rhapsody` @ dev/feature branches containing +`backends/execution/orbit.py`. Independently re-verified by four +fresh-eyes review agents. + +DT framework (`src/digitaltwin`, ~1200 lines): + +- `DTRuntime(flow: WorkflowEngine, streamer: PubSubClient)` + (`runtime.py:151`) never touches ZMQ directly; the only streaming + call is `streamer.subscribe_to_dtype` (`runtime.py:370`). +- **No `stop()`**: `start()` only sets an event; `running_tasks` are + never cancelled; demos tear down via `flow.shutdown()` only. The + existing done-callbacks call `.result()` before discarding + (`runtime.py:193-197`, `runtime.py:68-72`, `components.py:209`) and + are not cancellation-safe; component exceptions surface there as + loop log noise, invisible to any state machine. +- **Topics are un-namespaced**: `"runtime/dtypes/