Pool-backed engine roles (#7 phase 2, DT side)

src/digitaltwin/service/plugin.py

@@ -190,6 +190,13 @@ async def register_session(self, request: Request) -> dict:
                                     "backends": ["concurrent"]}}}
 
         `'learning'` is optional: unconfigured, it aliases `'inference'`.
+
+        A role may name a dispatcher-managed pool instead of an endpoint
+        (`{"pool": "exsitu"}`), with the pool configs declared once at the
+        session level (`"pools": [<PoolConfig>, ...]` -- the task
+        dispatcher's own schema).  Pool-backed roles run on the pool's
+        pilots, survive single-endpoint loss (tasks requeue; keep them
+        idempotent), and share one dispatcher session per DT session.
         """
 
         self._ensure_cleanup_task()

src/digitaltwin/service/session.py

@@ -595,13 +595,15 @@ async def _shutdown_engine(self, flow: WorkflowEngine) -> None:
                         self.sid, exc)
 
     async def _create_backend(self, name: str):
-        cfg = self._engine_config(name)
+        cfg  = self._engine_config(name)
+        pool = cfg.get("pool")
 
         log.info(
-            "[dt] session %s building %r backend on endpoint %s",
+            "[dt] session %s building %r backend on %s",
             self.sid,
             name,
-            cfg.get("endpoint_name") or "<auto>",
+            f"pool {pool!r}" if pool
+            else f"endpoint {cfg.get('endpoint_name') or '<auto>'}",
         )
 
         kwargs: dict = dict(
@@ -612,22 +614,56 @@ async def _create_backend(self, name: str):
             name=name,       # the asyncflow routing label: task backend=<role>
         )
 
+        params = inspect.signature(OrbitExecutionBackend.__init__).parameters
+
         # name the backend's broker participant after what it is for, so a
         # topology view shows `rhapsody.<session>.<role>` 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:
+        if "participant_name" in params:
             kwargs["participant_name"] = (
                 f"rhapsody.{self.sid.split('.')[-1]}.{name}")
 
+        if pool:
+            # A dispatcher-managed pool: the task dispatcher runs on the
+            # broker, and the pool's pilots pick the executing endpoints --
+            # this backend targets no endpoint of its own.  One dispatcher
+            # session per DT session, keyed by this session's own sid and
+            # shared by every role that names a pool; the session-level
+            # `pools` configs are declared with it, and re-declaring an
+            # owned pool is idempotent, so role build order does not
+            # matter.
+            #
+            # Contract: a task lost with its pilot is requeued and
+            # re-executes, so twin tasks must stay idempotent.  In return
+            # the twin no longer dies with a single endpoint -- pool-backed
+            # roles opt out of the R8 fail-fast (see `endpoints_lost`), and
+            # a task that ultimately fails surfaces through the normal
+            # task-status path.
+            if "pool" not in params:
+                raise RuntimeError(
+                    f"engine {name!r} names pool {pool!r}, but the installed"
+                    " rhapsody's OrbitExecutionBackend has no pool support")
+            session_kwargs: dict = {"sid": self.sid}
+            if self.config.get("pools"):
+                session_kwargs["pools"] = self.config["pools"]
+            kwargs.update(
+                plugin_name="task_dispatcher",
+                pool=pool,
+                session_kwargs=session_kwargs,
+            )
+
         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
+        # What a topology change is matched against: the endpoint the
+        # backend *settled on* (it auto-selects when the config named
+        # none).  A pool-backed role records its pool instead -- no lost
+        # endpoint ever matches it, which is exactly the R8 opt-out.
         self._endpoints[name] = (
-            getattr(backend, "_endpoint_name", None) or cfg.get("endpoint_name")
+            f"pool:{pool}" if pool
+            else getattr(backend, "_endpoint_name", None)
+            or cfg.get("endpoint_name")
         )
 
         return backend
@@ -645,6 +681,12 @@ def endpoints_lost(self, lost: set[str]) -> tuple[str, ...]:
 
         The loss is also remembered, because the broker announces it only
         once -- see `engine`.
+
+        A pool-backed role never matches: its `_endpoints` entry is
+        `pool:<name>`, not an endpoint.  Losing a pilot's endpoint there
+        requeues the pilot's tasks inside the dispatcher instead of
+        killing the twin -- the twin's failure surface for pools is a task
+        that ultimately fails, not a single endpoint going away.
         """
 
         names = set(self._endpoints) | set(self.config.get("engines") or {})

test/unit/test_service.py

@@ -817,3 +817,88 @@ def test_the_explorer_module_is_the_one_the_plugin_declares():
     assert module.is_file()
     assert module.name in UI_ASSETS
     assert "window.DTDash.mount" in module.read_text()
+
+
+# ---------------------------------------------------------------------------
+# pool-backed roles (dispatcher pools)
+# ---------------------------------------------------------------------------
+
+class _FakePoolBackend:
+    """Signature-faithful stand-in: the pool path checks the constructor's
+    parameters before using them."""
+
+    captured: dict = {}
+
+    def __init__(self, broker_url=None, endpoint_name=None, backends=None,
+                 name=None, participant_name=None, plugin_name="rhapsody",
+                 pool=None, session_kwargs=None, batch_window=0):
+        type(self).captured = dict(
+            endpoint_name=endpoint_name, name=name,
+            participant_name=participant_name, plugin_name=plugin_name,
+            pool=pool, session_kwargs=session_kwargs)
+        self._endpoint_name = endpoint_name or "broker"
+
+    def __await__(self):
+        async def _self():
+            return self
+        return _self().__await__()
+
+
+class _FakeOldBackend:
+    """A backend from before pool support -- no `pool` parameter."""
+
+    def __init__(self, broker_url=None, endpoint_name=None, backends=None,
+                 name=None, batch_window=0):
+        pass
+
+
+async def test_a_pool_backed_role_targets_the_dispatcher(monkeypatch):
+    """`pool` in a role's config routes it at the task dispatcher: one
+    dispatcher session per DT session (keyed by the DT sid), the
+    session-level pool configs declared with it."""
+
+    import digitaltwin.service.session as session_mod
+
+    session = DTSession("s1")
+    session.config = {
+        "pools": [{"name": "exsitu", "endpoint_name": "hpc1"}],
+        "engines": {"inference": {"endpoint_name": "ep1"},
+                    "learning": {"pool": "exsitu"}},
+    }
+    monkeypatch.setattr(session_mod, "OrbitExecutionBackend",
+                        _FakePoolBackend)
+
+    await session._create_backend("learning")
+
+    got = _FakePoolBackend.captured
+    assert got["plugin_name"] == "task_dispatcher"
+    assert got["pool"] == "exsitu"
+    assert got["session_kwargs"]["sid"] == "s1"
+    assert got["session_kwargs"]["pools"] == [
+        {"name": "exsitu", "endpoint_name": "hpc1"}]
+    # the role records its pool, not an endpoint -- the R8 opt-out
+    assert session._endpoints["learning"] == "pool:exsitu"
+
+
+async def test_a_pool_needs_a_pool_capable_backend(monkeypatch):
+    import digitaltwin.service.session as session_mod
+
+    session = DTSession("s1")
+    session.config = {"engines": {"learning": {"pool": "exsitu"}}}
+    monkeypatch.setattr(session_mod, "OrbitExecutionBackend",
+                        _FakeOldBackend)
+
+    with pytest.raises(RuntimeError, match="no pool support"):
+        await session._create_backend("learning")
+
+
+async def test_endpoint_loss_spares_pool_backed_roles():
+    """Losing an endpoint requeues a pool's tasks inside the dispatcher;
+    only endpoint-bound roles fail their twins (R8)."""
+
+    session = DTSession("s1")
+    session.config = {"engines": {"learning": {"pool": "exsitu"}}}
+    session._endpoints["learning"] = "pool:exsitu"
+
+    assert session.endpoints_lost({"hpc1", "exsitu", "pool"}) == ()
+    assert not session._lost