From 1b018054376aeccf4bbc52933052a388f91e2b0f Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Wed, 19 Aug 2026 12:21:20 +0200 Subject: [PATCH 1/5] orbit backend: let the caller name its broker participant The broker draws every connected runtime as a participant, so the backend's anonymous `rhapsody.` shows up as noise in a topology view -- two of them, unexplained, for a session with two engines. A caller which knows what the backend is for can now say so (`participant_name="rhapsody.."`); uniqueness becomes that caller's contract. Unnamed backends keep the unique suffix and the old behavior exactly. Co-Authored-By: Claude Opus 5 (1M context) --- src/rhapsody/backends/execution/orbit.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/rhapsody/backends/execution/orbit.py b/src/rhapsody/backends/execution/orbit.py index 90e4fbb..88547a6 100644 --- a/src/rhapsody/backends/execution/orbit.py +++ b/src/rhapsody/backends/execution/orbit.py @@ -94,6 +94,7 @@ def __init__( endpoint_name: str | None = None, backends: list[str] | None = None, name: str = "orbit", + participant_name: str | None = None, plugin_name: str = _PLUGIN_NAME, batch_window: float | None = None, batch_limit: int = 1024, @@ -109,6 +110,7 @@ def __init__( self.logger = logging.getLogger(__name__) self._broker_url = broker_url + self._participant_name = participant_name self._endpoint_name = endpoint_name self._plugin_name = plugin_name self._remote_backends = backends or ["dragon_v3"] @@ -325,12 +327,16 @@ def _get_rhapsody_handle(self) -> Any: """ import uuid - # A unique name suffix avoids the broker's name-in-use rejection when - # several rhapsody clients connect (or one restarts within the - # liveness grace window). + # The broker shows every runtime as a participant, so an anonymous + # `rhapsody.` reads as noise in a topology view. A caller + # which knows what this backend is *for* can say so via + # ``participant_name`` -- uniqueness is then the caller's contract. + # Without one, a unique suffix avoids the broker's name-in-use + # rejection when several rhapsody clients connect (or one restarts + # within the liveness grace window). rt = EndpointRuntime( broker_url=self._broker_url, - name=f"rhapsody.{uuid.uuid4().hex[:8]}", + name=self._participant_name or f"rhapsody.{uuid.uuid4().hex[:8]}", ) try: rt.start(wait=True, timeout=self._start_timeout) From 42f28fd77d01c01552be623013fd9229b71144ef Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Wed, 19 Aug 2026 12:53:16 +0200 Subject: [PATCH 2/5] orbit backend: advertise as 'engine', not 'consumer' 'consumer' is the runtime's say-nothing default. A topology viewer now sees what the participant is: the engine side of the compute hand-off. Co-Authored-By: Claude Opus 5 (1M context) --- src/rhapsody/backends/execution/orbit.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/rhapsody/backends/execution/orbit.py b/src/rhapsody/backends/execution/orbit.py index 88547a6..1cacea8 100644 --- a/src/rhapsody/backends/execution/orbit.py +++ b/src/rhapsody/backends/execution/orbit.py @@ -337,6 +337,9 @@ def _get_rhapsody_handle(self) -> Any: rt = EndpointRuntime( broker_url=self._broker_url, name=self._participant_name or f"rhapsody.{uuid.uuid4().hex[:8]}", + # the advertised role defaults to 'consumer', which says nothing. + # This participant is the workflow engine's hand-off into ORBIT. + role="engine", ) try: rt.start(wait=True, timeout=self._start_timeout) From 5ffac8914b4e9be62ce3b7da1bc0e026c1075472 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 25 Aug 2026 00:11:45 +0200 Subject: [PATCH 3/5] orbit backend: run tasks in a dispatcher-managed pool A `pool` argument turns the backend's target from a per-endpoint rhapsody session into a task-dispatcher-style plugin: with no explicit endpoint the broker participant (which hosts the dispatcher) is the target and endpoint auto-selection is skipped -- the pool's pilots pick the executing endpoints, not this backend. Every submitted task is stamped with the pool (a task already naming one keeps it), and the python-version handshake for cloudpickled tasks resolves against the pool's executing endpoint (via the dispatcher's pool detail) instead of the submission target. `session_kwargs` forwards into the remote session registration, so a backend can join an existing dispatcher session by sid -- pools are keyed per session. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- src/rhapsody/backends/execution/orbit.py | 40 +++++++++++- tests/unit/test_backend_execution_orbit.py | 75 ++++++++++++++++++++++ 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/src/rhapsody/backends/execution/orbit.py b/src/rhapsody/backends/execution/orbit.py index 1cacea8..8d38435 100644 --- a/src/rhapsody/backends/execution/orbit.py +++ b/src/rhapsody/backends/execution/orbit.py @@ -72,6 +72,22 @@ class OrbitExecutionBackend(BaseBackend): (default ``["dragon_v3"]``). name: Backend name for Rhapsody registration (default ``"orbit"``). + pool: Name of a dispatcher-managed pool to run tasks in. + Setting it targets a task-dispatcher-style plugin + instead of a per-endpoint rhapsody session: every + submitted task is stamped ``pool=`` (a task + already carrying one keeps it), endpoint + auto-selection is skipped in favour of the broker + participant hosting the dispatcher, and the + python-version handshake for cloudpickled tasks + resolves against the pool's *executing* endpoint + rather than the submission target. Combine with + ``plugin_name="task_dispatcher"``. + session_kwargs: Extra keyword arguments for the remote session + registration (``get_plugin``) -- e.g. ``sid`` to + join an existing dispatcher session whose pools + this backend should see, or ``pools`` to declare + them. batch_window: Seconds to collect tasks before flushing (default 0.25). Set to 0 to disable batching. batch_limit: Max tasks per batch — triggers an immediate flush @@ -96,6 +112,8 @@ def __init__( name: str = "orbit", participant_name: str | None = None, plugin_name: str = _PLUGIN_NAME, + pool: str | None = None, + session_kwargs: dict | None = None, batch_window: float | None = None, batch_limit: int = 1024, start_timeout: float = 30.0, @@ -113,6 +131,8 @@ def __init__( self._participant_name = participant_name self._endpoint_name = endpoint_name self._plugin_name = plugin_name + self._pool = pool + self._session_kwargs = dict(session_kwargs or {}) self._remote_backends = backends or ["dragon_v3"] self._start_timeout = start_timeout self._init_timeout = init_timeout @@ -353,6 +373,12 @@ def _get_rhapsody_handle(self) -> Any: ) self._broker_url = rt.broker_url + # pool mode: the dispatcher lives on the broker participant, so + # there is no endpoint to select -- the pool's pilots pick the + # executing endpoints, not this backend + if self._pool and not self._endpoint_name: + self._endpoint_name = "broker" + # find a suitable endpoint from the (local) topology snapshot if not self._endpoint_name: for eid, info in rt.topology().items(): @@ -382,6 +408,7 @@ def _get_rhapsody_handle(self) -> Any: self._plugin_name, backends=self._remote_backends, init_timeout=self._init_timeout, + **self._session_kwargs, ) except Exception: # Don't leak the runtime's daemon threads / WebSocket on a failed @@ -430,10 +457,17 @@ def needs_compat(t: dict) -> bool: info = None def _lookup(): + # Pool mode: the submission target (the dispatcher's host) + # does not execute anything — the pool's endpoint does, so + # that is what the cloudpickle handshake compares against. + target = self._endpoint_name + if self._pool and hasattr(self._rh, "pool_detail"): + target = self._rh.pool_detail(self._pool).get( + "endpoint_name") or target # get_plugin() auto-registers an ephemeral session even # though host_role() needs none — close the client so failed # retries don't accumulate sessions on the endpoint. - si = self._runtime.get_plugin(self._endpoint_name, "sysinfo") + si = self._runtime.get_plugin(target, "sysinfo") try: return si.host_role() finally: @@ -505,6 +539,10 @@ async def submit_tasks(self, tasks: list[dict[str, Any]]) -> None: prof = self._prof for task in tasks: task.setdefault("uid", f"task.{uuid.uuid4().hex[:8]}") + # pool mode: route every task into this backend's pool; a task + # that already names one keeps it + if self._pool: + task.setdefault("pool", self._pool) self._tasks[task["uid"]] = task if prof: prof.prof("task_submit", uid=task["uid"]) diff --git a/tests/unit/test_backend_execution_orbit.py b/tests/unit/test_backend_execution_orbit.py index 1974cb9..30eff2b 100644 --- a/tests/unit/test_backend_execution_orbit.py +++ b/tests/unit/test_backend_execution_orbit.py @@ -893,3 +893,78 @@ async def test_cancel_all_drops_buffered_tasks(): await backend.shutdown() backend._mock_rh.submit_tasks.assert_not_called() + + +# --------------------------------------------------------------------------- +# Pool mode (dispatcher-managed pools) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pool_mode_targets_broker_and_forwards_session_kwargs(): + """With a pool and no endpoint, the broker participant (hosting the + dispatcher) is the target -- no endpoint auto-selection -- and session + kwargs ride into get_plugin so the backend can join a session by sid.""" + backend = await _init_backend( + endpoint_name=None, + pool="insitu", + plugin_name="task_dispatcher", + session_kwargs={"sid": "session.abc"}, + ) + + assert backend._endpoint_name == "broker" + backend._mock_rt.get_plugin.assert_called_once_with( + "broker", + "task_dispatcher", + backends=["dragon_v3"], + init_timeout=120.0, + sid="session.abc", + ) + + +@pytest.mark.asyncio +async def test_pool_mode_stamps_tasks(): + """Every submitted task carries the backend's pool; an explicit per-task + pool wins.""" + backend = await _init_backend( + endpoint_name=None, pool="insitu", plugin_name="task_dispatcher", batch_window=0 + ) + + await backend.submit_tasks( + [ + {"uid": "t.001", "executable": "/bin/echo"}, + {"uid": "t.002", "executable": "/bin/echo", "pool": "special"}, + ] + ) + + submitted = backend._mock_rh.submit_tasks.call_args[0][0] + assert submitted[0]["pool"] == "insitu" + assert submitted[1]["pool"] == "special" + + +@pytest.mark.asyncio +async def test_pool_mode_compat_resolves_executing_endpoint(): + """The cloudpickle python handshake compares against the pool's executing + endpoint (from the dispatcher's pool detail), not the dispatcher's host.""" + import sys + + backend = await _init_backend( + endpoint_name=None, pool="insitu", plugin_name="task_dispatcher", batch_window=0 + ) + backend._mock_rh.pool_detail = MagicMock( + return_value={"name": "insitu", "endpoint_name": "compute_ep"} + ) + + si = MagicMock() + mm = sys.version_info + si.host_role = MagicMock(return_value={"python_version": f"{mm.major}.{mm.minor}.0"}) + si.close = MagicMock() + backend._mock_rt.get_plugin = MagicMock(return_value=si) + + def fn(): + return 42 + + await backend.submit_tasks([{"uid": "t.003", "function": fn}]) + + backend._mock_rt.get_plugin.assert_called_once_with("compute_ep", "sysinfo") + assert backend._endpoint_python_mm == (mm.major, mm.minor) From 237196ca790c0ac458a589542c9733cee142b3d0 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 25 Aug 2026 11:21:12 +0200 Subject: [PATCH 4/5] tests: wrap pool-mode docstrings the way docformatter wants Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- tests/unit/test_backend_execution_orbit.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/tests/unit/test_backend_execution_orbit.py b/tests/unit/test_backend_execution_orbit.py index 30eff2b..9b21927 100644 --- a/tests/unit/test_backend_execution_orbit.py +++ b/tests/unit/test_backend_execution_orbit.py @@ -902,9 +902,9 @@ async def test_cancel_all_drops_buffered_tasks(): @pytest.mark.asyncio async def test_pool_mode_targets_broker_and_forwards_session_kwargs(): - """With a pool and no endpoint, the broker participant (hosting the - dispatcher) is the target -- no endpoint auto-selection -- and session - kwargs ride into get_plugin so the backend can join a session by sid.""" + """With a pool and no endpoint, the broker participant (hosting the dispatcher) is the target -- + no endpoint auto-selection -- and session kwargs ride into get_plugin so the backend can join a + session by sid.""" backend = await _init_backend( endpoint_name=None, pool="insitu", @@ -924,8 +924,7 @@ async def test_pool_mode_targets_broker_and_forwards_session_kwargs(): @pytest.mark.asyncio async def test_pool_mode_stamps_tasks(): - """Every submitted task carries the backend's pool; an explicit per-task - pool wins.""" + """Every submitted task carries the backend's pool; an explicit per-task pool wins.""" backend = await _init_backend( endpoint_name=None, pool="insitu", plugin_name="task_dispatcher", batch_window=0 ) @@ -944,8 +943,8 @@ async def test_pool_mode_stamps_tasks(): @pytest.mark.asyncio async def test_pool_mode_compat_resolves_executing_endpoint(): - """The cloudpickle python handshake compares against the pool's executing - endpoint (from the dispatcher's pool detail), not the dispatcher's host.""" + """The cloudpickle python handshake compares against the pool's executing endpoint (from the + dispatcher's pool detail), not the dispatcher's host.""" import sys backend = await _init_backend( From 380f19935cf8c5a5b8608f6d001f227774dcd787 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 25 Aug 2026 11:24:21 +0200 Subject: [PATCH 5/5] orbit backend: let ruff format the pool_detail line Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- src/rhapsody/backends/execution/orbit.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/rhapsody/backends/execution/orbit.py b/src/rhapsody/backends/execution/orbit.py index 8d38435..0a8f3ba 100644 --- a/src/rhapsody/backends/execution/orbit.py +++ b/src/rhapsody/backends/execution/orbit.py @@ -462,8 +462,7 @@ def _lookup(): # that is what the cloudpickle handshake compares against. target = self._endpoint_name if self._pool and hasattr(self._rh, "pool_detail"): - target = self._rh.pool_detail(self._pool).get( - "endpoint_name") or target + target = self._rh.pool_detail(self._pool).get("endpoint_name") or target # get_plugin() auto-registers an ephemeral session even # though host_role() needs none — close the client so failed # retries don't accumulate sessions on the endpoint.