fix(apps): advertise an app's service address only once it resolves

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"],

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(

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

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.

docs/glossary.md

@@ -60,16 +60,18 @@ Fully qualified handle for a service. Grammar:
 **Two BioEngine service patterns:**
 
 - **Worker service** — one per worker. ID = `<workspace>/<client_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 = `<workspace>/<worker_client_id>-<replica_id>:<application_id>` (plus `<application_id>-rtc` for WebRTC). The `<replica_id>` 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 = `<workspace>/<worker_client_id>-<app_hash>:<application_id>` (plus `<application_id>-rtc` for WebRTC). The `<app_hash>` 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": "<workspace>/<worker_client_id>-<replica_id>:<application_id>",
-    "webrtc_service_id":    "<workspace>/<worker_client_id>-<replica_id>:<application_id>-rtc",
+    "websocket_service_id": "<workspace>/<worker_client_id>-<app_hash>:<application_id>",
+    "webrtc_service_id":    "<workspace>/<worker_client_id>-<app_hash>:<application_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.

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

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