diff --git a/bioengine/_app/bootstrap.py b/bioengine/_app/bootstrap.py index 8dc6932c..542b397d 100644 --- a/bioengine/_app/bootstrap.py +++ b/bioengine/_app/bootstrap.py @@ -21,6 +21,7 @@ import importlib import inspect import sys +from pathlib import Path from typing import Any, Dict, List, Optional from bioengine._app.errors import ( @@ -76,6 +77,25 @@ def _purge_stale_source_modules(source_root: str) -> None: importlib.invalidate_caches() +def hash_source_tree(source: Path) -> str: + """Content hash of a materialised app source tree (bytecode caches excluded). + + A version string can't distinguish same-version-different-content; this can. + The submit task bakes it onto each user class as ``code_hash`` so a running + replica reports the code it actually loaded, and the introspect task returns + it so the worker can tell a real content change from an unchanged redeploy. + Both must hash the same way for those two values to be comparable. + """ + import hashlib + + hasher = hashlib.md5() + for path in sorted(source.rglob("*")): + if path.is_file() and "__pycache__" not in path.parts: + hasher.update(path.relative_to(source).as_posix().encode()) + hasher.update(path.read_bytes()) + return hasher.hexdigest()[:16] + + # ───────────────────────────── introspection ───────────────────────────── @@ -139,7 +159,7 @@ def introspect_app_in_ray_task( ) -> Dict[str, Any]: """Phase-1 Ray task: download user source and introspect it. - Returns ``{"spec": …}``. + Returns ``{"spec": …, "source_signature": …}``. The download uses the Hypha ``BIOENGINE_ARTIFACT_FILES_URL`` (+ optional ``_DOWNLOAD_TOKEN``) env vars via ``_ensure_source``. Replicas materialise @@ -177,7 +197,12 @@ def introspect_app_in_ray_task( _purge_stale_source_modules(src_str) spec = introspect_app(entry_id) - return {"spec": spec} + + # Fingerprint what was actually synced, so the worker can tell a real + # content change from a version string that stayed the same (a re-staged + # version, or a redeploy at "latest"). The spec alone can't: a changed + # method body leaves qualnames and schemas identical. + return {"spec": spec, "source_signature": hash_source_tree(source)} def _walk( @@ -505,18 +530,7 @@ def build_and_run_application( sys.path.insert(0, src_str) _purge_stale_source_modules(src_str) - # Content hash of the materialised source — the code identity we bake into - # each user class below so a running replica reports what it *actually* - # loaded (a version string can't distinguish same-version-different-content; - # a content hash can). Excludes bytecode caches. - import hashlib as _hashlib - - _src_hasher = _hashlib.md5() - for _p in sorted(Path(src_str).rglob("*")): - if _p.is_file() and "__pycache__" not in _p.parts: - _src_hasher.update(_p.relative_to(src_str).as_posix().encode()) - _src_hasher.update(_p.read_bytes()) - source_hash = _src_hasher.hexdigest()[:16] + source_hash = hash_source_tree(head_source) head_artifact_id = replica_env_vars.get("BIOENGINE_ARTIFACT_ID") handles: Dict[str, Any] = {} diff --git a/bioengine/apps/builder.py b/bioengine/apps/builder.py index 9293f2dc..6878b5d8 100644 --- a/bioengine/apps/builder.py +++ b/bioengine/apps/builder.py @@ -312,8 +312,9 @@ async def _introspect_via_ray_task( """Submit :func:`introspect_app_in_ray_task` as a Ray task. The task syncs the user package from Hypha (token in ``env_vars``), - walks the type-hint composition graph, and returns the ``{spec}`` - payload. We never touch the worker's filesystem. + walks the type-hint composition graph, and returns the + ``{spec, source_signature}`` payload. We never touch the worker's + filesystem. Strips the ``_BIOENGINE_SECRET_*`` keys from ``env_vars`` before passing into the task to keep secrets out of Ray's logs; the @@ -669,11 +670,13 @@ async def build( # 3. Introspect the user package via a Ray task — the task syncs the # source from Hypha and walks the @bioengine.app composition. Returns - # spec only; replicas sync their own source per file from Hypha. + # the spec plus a content hash of the synced source; replicas sync their + # own source per file from Hypha. introspect_result = await self._introspect_via_ray_task( entry_id, env_vars, runtime_env ) spec = introspect_result["spec"] + source_signature = introspect_result.get("source_signature") self.logger.info(f"Introspect task returned for '{application_id}'") # Sanity check: format_version round-trip. @@ -731,6 +734,7 @@ async def build( "proxy_service_token_ttl_seconds": proxy_service_token_ttl_seconds, "entry": entry_id, "spec_hash": spec_hash, + "source_signature": source_signature, "display_name": manifest["name"], "description": manifest["description"], "artifact_id": artifact_id, @@ -779,6 +783,7 @@ async def build( "name": manifest["name"], "description": manifest["description"], "version": version, + "source_signature": source_signature, "resources": required_resources, "authorized_users": effective_authorized_users, "available_methods": available_methods, diff --git a/bioengine/apps/manager.py b/bioengine/apps/manager.py index 783ca792..5e1faf79 100644 --- a/bioengine/apps/manager.py +++ b/bioengine/apps/manager.py @@ -248,6 +248,12 @@ def __init__( # moment an app is observed healthy again. See monitor_applications. self._redeploy_backoff: Dict[str, Dict[str, Any]] = {} + # Application ids already warned about an identity mismatch the monitor + # cannot self-heal. Both such conditions persist until a human + # redeploys, and the monitor ticks every ~10s — without this the log + # fills with the same line for as long as the worker lives. + self._identity_warned: set = set() + # Timestamp of the last in-place Serve-controller-loss recovery sweep, # used to rate-limit re-firing while a redeploy's serve.run is still # bootstrapping the controller. See _recover_from_controller_loss. @@ -681,6 +687,7 @@ async def _undeploy_application( # Remove from internal tracking after all cleanup operations complete self._deployed_applications.pop(application_id, None) self._redeploy_backoff.pop(application_id, None) + self._identity_warned.discard(application_id) self.logger.info(f"Undeployment of application '{application_id}' completed.") def _project_replicas( @@ -854,7 +861,7 @@ async def _verify_running_identities( application_id: str, application_details: Dict[str, Any], expected_version: str, - ) -> Tuple[Optional[str], Optional[bool]]: + ) -> Tuple[Optional[str], Optional[bool], Optional[bool]]: """Cross-reference each live replica's baked identity against the deployed version, entirely off the data plane. @@ -866,16 +873,28 @@ async def _verify_running_identities( deployment reports its *baked* stale version here, so the worker can detect and force a real restart without ever issuing an in-band request. - Returns ``(running_version, version_verified)``: the entry deployment's - running replica version, and whether every live replica booted the - expected version. Either is ``None`` when no live replica has a known - identity yet, so the caller never acts on partial data. + Returns ``(running_version, version_verified, code_verified)``: the entry + deployment's running replica version, whether every live replica booted + the expected version, and whether every live replica booted the expected + source content. Each is ``None`` when it can't be determined, so the + caller never acts on partial data. + + ``code_verified`` catches the case a version string cannot: the same + version re-staged with different files. It is reported only — a false + value never triggers a delete, because a systematic hash disagreement + would turn the monitor's self-heal into a redeploy loop on a healthy app. + + A recovered app carries no ``built_app``, so the entry deployment can't + be named and ``running_version`` stays ``None`` — but the verification + itself covers every deployment and does not need the spec. Bailing out + early on a missing spec would leave apps that survived a worker restart + permanently unverified, which is exactly when a warm replica is most + likely to be running code the version pin no longer describes. """ - info = self._deployed_applications.get(application_id) - built_app = info.get("built_app") if info else None - spec = getattr(built_app, "spec", None) - if not spec: - return None, None + info = self._deployed_applications.get(application_id) or {} + built_app = info.get("built_app") + expected_signature = info.get("source_signature") + spec = getattr(built_app, "spec", None) or {} entry_cid = spec.get("entry_id") classes = spec.get("classes") or {} entry_name = ( @@ -893,10 +912,11 @@ async def _verify_running_identities( self.logger.debug( f"Could not read replica identities for '{application_id}': {exc}" ) - return None, None + return None, None, None running_version = None verified = True + code_verified = None checked = 0 for deployment_name, deployment_info in ( application_details.get("deployments") or {} @@ -911,11 +931,16 @@ async def _verify_running_identities( checked += 1 if ident.get("version") != expected_version: verified = False + running_hash = ident.get("code_hash") + if expected_signature is not None and running_hash is not None: + code_verified = (running_hash == expected_signature) and ( + code_verified is not False + ) if deployment_name == entry_name and running_version is None: running_version = ident.get("version") if not checked: - return running_version, None - return running_version, verified + return running_version, None, None + return running_version, verified, code_verified async def _get_app_status( self, @@ -976,12 +1001,19 @@ async def _get_app_status( # version — a stale reused replica reads as "healthy" otherwise. # ``running_version`` shows the entry replica's baked version; # ``version_verified`` reflects EVERY live replica (a reused replica of - # any deployment, not just the entry, counts as unverified). Both are - # None when they can't be determined. + # any deployment, not just the entry, counts as unverified), and + # ``code_verified`` does the same for the source content, catching a + # re-staged version the version string alone cannot. All are None when + # they can't be determined. running_version = None version_verified = None + code_verified = None if status == "RUNNING": - running_version, version_verified = await self._verify_running_identities( + ( + running_version, + version_verified, + code_verified, + ) = await self._verify_running_identities( application_id, application_details, application_info["version"] ) @@ -1005,6 +1037,7 @@ async def _get_app_status( "version": application_info["version"] or "latest", "running_version": running_version, "version_verified": version_verified, + "code_verified": code_verified, "recovered_app": application_info["recovered_app"], "status": status, "message": message, @@ -1200,6 +1233,7 @@ async def recover_deployed_applications(self) -> None: "description": app_data["description"], "artifact_id": app_data["artifact_id"], "version": app_data["version"], + "source_signature": app_data.get("source_signature"), "application_kwargs": app_data["application_kwargs"], "application_env_vars": app_data["application_env_vars"], "hypha_token": None, @@ -1403,10 +1437,47 @@ async def monitor_applications(self) -> None: application_details = (instance_details.get("applications") or {}).get( application_id, {} ) - _, version_verified = await self._verify_running_identities( + ( + _, + version_verified, + code_verified, + ) = await self._verify_running_identities( application_id, application_details, application_info["version"] ) + if version_verified is not False and code_verified is not False: + self._identity_warned.discard(application_id) + if code_verified is False and version_verified is not False: + # Same version, different files. Report only: acting on this + # would delete on every tick if the two hashes ever disagree + # systematically. Redeploy explicitly to clear it. + if application_id not in self._identity_warned: + self._identity_warned.add(application_id) + self.logger.warning( + f"Application '{application_id}' reports RUNNING at " + f"the expected version " + f"{application_info['version']!r}, but a live " + f"replica loaded different source content than the " + f"deployed bundle. Redeploy it to load the current " + f"source." + ) if version_verified is False: + if application_info.get("built_app") is None: + # Recovered app: deleting it would strand it, since the + # redeploy path has no built application to resubmit. + # Report the mismatch and leave it serving — + # get_app_status carries version_verified=False so the + # split-brain is visible instead of silent. + if application_id not in self._identity_warned: + self._identity_warned.add(application_id) + self.logger.warning( + f"Application '{application_id}' reports RUNNING " + f"but a live replica loaded a version != " + f"{application_info['version']!r}. It was " + f"recovered from a previous worker, so it cannot " + f"be rebuilt here — redeploy it explicitly to " + f"load the pinned version." + ) + continue self.logger.warning( f"Application '{application_id}' reports RUNNING but a " f"live replica loaded a version != requested " @@ -1491,6 +1562,19 @@ def _fire_redeploy( attempt: int, ) -> None: """Schedule a redeploy task and log the attempt number.""" + if application_info.get("built_app") is None: + # Recovered from a previous worker: there is nothing to resubmit, + # and _deploy_application would only raise on the None built_app. + # Say so once per attempt instead; recovery is the liveness + # backstop's pod cycle, or an explicit deploy_app by a user. + self.logger.warning( + f"Application '{application_id}' for artifact " + f"'{application_info['artifact_id']}' is unhealthy but was " + f"recovered from a previous worker and carries no built " + f"application; skipping auto-redeploy (attempt #{attempt}). " + f"Redeploy it explicitly to bring it back under this worker." + ) + return self.logger.warning( f"Application '{application_id}' for artifact " f"'{application_info['artifact_id']}' is unhealthy; triggering " @@ -2418,35 +2502,6 @@ async def deploy_app( f"version '{version}'; kwargs: {kwargs_str}; env_vars: {env_vars_str}" ) await self._cancel_deployment_process(application_id=application_id) - - # A content change (new version, or a different artifact under - # this application_id) must reach every replica. serve.run's - # in-place update can silently reuse replicas — at - # num_replicas=1 with no surge headroom (e.g. a single GPU) - # the old replica keeps serving stale in-memory code. Delete - # first so the slot frees and fresh replicas load the new - # source. Clear is_deployed so the monitor loop doesn't fire a - # redundant redeploy during the delete→rebuild window. - content_changed = ( - version != existing_app["version"] - or artifact_id != existing_app["artifact_id"] - ) - if content_changed: - existing_app["is_deployed"].clear() - try: - await self.ray_cluster.call_with_reconnect( - serve.delete, application_id - ) - self.logger.info( - f"Deleted Ray Serve application '{application_id}' " - f"before redeploy so replicas are recreated with " - f"the new source." - ) - except Exception as delete_err: - self.logger.error( - f"Error deleting Ray Serve application " - f"'{application_id}' before redeploy: {delete_err}" - ) else: # Create a new application self.logger.info( @@ -2514,6 +2569,55 @@ async def deploy_app( f"{nr}; must be >= 0." ) + # A content change (new version, different artifact, or the same + # version string re-staged with different files) must reach every + # replica. serve.run's in-place update can silently reuse replicas — + # at num_replicas=1 with no surge headroom (e.g. a single GPU) the + # old replica keeps serving the module it imported at first start. + # Delete first so the slot frees and fresh replicas load the new + # source. Clear is_deployed so the monitor loop doesn't fire a + # redundant redeploy during the delete→rebuild window. + # + # This runs AFTER the build because only the build fingerprints the + # files it actually synced. A version string cannot see content that + # changed underneath it, so a re-staged version took the in-place + # serve.run path while _ensure_source had already refreshed the + # source on disk. It stays BEFORE _check_resources so the old app's + # reservation is already released when free capacity is measured. + if is_update: + new_signature = app.metadata.get("source_signature") + old_signature = existing_app.get("source_signature") + content_changed = ( + artifact_id != existing_app["artifact_id"] + or app.metadata["version"] != existing_app["version"] + # Both known and different: files changed under one version. + # Either unknown: fall back to the identity check above + # rather than restarting a healthy app on a config-only + # update (apps recovered from a pre-0.16.6 worker carry no + # signature). + or ( + new_signature is not None + and old_signature is not None + and new_signature != old_signature + ) + ) + if content_changed: + existing_app["is_deployed"].clear() + try: + await self.ray_cluster.call_with_reconnect( + serve.delete, application_id + ) + self.logger.info( + f"Deleted Ray Serve application '{application_id}' " + f"before redeploy so replicas are recreated with " + f"the new source." + ) + except Exception as delete_err: + self.logger.error( + f"Error deleting Ray Serve application " + f"'{application_id}' before redeploy: {delete_err}" + ) + # Check resources before creating deployment task await self._check_resources( application_id=application_id, @@ -2535,6 +2639,7 @@ async def deploy_app( "description": app.metadata["description"], "artifact_id": artifact_id, "version": app.metadata["version"], + "source_signature": app.metadata.get("source_signature"), "application_kwargs": app.metadata["application_kwargs"], "application_env_vars": app.metadata["application_env_vars"], "hypha_token": hypha_token, diff --git a/tests/apps/test_stale_actor_holes.py b/tests/apps/test_stale_actor_holes.py new file mode 100644 index 00000000..ea6b2a4f --- /dev/null +++ b/tests/apps/test_stale_actor_holes.py @@ -0,0 +1,369 @@ +"""Pin the three holes through which a stale actor kept serving. + +The worker's guard against a reused Ray Serve replica was a version-string +comparison, made only for apps this worker had itself built. That left three +ways for an actor running old code to be reported as healthy: + +1. A version string cannot see content that changed underneath it. Re-staging + the same version with different files took the in-place ``serve.run`` path, + which may reuse the replica, and nothing compared what the replica had + actually loaded. +2. ``_verify_running_identities`` returned early when the app carried no + ``built_app`` — exactly the apps recovered from a previous worker, which are + the most likely to be running code the version pin no longer describes. +3. The recovered app that *was* found stale was deleted anyway, even though the + redeploy path has nothing to resubmit for it, so the delete stranded it. + +The counterpart constraint: a content mismatch is reported, never acted on. If +the two hashes ever disagreed systematically, deleting on it would turn the +monitor into a redeploy loop against a healthy app. +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from bioengine._app.bootstrap import hash_source_tree +from bioengine.apps.manager import AppsManager + +APP_ID = "model-runner" +VERSION = "2.8.0" +SIGNATURE = "0123456789abcdef" + + +def _built_app(entry_deployment: str = "ModelRunner"): + return SimpleNamespace( + spec={"entry_id": "cid0", "classes": {"cid0": {"qualname": entry_deployment}}} + ) + + +def _details(*replicas: tuple[str, str, str]) -> dict: + """Serve instance details for one app: (deployment, replica_id, state).""" + deployments: dict = {} + for deployment, replica_id, state in replicas: + deployments.setdefault(deployment, {"replicas": []})["replicas"].append( + {"replica_id": replica_id, "state": state} + ) + return {"deployments": deployments} + + +def _make_manager(*, built_app=_built_app(), source_signature=SIGNATURE) -> AppsManager: + is_deployed = asyncio.Event() + is_deployed.set() + + manager = object.__new__(AppsManager) + manager.logger = logging.getLogger("test.holes") + manager._redeploy_backoff = {} + manager._identity_warned = set() + # Bookkeeping the monitor keeps for its other branches; empty is the + # nothing-pending state these tests want. + manager._missing_from_status = {} + manager._deleted_pending_redeploy = set() + manager._deployed_applications = { + APP_ID: { + "is_deployed": is_deployed, + "artifact_id": "bioimage-io/model-runner", + "version": VERSION, + "source_signature": source_signature, + "auto_redeploy": True, + "built_app": built_app, + "deployment_task": None, + } + } + + ray_cluster = MagicMock() + ray_cluster.check_connection = AsyncMock() + ray_cluster.proxy_actor_handle.get_serve_instance_details.remote = AsyncMock( + return_value={} + ) + ray_cluster.proxy_actor_handle.get_replica_identities.remote = AsyncMock( + return_value={} + ) + manager.ray_cluster = ray_cluster + manager._deploy_application = AsyncMock() + + return manager + + +def _warnings(caplog) -> list[str]: + return [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING] + + +# ───────────────────── hole 2: verification without a spec ───────────────────── + + +@pytest.mark.asyncio +async def test_a_recovered_app_is_verified_even_without_a_built_app() -> None: + # The whole point: an app that survived a worker restart has no built_app, + # and used to skip verification entirely. Its replicas are precisely the + # ones that have been alive longest. + manager = _make_manager(built_app=None) + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": "2.7.0", "code_hash": SIGNATURE}} + } + + running_version, verified, _ = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert verified is False, ( + "A recovered app running the wrong version must be detectable; " + "returning None here is what let it serve stale code unnoticed." + ) + # No spec means the entry deployment cannot be named, so this stays unknown + # rather than guessing — but that does not block the verdict above. + assert running_version is None + + +@pytest.mark.asyncio +async def test_a_stale_replica_of_a_non_entry_deployment_fails_the_check() -> None: + # Serve may reuse a replica of any deployment, not just the entry one. + manager = _make_manager() + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": SIGNATURE}}, + "Helper": {"r2": {"version": "2.7.0", "code_hash": SIGNATURE}}, + } + + running_version, verified, _ = await manager._verify_running_identities( + APP_ID, + _details(("ModelRunner", "r1", "RUNNING"), ("Helper", "r2", "RUNNING")), + VERSION, + ) + + assert running_version == VERSION # the entry replica alone looks fine + assert verified is False + + +@pytest.mark.asyncio +async def test_nothing_to_check_yields_unknown_not_verified() -> None: + # A replica that has not yet pushed its identity must not read as proof. + manager = _make_manager() + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "STARTING")), VERSION + ) + + assert verified is None + assert code_verified is None + + +# ───────────────────── hole 1: same version, different files ───────────────────── + + +@pytest.mark.asyncio +async def test_a_restaged_version_is_caught_by_content_not_by_version() -> None: + manager = _make_manager() + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": "deadbeefdeadbeef"}} + } + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert verified is True, "The version string genuinely matches — that is the trap." + assert code_verified is False + + +@pytest.mark.asyncio +async def test_matching_content_verifies() -> None: + manager = _make_manager() + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": SIGNATURE}} + } + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert (verified, code_verified) == (True, True) + + +@pytest.mark.asyncio +async def test_an_app_from_an_older_worker_has_no_signature_to_compare() -> None: + # Apps recovered from a pre-0.16.6 worker carry no source_signature. That + # must read as unknown, not as a mismatch, or every one of them would warn + # on the first tick after an upgrade. + manager = _make_manager(built_app=None, source_signature=None) + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": "deadbeefdeadbeef"}} + } + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert verified is True + assert code_verified is None + + +# ───────────────────── hole 3: what the monitor does about it ───────────────────── + + +def _status(state: str = "RUNNING"): + return SimpleNamespace( + applications={APP_ID: SimpleNamespace(status=SimpleNamespace(value=state))} + ) + + +def _wire_monitor(manager: AppsManager, verdict: tuple, state: str = "RUNNING") -> list: + """Route serve.status through call_with_reconnect; record serve.delete calls.""" + deletes: list = [] + + async def call_with_reconnect(fn, *args): + name = getattr(fn, "__name__", "") + if name == "delete": + deletes.append(args[0]) + return None + return _status(state) + + manager.ray_cluster.call_with_reconnect = AsyncMock(side_effect=call_with_reconnect) + manager._verify_running_identities = AsyncMock(return_value=verdict) + return deletes + + +@pytest.mark.asyncio +async def test_a_recovered_app_with_a_version_mismatch_is_reported_not_deleted( + caplog, +) -> None: + manager = _make_manager(built_app=None) + deletes = _wire_monitor(manager, (None, False, None)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + + assert deletes == [], ( + "Deleting a recovered app strands it: _fire_redeploy has no built " + "application to resubmit, so nothing would bring it back." + ) + assert "recovered from a previous worker" in _warnings(caplog)[0] + + +@pytest.mark.asyncio +async def test_a_managed_app_with_a_version_mismatch_is_still_deleted() -> None: + # The pre-existing self-heal must not regress: an app this worker built can + # be deleted, because the next tick redeploys it from the cached built_app. + manager = _make_manager() + deletes = _wire_monitor(manager, (None, False, None)) + + await manager.monitor_applications() + + assert deletes == [APP_ID] + + +@pytest.mark.asyncio +async def test_a_content_mismatch_warns_but_never_deletes(caplog) -> None: + manager = _make_manager() + deletes = _wire_monitor(manager, (VERSION, True, False)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + + assert deletes == [], ( + "A hash disagreement that turned out to be systematic would delete the " + "app on every tick. Report only." + ) + assert "different source content" in _warnings(caplog)[0] + + +@pytest.mark.asyncio +async def test_the_identity_warning_does_not_repeat_every_tick(caplog) -> None: + # The condition persists until a human redeploys and the monitor ticks every + # ~10 s, so without de-duplication this is one line per tick for the life of + # the worker. + manager = _make_manager() + _wire_monitor(manager, (VERSION, True, False)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + for _ in range(5): + await manager.monitor_applications() + + assert len(_warnings(caplog)) == 1 + + +@pytest.mark.asyncio +async def test_the_warning_rearms_once_the_app_verifies_again(caplog) -> None: + # Suppression may not be permanent: a later, genuinely new mismatch after a + # clean period has to be visible. + manager = _make_manager() + _wire_monitor(manager, (VERSION, True, False)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + manager._verify_running_identities = AsyncMock(return_value=(VERSION, True, True)) + await manager.monitor_applications() + manager._verify_running_identities = AsyncMock( + return_value=(VERSION, True, False) + ) + await manager.monitor_applications() + + assert len(_warnings(caplog)) == 2 + + +@pytest.mark.asyncio +async def test_a_recovered_app_is_never_auto_redeployed(caplog) -> None: + # _deploy_application would only raise on the None built_app; say so once + # per attempt instead of scheduling a task that cannot succeed. + manager = _make_manager(built_app=None) + _wire_monitor(manager, (None, None, None), state="UNHEALTHY") + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + await asyncio.sleep(0) + + manager._deploy_application.assert_not_awaited() + assert "skipping auto-redeploy" in _warnings(caplog)[0] + + +@pytest.mark.asyncio +async def test_undeploying_clears_the_warning_marker() -> None: + # Otherwise an application_id redeployed under the same name would inherit + # the previous instance's suppression and never warn. + manager = _make_manager() + manager._identity_warned.add(APP_ID) + _wire_monitor(manager, (VERSION, True, True)) + + await manager.monitor_applications() + + assert APP_ID not in manager._identity_warned + + +# ───────────────────── the fingerprint both sides must agree on ───────────────────── + + +def test_the_source_hash_tracks_content_and_ignores_bytecode(tmp_path: Path) -> None: + # The submit task bakes this onto each replica and the introspect task + # returns it to the worker; the two are only comparable if they hash the + # same way, which is why there is one function rather than two copies. + source = tmp_path / "src" + (source / "pkg").mkdir(parents=True) + (source / "pkg" / "main.py").write_text("x = 1\n") + + baseline = hash_source_tree(source) + + (source / "pkg" / "__pycache__").mkdir() + (source / "pkg" / "__pycache__" / "main.cpython-311.pyc").write_bytes(b"\x00\x01") + assert hash_source_tree(source) == baseline + + (source / "pkg" / "main.py").write_text("x = 2\n") + assert hash_source_tree(source) != baseline, ( + "A changed method body leaves qualnames and schemas identical — the " + "content hash is the only thing that sees it." + ) + + +def test_the_source_hash_tracks_file_names_not_just_bytes(tmp_path: Path) -> None: + a = tmp_path / "a" + b = tmp_path / "b" + a.mkdir() + b.mkdir() + (a / "one.py").write_text("x = 1\n") + (b / "two.py").write_text("x = 1\n") + + assert hash_source_tree(a) != hash_source_tree(b)