fix: expose comparable Function runtime identity (#4975)

Expose `FunctionVersion.runtime_digest` through Rust and Python, and add
`PythonRuntimeSpec.digest` so clients can compare declared environment
inputs before registering a Function. Dependency pins, environment
variables, Python version, and GPU settings affect this digest; package
ordering does not.

Clarify that matching `source_digest` alone is insufficient for reuse.
Older servers/images leave `runtime_digest` unset, which cannot
establish a match. Both digests and the other registration inputs must
match before reusing a version.

```python
request = definition.registration_request
same_source = version.source_digest == request.artifact.digest
same_runtime = version.runtime_digest == request.runtime.digest
```

Add shared runtime digest fixtures and coverage for environment-only
changes and catalog decoding. Local builds and tests were skipped as
requested.

python/python/lancedb/functions.py

@@ -495,6 +495,35 @@ class PythonRuntimeSpec(_RemoteValue):
             object.__setattr__(self, "gpu_class", None)
         return self
 
+    @property
+    def digest(self) -> str:
+        """SHA-256 identity of the declared runtime, before building an image.
+
+        Covers Python version, environment sources, environment variables, and
+        GPU settings. Package and system-package order is ignored; conda channel
+        order is preserved. Compare with ``runtime_digest`` on
+        [FunctionVersion][lancedb.functions.FunctionVersion].
+        This identifies build inputs, not the resolved contents of unpinned
+        dependencies or the deployment's default base image.
+        """
+        if self.kind not in {"python", "python_v2"}:
+            raise ValueError(f"cannot digest unknown runtime kind {self.kind!r}")
+        value = self._known_dict()
+        if not value.get("env"):
+            value.pop("env", None)
+        environment = value["environment"]
+        for field in ("packages", "system_packages"):
+            if field in environment:
+                environment[field] = sorted(environment[field])
+        encoded = json.dumps(
+            value,
+            ensure_ascii=False,
+            allow_nan=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
+
 
 class FunctionImage(_RemoteValue):
     """A complete OCI Function image identified by its exact manifest digest."""
@@ -536,8 +565,17 @@ class FunctionVersion(_RemoteValue):
     of the built ``image``. Compare it with
     ``definition.registration_request.artifact.digest`` to check whether a local
     [FunctionDefinition][lancedb.functions.FunctionDefinition] has the same source.
+    It does not cover dependencies, environment variables, Python version, or
+    GPU settings. Equal source digests alone do not mean a version can be reused.
     It is ``None`` when the server does not provide a source digest.
 
+    ``runtime_digest`` identifies the declared environment inputs. Compare it
+    with ``definition.registration_request.runtime.digest`` to check whether
+    the local runtime matches. It is ``None`` for images built before runtime
+    identity was exposed. A missing digest cannot establish a match. Reusing a
+    version also requires matching the source and all other registration inputs,
+    including signature, Secret bindings, models, and execution controls.
+
     ``gpu_class`` is the operator-defined GPU pool fixed into this Function
     version, or ``None`` when registration used the deployment default.
 
@@ -554,6 +592,7 @@ class FunctionVersion(_RemoteValue):
     version: _ObjectVersion
     image: FunctionImage
     source_digest: Optional[str] = None
+    runtime_digest: Optional[str] = None
     signature: FunctionSignature
     secret_bindings: tuple[SecretBinding, ...] = ()
     created_at_millis: int

python/python/tests/test_first_class_function_slice1.py

@@ -34,7 +34,7 @@ FIXTURES = (
 
 
 def fixture(name: str) -> str:
-    return (FIXTURES / name).read_text()
+    return (FIXTURES / name).read_text(encoding="utf-8")
 
 
 def job_result(name: str) -> dict:
@@ -148,6 +148,38 @@ def test_function_version_identity_is_immutable_and_exact():
             FunctionVersion(**{**value, "version": invalid})
 
 
+@pytest.mark.parametrize(
+    "case", json.loads(fixture("runtime_identity.json")), ids=lambda case: case["name"]
+)
+def test_runtime_digests_match_server_golden_values(case):
+    runtime = PythonRuntimeSpec.from_json(json.dumps(case["runtime"]))
+    assert runtime.digest == case["digest"]
+    assert (
+        PythonRuntimeSpec.from_json(runtime.to_canonical_json()).digest
+        == runtime.digest
+    )
+
+
+def test_unknown_runtime_digest_fails_closed():
+    runtime = PythonRuntimeSpec.from_json('{"kind":"python_v3","future":"value"}')
+    with pytest.raises(ValueError, match="cannot digest unknown runtime kind"):
+        runtime.digest
+
+
+@pytest.mark.parametrize("runtime_digest", [None, "sha256:runtime-inputs"])
+def test_function_version_preserves_runtime_digest(runtime_digest):
+    value = job_result("remote_function_job.json")
+    legacy = FunctionVersion.from_json(json.dumps(value))
+    assert legacy.runtime_digest is None
+    assert "runtime_digest" not in json.loads(legacy.to_canonical_json())
+    value["runtime_digest"] = runtime_digest
+    version = FunctionVersion.from_json(json.dumps(value))
+    assert version.runtime_digest == runtime_digest
+    assert FunctionVersion.from_json(version.to_canonical_json()) == version
+    with pytest.raises((TypeError, ValueError)):
+        version.runtime_digest = "sha256:changed"
+
+
 @pytest.mark.parametrize("source_digest", [None, "sha256:source-artifact"])
 def test_function_version_preserves_source_digest(source_digest):
     value = job_result("remote_function_job.json")

python/python/tests/test_first_class_function_slice2.py

@@ -1891,6 +1891,29 @@ def test_annotation_and_explicit_schema_validation_fail_closed():
                 return value
 
 
+def test_default_function_runtime_matches_server_identity():
+    definition = function(python_version="3.12")(analyze_caption.__wrapped__)
+    runtime = definition.registration_request.runtime
+    assert runtime.env == {}
+    cases = json.loads((FIXTURES / "runtime_identity.json").read_text(encoding="utf-8"))
+    expected = next(
+        case["digest"] for case in cases if case["name"] == "empty-environment"
+    )
+    assert runtime.digest == expected
+
+
+def test_environment_changes_do_not_change_source_identity():
+    definition = function(
+        pip=["openai==3.46.0"],
+        env={"MODE": "test"},
+        python_version="3.12",
+    )(analyze_caption.__wrapped__)
+    previous = analyze_caption.registration_request
+    changed = definition.registration_request
+    assert previous.artifact.digest == changed.artifact.digest
+    assert previous.runtime.digest != changed.runtime.digest
+
+
 def test_local_function_catalog_operations_are_not_supported(tmp_path):
     db = lancedb.connect(tmp_path)
     message = "Function catalog operations are not supported by this database"
@@ -1950,6 +1973,7 @@ async def test_async_catalog_path_preserves_execution_controls():
         (FIXTURES / "remote_function_version.canonical.json").read_text()
     )
     definition.update(
+        runtime_digest=normalize_score.registration_request.runtime.digest,
         batch_size=12,
         row_timeout_seconds=8.0,
         result_stability="input_and_context",
@@ -1977,6 +2001,7 @@ def test_sync_catalog_path_preserves_execution_controls():
         (FIXTURES / "remote_function_version.canonical.json").read_text()
     )
     definition.update(
+        runtime_digest=normalize_score.registration_request.runtime.digest,
         batch_size=12,
         row_timeout_seconds=8.0,
         result_stability="input_and_context",
@@ -2090,6 +2115,9 @@ def _mock_remote_function_catalog():
                     )["image"],
                     "signature": body["signature"],
                     "source_digest": body["artifact"]["digest"],
+                    "runtime_digest": PythonRuntimeSpec.from_json(
+                        json.dumps(body["runtime"])
+                    ).digest,
                     "secret_bindings": body.get("secret_bindings", []),
                     "created_at_millis": 1787270400000,  # 2026-08-21T00:00:00Z
                 }

python/python/tests/test_remote_db.py

@@ -6304,6 +6304,8 @@ def test_remote_function_registration_transports_max_concurrency():
     def registered(value: float) -> float:
         return value
 
+    definition["runtime_digest"] = registered.registration_request.runtime.digest
+
     class RegistrationServer(flight.FlightServerBase):
         def __init__(self):
             super().__init__("grpc://127.0.0.1:0")
@@ -6322,6 +6324,7 @@ def test_remote_function_registration_transports_max_concurrency():
     assert len(server.tickets) == 1
     assert b'"max_concurrency":4' in server.tickets[0]
     assert version.max_concurrency == 4
+    assert version.runtime_digest == registered.registration_request.runtime.digest
 
 
 @pytest.mark.asyncio