From 1d62ddc90f403595decb692dec0a138c17961132 Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Mon, 7 Sep 2026 13:36:27 +0200 Subject: [PATCH] fix(apps): advertise an app's service address only once it resolves MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_app_status derived service_ids from the worker's client id and gated only on "a ProxyDeployment replica is alive". That is a precondition for Hypha registration, not evidence of it: the proxy reports healthy to Ray while it waits for its siblings to come up and registers afterwards, so the worker handed out an address — and a static_site_url built from it — tens of seconds before anything answered there, with status already RUNNING. Measured gaps of 28.3 s and 20.6 s. The proxy now pushes its registration state to the BioEngineProxyActor (False at replica init, True after _register_services succeeds, False on deregistration) and the worker reads it before advertising. A never- reported app falls back to the replica-alive gate: an app recovered under a newer worker, or one whose actor was recreated, serves fine without ever reporting, and hiding its id would break a working deployment. get_app_status gains a service_registered field so a client can tell "not registered yet" from "this worker cannot tell", and the deploy log no longer claims completion while serve.run is still starting replicas. Closes svamp issue #0024. --- bioengine/apps/manager.py | 50 ++- bioengine/apps/proxy_deployment.py | 26 ++ bioengine/cluster/proxy_actor.py | 35 ++ docs/apps-guide.md | 2 + docs/glossary.md | 8 +- .../test_proxy_entry_saturation_tolerance.py | 1 + .../apps/test_proxy_hypha_decoupled_health.py | 1 + .../apps/test_service_id_registration_gate.py | 327 ++++++++++++++++++ tests/end_to_end/test_applications.py | 10 +- 9 files changed, 446 insertions(+), 14 deletions(-) create mode 100644 tests/apps/test_service_id_registration_gate.py diff --git a/bioengine/apps/manager.py b/bioengine/apps/manager.py index 783ca79..c8c54e3 100644 --- a/bioengine/apps/manager.py +++ b/bioengine/apps/manager.py @@ -528,10 +528,14 @@ async def _deploy_application( built_app = self._deployed_applications[application_id]["built_app"] await self.app_builder.submit(built_app, application_id) - # Track the application in the internal state + # ``submit`` ends in serve.run(blocking=False), so the replicas are + # only starting here and the Hypha service does not exist until the + # proxy registers it. Saying "completed" sent callers to the service + # address tens of seconds before anything answered it. self.logger.info( - f"Successfully completed deployment of application '{application_id}' from " - f"artifact '{artifact_id}', version '{version}'." + f"Submitted application '{application_id}' from artifact " + f"'{artifact_id}', version '{version}' to Ray Serve; replicas are " + f"starting and the Hypha service is registered once they run." ) # Mark the application as deployed @@ -815,7 +819,16 @@ def _filter_secret_env_vars( async def _get_application_service_ids( self, application_id: str - ) -> Dict[str, Optional[str]]: + ) -> Tuple[Dict[str, Optional[str]], Optional[bool]]: + """Service ids to advertise for an app, and whether its proxy has + registered them. + + Returns ``({websocket_service_id, webrtc_service_id}, registered)``. + Both ids are ``None`` unless a proxy replica is alive and has not + reported itself unregistered; ``registered`` is ``None`` when the proxy + has never reported — see ``BioEngineProxyActor.get_service_registration``. + """ + no_ids = {"websocket_service_id": None, "webrtc_service_id": None} # The proxy's Hypha client_id is derived deterministically from the # worker's client_id and a hash of application_id (see # bioengine.apps.proxy_deployment.ProxyDeployment.__init__). The URL @@ -829,7 +842,25 @@ async def _get_application_service_ids( ) ) if not replica_ids: - return {"websocket_service_id": None, "webrtc_service_id": None} + return no_ids, None + + # A live proxy replica is a precondition for registration, not proof of + # it: the replica reports healthy to Ray while it waits for its siblings + # to come up, and only registers with Hypha afterwards. Advertising the + # id in between hands out an address that does not resolve yet. + registered = None + try: + registered = ( + await self.ray_cluster.proxy_actor_handle.get_service_registration.remote( + application_id + ) + ) + except Exception as exc: + self.logger.debug( + f"Could not read service registration for '{application_id}': {exc}" + ) + if registered is False: + return no_ids, False workspace = self.server.config.workspace # For a recovered app, the ProxyDeployment is still registered @@ -847,7 +878,7 @@ async def _get_application_service_ids( return { "websocket_service_id": f"{workspace}/{proxy_client_id}:{application_id}", "webrtc_service_id": f"{workspace}/{proxy_client_id}:{application_id}-rtc", - } + }, registered async def _verify_running_identities( self, @@ -970,7 +1001,9 @@ async def _get_app_status( message = f"Application '{application_id}' has not been deployed yet." deployments = {} - service_ids = await self._get_application_service_ids(application_id) + service_ids, service_registered = await self._get_application_service_ids( + application_id + ) # Report what the replicas actually loaded, not just the requested # version — a stale reused replica reads as "healthy" otherwise. @@ -1021,6 +1054,9 @@ async def _get_app_status( "scaling": dict(application_info.get("scaling") or {}), "static_site_url": static_site_url, "service_ids": service_ids, + # None while the proxy has never reported, so a client can tell + # "not registered yet" from "this worker cannot tell". + "service_registered": service_registered, "start_time": application_info["started_at"], "last_updated_at": application_info["last_updated_at"], "last_updated_by": application_info["last_updated_by"], diff --git a/bioengine/apps/proxy_deployment.py b/bioengine/apps/proxy_deployment.py index 7935f79..3dfdf6d 100644 --- a/bioengine/apps/proxy_deployment.py +++ b/bioengine/apps/proxy_deployment.py @@ -332,6 +332,30 @@ def __init__( # Lock for service registration self._registration_lock = asyncio.Lock() + # No Hypha service exists until _register_services succeeds, so seed the + # worker-visible record as unregistered rather than leaving it unknown. + self._proxy_actor_handle = proxy_actor_handle + self._report_service_registration(False) + + def _report_service_registration(self, registered: bool) -> None: + """Tell the proxy actor whether our Hypha services exist right now. + + Fire-and-forget: the worker reads this to decide whether to advertise + this app's service address, and a failed report must never affect the + replica. + """ + if self._proxy_actor_handle is None: + return + try: + self._proxy_actor_handle.report_service_registration.remote( + application_id=self.application_id, registered=registered + ) + except Exception as e: + logger.warning( + f"⚠️ Could not report service registration state for " + f"'{self.application_id}': {e}" + ) + async def get_app_data(self) -> Dict[str, Any]: """Return non-secret application metadata used for worker recovery.""" return self.app_data @@ -965,6 +989,7 @@ async def _deregister_services(self) -> None: self._ice_expires_at = None # Reset so the next successful entry health check triggers re-registration self.entry_deployment_ready = False + self._report_service_registration(False) async def _reset_server_connection(self) -> None: """Cleanly disconnect ``self.server`` before we forget about it. @@ -1228,6 +1253,7 @@ async def _maintenance_tick(self) -> None: self._connection_lost = False self._probe_due_at = time.time() + _REACHABILITY_PROBE_INTERVAL_S self._next_register_at = 0.0 + self._report_service_registration(True) except Exception as e: if self._is_permanent_registration_error(e): logger.error( diff --git a/bioengine/cluster/proxy_actor.py b/bioengine/cluster/proxy_actor.py index 713e275..384029e 100644 --- a/bioengine/cluster/proxy_actor.py +++ b/bioengine/cluster/proxy_actor.py @@ -150,6 +150,13 @@ def __init__( str, Dict[str, Dict[str, Dict[str, Optional[str]]]] ] = {} + # Whether each application's ProxyDeployment currently has its Hypha + # services registered, pushed by the proxy replica itself. A missing + # entry means "never reported", which is not the same as False — an app + # that predates this actor keeps serving without ever reporting here. + # Structure: {app_id: bool} + self.service_registrations: Dict[str, bool] = {} + self._cached_geo_location: Optional[Dict[str, Optional[Union[str, float]]]] = None # Last successful per-node GPU memory read from the Ray dashboard. The @@ -825,10 +832,38 @@ def clear_application_replicas(self, application_id: str) -> None: """ self.application_replicas.pop(application_id, None) self.replica_identities.pop(application_id, None) + self.service_registrations.pop(application_id, None) logger.info( f"Cleared all registered replicas for application '{application_id}'." ) + @_touch_on_call + def report_service_registration( + self, application_id: str, registered: bool + ) -> None: + """Record whether an application's Hypha services exist right now. + + Pushed by ``ProxyDeployment``: False at replica init, True once + ``_register_services`` succeeds, False again on deregistration. The + worker reads it before advertising the app's service address, so a + client is never handed an id that nothing answers yet. + """ + self.service_registrations[application_id] = registered + logger.info( + f"Application '{application_id}' reported its Hypha services as " + f"{'registered' if registered else 'not registered'}." + ) + + @_touch_on_call + def get_service_registration(self, application_id: str) -> Optional[bool]: + """Last reported Hypha registration state, or None if never reported. + + None is not False: an app whose proxy replica started before this actor + existed — a different BioEngine version, or an actor recreated after + eviction — serves perfectly well without ever reporting here. + """ + return self.service_registrations.get(application_id) + @_touch_on_call def get_replica_identities( self, application_id: str diff --git a/docs/apps-guide.md b/docs/apps-guide.md index 6a6c0f4..4c6cd4f 100644 --- a/docs/apps-guide.md +++ b/docs/apps-guide.md @@ -1048,6 +1048,8 @@ websocket_service_id = app_status["service_ids"]["websocket_service_id"] webrtc_service_id = app_status["service_ids"]["webrtc_service_id"] ``` +Both ids are `None` until the proxy has registered the services with Hypha, which happens after every deployment of the app has a running replica. `status` reaches `RUNNING` before that, so poll for a non-`None` `websocket_service_id` rather than for `RUNNING`. + #### WebSocket Connection Websocket connections send and receive their data through the connected Hypha server. diff --git a/docs/glossary.md b/docs/glossary.md index 7681ff8..d2eeb5b 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -60,16 +60,18 @@ Fully qualified handle for a service. Grammar: **Two BioEngine service patterns:** - **Worker service** — one per worker. ID = `/:bioengine-worker`. The literal `bioengine-worker` is hardcoded; it exposes the worker's admin API (`get_status`, `deploy_app`, etc.). -- **Application services** — one **ProxyDeployment replica per application** (fixed). ID = `/-:` (plus `-rtc` for WebRTC). The `` is appended to the worker's `client_id`. An app's internal deployments may have their own Ray Serve replicas, but only the single ProxyDeployment in front of them yields Hypha services; internal deployments are addressed through Ray Serve handles inside the worker. +- **Application services** — one **ProxyDeployment replica per application** (fixed). ID = `/-:` (plus `-rtc` for WebRTC). The `` is the first 8 hex digits of `sha1(application_id)`, appended to the worker's `client_id` — so the id is stable across ProxyDeployment restarts and does **not** change on redeploy. An app's internal deployments may have their own Ray Serve replicas, but only the single ProxyDeployment in front of them yields Hypha services; internal deployments are addressed through Ray Serve handles inside the worker. **Calling an application.** `get_app_status()` returns a `service_ids` block of shape: ```python { - "websocket_service_id": "/-:", - "webrtc_service_id": "/-:-rtc", + "websocket_service_id": "/-:", + "webrtc_service_id": "/-:-rtc", } ``` +Both are `None` until the proxy has actually registered the services with Hypha, which happens after every deployment of the app has a running replica — so an app can read `RUNNING` while its ids are still `None`. Poll for a non-`None` `websocket_service_id` rather than for `status == "RUNNING"`; the companion `service_registered` field says whether the proxy has reported registering (`None` if it has never reported, e.g. an app recovered under a newer worker). + - **WebSocket** — `get_service(service_ids["websocket_service_id"])`. No selection mode needed; one concrete client. - **WebRTC** — `get_rtc_service(hypha_client, service_ids["webrtc_service_id"])`. Same concrete replica; peer-connection handshake addresses it directly. diff --git a/tests/apps/test_proxy_entry_saturation_tolerance.py b/tests/apps/test_proxy_entry_saturation_tolerance.py index be23464..2288eef 100644 --- a/tests/apps/test_proxy_entry_saturation_tolerance.py +++ b/tests/apps/test_proxy_entry_saturation_tolerance.py @@ -71,6 +71,7 @@ def _bare_proxy(**attrs): inst._probe_due_at = 0.0 inst._next_register_at = 0.0 inst._maintenance_task = None + inst._proxy_actor_handle = None # The maintenance loop is exercised in test_proxy_hypha_decoupled_health; # here it would only spawn a background task the gate tests never await. inst._ensure_maintenance_task = lambda: None diff --git a/tests/apps/test_proxy_hypha_decoupled_health.py b/tests/apps/test_proxy_hypha_decoupled_health.py index 08165ba..6ec16a6 100644 --- a/tests/apps/test_proxy_hypha_decoupled_health.py +++ b/tests/apps/test_proxy_hypha_decoupled_health.py @@ -47,6 +47,7 @@ def _bare_proxy(**attrs): inst._registration_failure = None inst._probe_due_at = 0.0 inst._next_register_at = 0.0 + inst._proxy_actor_handle = None for key, value in attrs.items(): setattr(inst, key, value) return inst diff --git a/tests/apps/test_service_id_registration_gate.py b/tests/apps/test_service_id_registration_gate.py new file mode 100644 index 0000000..603b6c6 --- /dev/null +++ b/tests/apps/test_service_id_registration_gate.py @@ -0,0 +1,327 @@ +"""Pin that a service address is advertised only once it actually resolves. + +``get_app_status`` used to derive ``service_ids`` from the worker's client id +and gate only on "a ProxyDeployment replica is alive". A live proxy replica is +a *precondition* for Hypha registration, not evidence of it: the replica +reports healthy to Ray while it waits for its siblings to come up, and only +then registers with Hypha. So the worker handed out an id — and a +``static_site_url`` built from it — tens of seconds before anything answered +at that address, with ``status`` already reading ``RUNNING``. + +The contract now: + +* The proxy pushes its registration state to the ``BioEngineProxyActor`` + (False at replica init, True after ``_register_services`` succeeds, False on + deregistration), and the worker reads it before advertising. +* A ``False`` report withholds both ids and the ``static_site_url``. +* *Never reported* is not ``False``. An app whose proxy started before this + actor existed — a newer worker, or an actor recreated after eviction — + serves perfectly well and must keep its id, so the worker falls back to the + replica-alive gate. +""" +from __future__ import annotations + +import asyncio +import inspect +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from bioengine.apps import proxy_deployment as pd_module +from bioengine.apps.manager import AppsManager +from bioengine.cluster.proxy_actor import BioEngineProxyActor + +_ProxyCls = pd_module.ProxyDeployment.func_or_class +_ActorCls = BioEngineProxyActor.__ray_actor_class__ + +APP_ID = "nuclei-seg" +WORKER_CLIENT_ID = "worker-abc" + + +def _make_app_info() -> dict: + deployed = asyncio.Event() + deployed.set() + return { + "is_deployed": deployed, + "display_name": "Nuclei Segmentation", + "description": "Segment nuclei.", + "artifact_id": "bioimage-io/nuclei-seg", + "version": "1.0.1", + "recovered_app": False, + "application_kwargs": {}, + "application_env_vars": {}, + "disable_gpu": False, + "application_resources": {}, + "authorized_users": ["*"], + "available_methods": ["segment"], + "max_ongoing_requests": 1, + "scaling": {}, + "static_site_url": "https://static.example/nuclei-seg/", + "started_at": 1_700_000_000.0, + "last_updated_at": 1_700_000_000.0, + "last_updated_by": "user@example.com", + "auto_redeploy": False, + "deployed_by_worker_client_id": WORKER_CLIENT_ID, + "proxy_service_token_issued_at": None, + "proxy_service_token_ttl_seconds": None, + } + + +def _make_manager(*, replicas: dict, registered) -> AppsManager: + """An AppsManager wired with only what the service-id path touches. + + ``registered`` is the proxy actor's answer: ``True``/``False``/``None``, or + an exception instance to raise from the actor call. + """ + manager = object.__new__(AppsManager) + manager.logger = logging.getLogger("test") + manager._deployed_applications = {APP_ID: _make_app_info()} + + server = MagicMock() + server.config.workspace = "bioimage-io" + server.config.client_id = WORKER_CLIENT_ID + server.config.public_base_url = "https://hypha.example" + manager.server = server + + ray_cluster = MagicMock() + handle = ray_cluster.proxy_actor_handle + handle.get_deployment_replicas.remote = AsyncMock(return_value=replicas) + if isinstance(registered, BaseException): + handle.get_service_registration.remote = AsyncMock(side_effect=registered) + else: + handle.get_service_registration.remote = AsyncMock(return_value=registered) + manager.ray_cluster = ray_cluster + return manager + + +async def _status(manager: AppsManager) -> dict: + return await manager._get_app_status( + application_id=APP_ID, + instance_details={ + "applications": {APP_ID: {"status": "RUNNING", "deployments": {}}} + }, + n_previous_replica=0, + logs_tail=30, + ) + + +# ===== the worker only advertises what the proxy has registered ===== + + +@pytest.mark.asyncio +async def test_a_live_but_unregistered_proxy_advertises_no_service_id() -> None: + # The reported bug verbatim: the ProxyDeployment replica is alive, so the + # old gate passed, but its Hypha services do not exist yet. + manager = _make_manager(replicas={"r0": "actor0"}, registered=False) + + ids, registered = await manager._get_application_service_ids(APP_ID) + + assert ids == {"websocket_service_id": None, "webrtc_service_id": None} + assert registered is False + + +@pytest.mark.asyncio +async def test_a_registered_proxy_advertises_the_service_id() -> None: + manager = _make_manager(replicas={"r0": "actor0"}, registered=True) + + ids, registered = await manager._get_application_service_ids(APP_ID) + + assert registered is True + assert ids["websocket_service_id"].endswith(f":{APP_ID}") + assert ids["webrtc_service_id"].endswith(f":{APP_ID}-rtc") + + +@pytest.mark.asyncio +async def test_no_replica_still_means_no_service_id() -> None: + # The pre-existing gate must survive: no proxy replica, no address. + manager = _make_manager(replicas={}, registered=True) + + ids, registered = await manager._get_application_service_ids(APP_ID) + + assert ids == {"websocket_service_id": None, "webrtc_service_id": None} + assert registered is None + + +# ===== absence of a report is not a negative report ===== + + +@pytest.mark.asyncio +async def test_an_app_that_never_reported_keeps_its_service_id() -> None: + # A recovered app, or one whose proxy predates this actor, serves without + # ever reporting. Hiding its id would break a working deployment. + manager = _make_manager(replicas={"r0": "actor0"}, registered=None) + + ids, registered = await manager._get_application_service_ids(APP_ID) + + assert registered is None + assert ids["websocket_service_id"] is not None + + +@pytest.mark.asyncio +async def test_an_unreadable_registration_record_does_not_hide_a_live_app() -> None: + manager = _make_manager( + replicas={"r0": "actor0"}, registered=RuntimeError("actor unreachable") + ) + + ids, registered = await manager._get_application_service_ids(APP_ID) + + assert registered is None + assert ids["websocket_service_id"] is not None + + +# ===== what the status payload says ===== + + +@pytest.mark.asyncio +async def test_status_withholds_the_static_site_url_until_registration() -> None: + # The static site is handed the service id as a query parameter, so an + # unregistered app must not produce a URL either. + manager = _make_manager(replicas={"r0": "actor0"}, registered=False) + + status = await _status(manager) + + assert status["status"] == "RUNNING" + assert status["static_site_url"] is None + assert status["service_registered"] is False + + +@pytest.mark.asyncio +async def test_status_builds_the_static_site_url_once_registered() -> None: + manager = _make_manager(replicas={"r0": "actor0"}, registered=True) + + status = await _status(manager) + + assert status["service_registered"] is True + assert status["static_site_url"].startswith("https://static.example/nuclei-seg/?") + assert status["service_ids"]["websocket_service_id"] in status["static_site_url"] + + +# ===== the proxy actor's side of the record ===== + + +def _bare_actor() -> object: + actor = object.__new__(_ActorCls) + actor._last_called_at = 0.0 + actor.application_replicas = {} + actor.replica_identities = {} + actor.service_registrations = {} + return actor + + +def test_the_actor_reports_none_before_anything_is_pushed() -> None: + actor = _bare_actor() + assert actor.get_service_registration(APP_ID) is None + + +def test_the_actor_round_trips_both_registration_states() -> None: + actor = _bare_actor() + actor.report_service_registration(APP_ID, False) + assert actor.get_service_registration(APP_ID) is False + actor.report_service_registration(APP_ID, True) + assert actor.get_service_registration(APP_ID) is True + + +def test_clearing_an_application_forgets_its_registration() -> None: + # Otherwise a redeployed app inherits the previous deployment's verdict. + actor = _bare_actor() + actor.report_service_registration(APP_ID, True) + actor.clear_application_replicas(APP_ID) + assert actor.get_service_registration(APP_ID) is None + + +# ===== the proxy pushes the state it is in ===== + + +class _Handle: + """Stands in for the Ray actor handle; records the ``.remote`` payloads.""" + + def __init__(self) -> None: + self.reports = [] + self.report_service_registration = MagicMock() + self.report_service_registration.remote = self._record + + def _record(self, application_id: str, registered: bool) -> None: + self.reports.append((application_id, registered)) + + +def _bare_proxy(**attrs): + inst = object.__new__(_ProxyCls) + inst.application_id = APP_ID + inst.entry_deployment_ready = True + inst.server = None + inst.websocket_service_id = None + inst.rtc_service_id = None + inst.mcp_service_id = None + inst._rtc_config = None + inst._ice_expires_at = None + inst._registration_lock = asyncio.Lock() + inst._maintenance_task = None + inst._connection_lost = False + inst._registration_failure = None + inst._probe_due_at = 0.0 + inst._next_register_at = 0.0 + inst._proxy_actor_handle = _Handle() + for key, value in attrs.items(): + setattr(inst, key, value) + return inst + + +@pytest.mark.asyncio +async def test_a_successful_registration_is_reported() -> None: + inst = _bare_proxy() + + async def _register(): + inst.websocket_service_id = "ws" + + inst._register_services = _register + + await inst._maintenance_tick() + + assert inst._proxy_actor_handle.reports == [(APP_ID, True)] + + +@pytest.mark.asyncio +async def test_a_failed_registration_is_not_reported_as_registered() -> None: + inst = _bare_proxy() + + async def _register(): + raise RuntimeError("hypha down") + + inst._register_services = _register + + await inst._maintenance_tick() + + assert inst._proxy_actor_handle.reports == [] + + +@pytest.mark.asyncio +async def test_deregistering_reports_the_service_as_gone() -> None: + inst = _bare_proxy() + + await inst._deregister_services() + + assert inst._proxy_actor_handle.reports == [(APP_ID, False)] + + +def test_the_proxy_seeds_the_record_as_unregistered_at_init() -> None: + # Without the seed the worker sees "never reported" on a brand-new app and + # falls back to the replica-alive gate — i.e. the original bug. + src = inspect.getsource(_ProxyCls.__init__) + assert "self._report_service_registration(False)" in src + + +def test_a_missing_actor_handle_never_breaks_the_replica() -> None: + inst = _bare_proxy(_proxy_actor_handle=None) + inst._report_service_registration(True) # must not raise + + +# ===== the deployment log no longer claims completion ===== + + +def test_deploy_does_not_claim_completion_before_the_replicas_run() -> None: + src = inspect.getsource(AppsManager._deploy_application) + assert "Successfully completed deployment" not in src, ( + "serve.run is non-blocking here — the replicas are only starting, and " + "the Hypha service does not exist until the proxy registers it." + ) diff --git a/tests/end_to_end/test_applications.py b/tests/end_to_end/test_applications.py index be6150b..17e023e 100644 --- a/tests/end_to_end/test_applications.py +++ b/tests/end_to_end/test_applications.py @@ -30,9 +30,8 @@ def top_level_name(file: Dict) -> str: async def resolve_service(lookup, timeout: int = 60): """Retry a Hypha service lookup until the registration shows up. - `get_app_status` derives `service_ids` from the worker's client id and only - gates on a live ProxyDeployment replica, so an id is reported before that - replica has finished registering it with Hypha. + `get_app_status` reports an id only once the proxy has registered it, but + the lookup still races the registration's propagation through Hypha. """ start_time = time.time() while True: @@ -344,7 +343,10 @@ async def test_startup_application( while time.time() - start_time < application_check_timeout: apps_status = await bioengine_worker_service.get_app_status() running = [ - app for app in apps_status.values() if app["status"] == "RUNNING" + app + for app in apps_status.values() + if app["status"] == "RUNNING" + and (app["service_ids"] or {}).get("websocket_service_id") ] if len(running) >= expected_app_count: break