Change the M2 PR to use pools from Task Dispatcher

[BenCarter44]

Currently, the DTaaS-M2 PR is implementing its own "multi-engine" logic.

For separation of concerns, change the DTaaS to use Task Dispatcher's Pools.

---

Example use:

The user defines their digital twin with asyncflow tasks and then can label the pool they want to run their task.

# an example API:

# my_existu_pool is a pool name known to the broker, submitted by some config...

@flow.function_task(pool="my_exsitu_pool")
async def do_train(a, b, c):
    . . .

[BenCarter44]

This issue will make the #1 PR ready for merge.

[andre-merzky]

Design shape for the pools rework, after reading the dispatcher and asyncflow source:

• asyncflow routes per task already: function_task(backend=...) sets target_backend, the engine submits per named backend, unset tasks use the default (0.5.1). No asyncflow change needed.
• OrbitExecutionBackend takes plugin_name; pointed at the dispatcher it works as is, given the dispatcher speaks the rhapsody execution dialect.

Naming: the backend / pool roles are insitu and exsitu (symmetric, matches the vocabulary in the plan and in this issue). task from the merged M1 config stays as a deprecated alias for insitu for one cycle.

TODO, in order:

• phase 1, DT only (~60-100 LoC): one engine, two named backends (insitu, exsitu); the ROSE wrapper drops learn_flow and labels its training tasks backend="exsitu". Delivers the tag model without the dispatcher.
• dispatcher: submit_tasks(list[dict]) accepting rhapsody-style task dicts (cloudpickle convention), pool key per task; mixed-pool batches group server-side (~100-150 LoC)
• dispatcher: one ledger persist per batch; coalesce completion-side persists on a flush window. Removes the per-task fsync, the dominant latency cost of pool mode (measured baseline: direct path 19-20 ms in-situ p50; per-task pool mode adds 3-15 ms)
• dispatcher: batch task_status notifications (same NOTIFY_WINDOW semantics as plugin_rhapsody), so the backend's existing consumer works unchanged
• backend: target the broker participant, skip endpoint auto-selection, resolve the python-version handshake against the executing endpoint (~10-40 LoC)
• DT: pool config via dispatcher register_session, one dispatcher session per DT session (pool lifetime = session lifetime); replace the endpoint-loss fail-fast with pool-health mapping — pilot loss requeues tasks, a twin fails on pool exhaustion only
• contract note: requeued tasks re-execute; twin tasks stay idempotent

Sequencing: this lands as a follow-up PR series. #1/#2 merge first (demo-proven, CI green), phase 1 directly after as a small DT PR. The dispatcher items are orbit-side and can proceed in parallel; the DT pool wiring drops once those merge. In-situ inference stays on the direct backend path; pools serve exsitu (latency analysis above).

[andre-merzky]

Naming correction to the comment above: the reserved role names are inference and learning. insitu/exsitu was placement vocabulary leaking into role names (inference can run remotely; roles describe function, pools describe placement). No backward-compatibility aliases: the framework had one internal demo, phase 1 renames clean.

Rename touch points (phase 1, one sweep): service/session.py (engine/backend roles, endpoints map), service/wire.py, learn.py, service/ui/dt_dash.js (lane and badge labels), test/12-dtaas-live (driver ENGINES config), unit and integration tests. PRs touching these files after phase 1 resolve conflicts toward inference/learning; #6 uses this vocabulary in the new UI directly.

[andre-merzky]

Phase 2 implemented, three PRs:

• radical.orbit#117 — dispatcher speaks the rhapsody execution dialect: bulk submit_rh/{sid} with rhapsody-format task dicts + per-task pool key, mixed-pool batches grouped server-side, dicts forwarded verbatim to pilot rhapsody sessions (uid namespaced per owning session). One ledger persist per pool per batch/drain; completion persists and notifications coalesce on the plugin_rhapsody NOTIFY_WINDOW; dialect completions emit rhapsody-shaped task_status/task_status_batch.
• rhapsody#88 — OrbitExecutionBackend(pool=..., plugin_name='task_dispatcher', session_kwargs={'sid': ...}): targets the broker participant, skips endpoint auto-selection, stamps pool per task, resolves the cloudpickle python handshake against the pool's executing endpoint. Stacked on #87.
• digital.twins#18 — role config takes {"pool": "<name>"} with session-level pools configs; one dispatcher session per DT session (keyed by DT sid, shared across roles); pool-backed roles opt out of R8 (pilot loss requeues; twin fails on task failure, not endpoint loss). Requeue contract noted: twin tasks stay idempotent.

Merge order: orbit#117 and rhapsody#87+#88 first, then #18; pins move afterwards. The pool path is config-gated — nothing changes without a pool key.

Remaining from the TODO list: none code-side. Open follow-ups: end-to-end pool run against a live broker (needs the merged pins), and a pool-exhaustion policy in the dispatcher (currently tasks queue until a pilot returns; no exhaustion signal exists to fail a twin on).

[andre-merzky]

Phase 2 complete: radical.orbit#117 (dispatcher rhapsody dialect, released as 0.7.0), #18 (DT-side pool-backed roles), #20 (pin removal). rhapsody#87/#88 (backend pool mode) are green and await review — tracked there.