fix(apps): report a startup version pin that reverts a running app

bioengine/apps/manager.py

@@ -309,6 +309,56 @@ def _get_full_artifact_id(self, artifact_id: str) -> str:
             # If artifact_id does not contain a slash, prepend the workspace
             return f"{self.server.config.workspace}/{artifact_id}"
 
+    def _get_startup_version_pin(
+        self, application_id: str, artifact_id: str
+    ) -> Optional[str]:
+        """Return the version the startup config will restore on the next restart.
+
+        None when the app is not a startup application, or is pinned without a
+        version — in that case a restart redeploys whatever is already running.
+        """
+        for app_config in self.startup_applications:
+            if not isinstance(app_config, dict):
+                continue
+            if app_config.get("application_id") != application_id:
+                continue
+            config_artifact_id = app_config.get("artifact_id")
+            if not config_artifact_id:
+                continue
+            if self._get_full_artifact_id(config_artifact_id) == artifact_id:
+                return app_config.get("version")
+        return None
+
+    def _warn_on_startup_pin_divergence(self, app_config: Dict[str, Any]) -> None:
+        """Report a startup pin that is about to move a running app to another version.
+
+        ``recover_deployed_applications()`` runs first, so both numbers are in
+        hand here. Without this the revert is indistinguishable from a clean
+        start — same RUNNING status, same healthy probes, just older code.
+        """
+        pinned_version = app_config.get("version")
+        application_id = app_config.get("application_id")
+        if not pinned_version or not application_id:
+            return
+
+        running = self._deployed_applications.get(application_id)
+        if not running or running["version"] == pinned_version:
+            return
+        if running["artifact_id"] != self._get_full_artifact_id(
+            app_config["artifact_id"]
+        ):
+            return
+
+        self.logger.warning(
+            f"Startup application '{application_id}' (artifact "
+            f"'{running['artifact_id']}') was found running version "
+            f"{running['version'] or 'latest'!r} but is pinned to "
+            f"{pinned_version!r} — deploying the pinned version. If "
+            f"{running['version'] or 'latest'!r} was rolled out over the API, "
+            f"that change is being reverted; update the worker's "
+            f"startup_applications config to keep it."
+        )
+
     async def _generate_application_id(self) -> str:
         """
         Generate a unique identifier for a new application deployment.
@@ -985,6 +1035,13 @@ async def _get_app_status(
                 application_id, application_details, application_info["version"]
             )
 
+        # The version a worker restart would restore. Differs from ``version``
+        # only when the app was rolled out over the API without updating the
+        # worker's startup_applications config; None when it isn't pinned.
+        pinned_version = self._get_startup_version_pin(
+            application_id, application_info["artifact_id"]
+        )
+
         # Build static site URL with runtime config params so the frontend
         # knows which Hypha server and service to connect to.
         base_static_url = application_info.get("static_site_url")
@@ -1005,6 +1062,7 @@ async def _get_app_status(
             "version": application_info["version"] or "latest",
             "running_version": running_version,
             "version_verified": version_verified,
+            "pinned_version": pinned_version,
             "recovered_app": application_info["recovered_app"],
             "status": status,
             "message": message,
@@ -1292,6 +1350,8 @@ async def deploy_startup_applications(self) -> None:
                         f"{', '.join(sorted(invalid_keys))}. Valid keys are: {', '.join(sorted(valid_keys))}"
                     )
 
+                self._warn_on_startup_pin_divergence(app_config)
+
                 if "hypha_token" not in app_config:
                     app_config["hypha_token"] = startup_applications_token
 

tests/apps/test_startup_pin_divergence.py

@@ -0,0 +1,241 @@
+"""Pin the reporting of a startup version pin that reverts a running app.
+
+A worker pod restart replays ``--startup-applications``, but an app rolled out
+over the API only ever changed the worker's runtime state. When the pin still
+carries the old version the restart quietly reinstates it: status RUNNING,
+deployments HEALTHY, no diff against prior state — the app is simply running
+older code. That happened on deNBI on 2026-09-01 (model-runner 2.7.2 -> 2.4.2)
+and was found only because someone compared versions by hand.
+
+``recover_deployed_applications()`` runs before ``deploy_startup_applications()``,
+so the worker holds both numbers at that moment. Two pins here:
+
+- ``deploy_startup_applications`` logs a WARNING naming both versions before it
+  redeploys at the pin.
+- ``get_app_status`` returns ``pinned_version`` alongside the running one, so
+  the divergence is assertable over the API without reading cluster config.
+
+Neither changes what gets deployed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from bioengine.apps.manager import AppsManager
+
+RUNNING_VERSION = "2.7.2"
+PINNED_VERSION = "2.4.2"
+ARTIFACT_ID = "bioimage-io/model-runner"
+APP_ID = "model-runner"
+
+
+def _make_manager(
+    *,
+    startup_applications: list,
+    running_version: str | None = RUNNING_VERSION,
+    running_artifact_id: str = ARTIFACT_ID,
+) -> AppsManager:
+    """An AppsManager holding one recovered app plus the given startup pins."""
+    is_deployed = asyncio.Event()
+    is_deployed.set()
+
+    manager = object.__new__(AppsManager)
+    manager.logger = logging.getLogger("test.startup_pin")
+    manager.startup_applications = startup_applications
+    manager._deployed_applications = {
+        APP_ID: {
+            "is_deployed": is_deployed,
+            "display_name": "Model Runner",
+            "description": "Run bioimage.io models.",
+            "artifact_id": running_artifact_id,
+            "version": running_version,
+            "recovered_app": True,
+            "application_kwargs": {},
+            "application_env_vars": {},
+            "disable_gpu": False,
+            "application_resources": {},
+            "authorized_users": ["*"],
+            "available_methods": ["infer"],
+            "max_ongoing_requests": 1,
+            "scaling": {},
+            "static_site_url": None,
+            "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-abc",
+            "proxy_service_token_issued_at": None,
+            "proxy_service_token_ttl_seconds": None,
+        }
+    }
+
+    server = MagicMock()
+    server.config.workspace = "bioimage-io"
+    server.generate_token = AsyncMock(return_value="startup-token")
+    manager.server = server
+    manager.admin_users = ["admin@example.com"]
+
+    ray_cluster = MagicMock()
+    ray_cluster.proxy_actor_handle.get_deployment_replicas.remote = AsyncMock(
+        return_value={}
+    )
+    manager.ray_cluster = ray_cluster
+
+    # deploy_app is the thing under observation, not under test: replace it but
+    # keep the real __schema__, which deploy_startup_applications reads to
+    # validate config keys.
+    deploy_app = AsyncMock(return_value=APP_ID)
+    deploy_app.__schema__ = AppsManager.deploy_app.__schema__
+    manager.deploy_app = deploy_app
+
+    return manager
+
+
+@pytest.mark.asyncio
+async def test_divergent_pin_warns_with_both_versions(caplog) -> None:
+    manager = _make_manager(
+        startup_applications=[
+            {
+                "artifact_id": ARTIFACT_ID,
+                "application_id": APP_ID,
+                "version": PINNED_VERSION,
+            }
+        ]
+    )
+
+    with caplog.at_level(logging.WARNING, logger="test.startup_pin"):
+        await manager.deploy_startup_applications()
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert warnings, (
+        "A restart that reverts a running app to an older pin must say so. "
+        "Without this line the revert is indistinguishable from a clean start."
+    )
+    message = warnings[0].getMessage()
+    assert RUNNING_VERSION in message and PINNED_VERSION in message, message
+    assert APP_ID in message
+
+    # Observability only — the pinned version is still deployed.
+    manager.deploy_app.assert_awaited_once()
+    assert manager.deploy_app.await_args.kwargs["version"] == PINNED_VERSION
+
+
+@pytest.mark.asyncio
+async def test_matching_pin_is_silent(caplog) -> None:
+    manager = _make_manager(
+        startup_applications=[
+            {
+                "artifact_id": ARTIFACT_ID,
+                "application_id": APP_ID,
+                "version": RUNNING_VERSION,
+            }
+        ]
+    )
+
+    with caplog.at_level(logging.WARNING, logger="test.startup_pin"):
+        await manager.deploy_startup_applications()
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING] == []
+
+
+@pytest.mark.asyncio
+async def test_pin_without_a_version_is_silent(caplog) -> None:
+    # An unversioned pin inherits whatever is already running, so there is
+    # nothing to revert and nothing to report.
+    manager = _make_manager(
+        startup_applications=[
+            {"artifact_id": ARTIFACT_ID, "application_id": APP_ID},
+        ]
+    )
+
+    with caplog.at_level(logging.WARNING, logger="test.startup_pin"):
+        await manager.deploy_startup_applications()
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING] == []
+
+
+@pytest.mark.asyncio
+async def test_pin_for_a_different_artifact_is_silent(caplog) -> None:
+    # Same application_id, different artifact: the versions belong to two
+    # different release lines and comparing them says nothing.
+    manager = _make_manager(
+        startup_applications=[
+            {
+                "artifact_id": ARTIFACT_ID,
+                "application_id": APP_ID,
+                "version": PINNED_VERSION,
+            }
+        ],
+        running_artifact_id="bioimage-io/other-app",
+    )
+
+    with caplog.at_level(logging.WARNING, logger="test.startup_pin"):
+        await manager.deploy_startup_applications()
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING] == []
+
+
+@pytest.mark.asyncio
+async def test_status_surfaces_the_pin_next_to_the_running_version() -> None:
+    manager = _make_manager(
+        startup_applications=[
+            {
+                "artifact_id": ARTIFACT_ID,
+                "application_id": APP_ID,
+                "version": PINNED_VERSION,
+            }
+        ]
+    )
+
+    status = await manager._get_app_status(
+        application_id=APP_ID,
+        instance_details={"applications": {}},
+        n_previous_replica=0,
+        logs_tail=30,
+    )
+
+    assert status["version"] == RUNNING_VERSION
+    assert status["pinned_version"] == PINNED_VERSION
+
+
+@pytest.mark.asyncio
+async def test_status_reports_no_pin_for_an_unpinned_app() -> None:
+    manager = _make_manager(startup_applications=[])
+
+    status = await manager._get_app_status(
+        application_id=APP_ID,
+        instance_details={"applications": {}},
+        n_previous_replica=0,
+        logs_tail=30,
+    )
+
+    assert status["pinned_version"] is None
+
+
+@pytest.mark.asyncio
+async def test_status_resolves_a_short_artifact_id_in_the_pin() -> None:
+    # Startup configs may name the artifact without its workspace prefix; the
+    # tracked app always carries the full form.
+    manager = _make_manager(
+        startup_applications=[
+            {
+                "artifact_id": "model-runner",
+                "application_id": APP_ID,
+                "version": PINNED_VERSION,
+            }
+        ]
+    )
+
+    status = await manager._get_app_status(
+        application_id=APP_ID,
+        instance_details={"applications": {}},
+        n_previous_replica=0,
+        logs_tail=30,
+    )
+
+    assert status["pinned_version"] == PINNED_VERSION

tests/apps/test_status_app_missing_from_cluster.py

@@ -67,6 +67,7 @@ def _make_manager(app_id: str, app_info: dict) -> AppsManager:
     manager = object.__new__(AppsManager)
     manager.logger = logging.getLogger("test")
     manager._deployed_applications = {app_id: app_info}
+    manager.startup_applications = []
 
     ray_cluster = MagicMock()
     # No ProxyDeployment replicas exist once the head wiped Serve state;