fix(apps): key redeploy on source content and verify recovered apps

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] = {}

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,