fix(extensions): bind the declared validator implementation

loopx/extensions/presentation.py

@@ -21,7 +21,11 @@
     run_standalone_extension,
 )
 from .process_runtime import run_capped_process
-from .readiness import ResolvedRuntimeEntrypoint, runtime_process_environment
+from .readiness import (
+    CORE_VIEW_VALIDATORS,
+    ResolvedRuntimeEntrypoint,
+    runtime_process_environment,
+)
 from .manifest import validate_extension_id
 
 
@@ -76,10 +80,6 @@
     "secret",
     "token",
 }
-_CORE_VIEW_VALIDATORS = {
-    "loopx.extensions.presentation:validate_opaque_presentation_view",
-}
-
 _ISOLATED_VIEW_VALIDATOR = """\
 import importlib
 import json
@@ -334,7 +334,7 @@ def load_presentation_view_validator(
     if not isinstance(reference, str) or ":" not in reference:
         raise ValueError("presentation surface has no declared view_validator")
     module_name, attribute_name = reference.split(":", 1)
-    if runtime_entrypoint is not None and reference not in _CORE_VIEW_VALIDATORS:
+    if runtime_entrypoint is not None and reference not in CORE_VIEW_VALIDATORS:
         python_executable = runtime_entrypoint.python_executable
         if python_executable is None:
             raise ValueError(

loopx/extensions/readiness.py

@@ -10,11 +10,20 @@
 import sys
 from collections.abc import Mapping
 from dataclasses import dataclass
+from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
 EXTENSION_DOCTOR_SCHEMA_VERSION = "loopx_extension_doctor_v0"
 RUNTIME_ENTRYPOINT_IDENTITY_SCHEMA_VERSION = "loopx_runtime_entrypoint_identity_v1"
+RUNTIME_EXECUTABLE_IDENTITY_SCHEMA_VERSION = "loopx_runtime_executable_identity_v1"
+
+# LoopX-owned validators execute in the LoopX process and are not extension
+# artifacts. Binding them would invalidate every installed extension whenever
+# LoopX itself is upgraded.
+CORE_VIEW_VALIDATORS = frozenset(
+    {"loopx.extensions.presentation:validate_opaque_presentation_view"}
+)
 
 
 @dataclass(frozen=True)
@@ -41,17 +50,17 @@ def _python_executable_for_script(path: Path) -> str | None:
         if command:
             executable = command[0]
             if Path(executable).name == "env":
-                candidates = [
+                command_candidates = [
                     item for item in command[1:] if not item.startswith("-")
                 ]
                 executable = (
                     shutil.which(
-                        candidates[0],
+                        command_candidates[0],
                         path=str(path.parent)
                         + os.pathsep
                         + os.environ.get("PATH", os.defpath),
                     )
-                    if candidates
+                    if command_candidates
                     else None
                 ) or ""
             selected = Path(executable).expanduser()
@@ -140,19 +149,173 @@ def resolved_entrypoint_identity(command: str) -> tuple[Path, str] | None:
     return path, identified[1]
 
 
+_RESOLVE_DECLARED_MODULE_ORIGIN = """\
+import importlib.util
+import json
+import sys
+
+try:
+    spec = importlib.util.find_spec(sys.argv[1])
+except Exception:
+    spec = None
+origin = getattr(spec, "origin", None) if spec is not None else None
+json.dump({"origin": origin if isinstance(origin, str) else None}, sys.stdout)
+"""
+
+
+def declared_view_validators(manifest: Mapping[str, Any]) -> tuple[str, ...]:
+    """List the extension-owned validator references one manifest declares."""
+
+    references = {
+        str(surface["view_validator"])
+        for surface in manifest.get("presentation_surfaces") or []
+        if isinstance(surface, Mapping)
+        and isinstance(surface.get("view_validator"), str)
+        and str(surface["view_validator"]) not in CORE_VIEW_VALIDATORS
+    }
+    return tuple(sorted(references))
+
+
+@lru_cache(maxsize=64)
+def _declared_module_origin(
+    python_executable: str,
+    module_name: str,
+) -> str | None:
+    """Resolve one declared module's source file in the runtime interpreter.
+
+    The isolated validator imports this module in a ``python -I`` process, so
+    the identity owner asks that same interpreter where the implementation lives
+    instead of guessing a site-packages layout or reusing LoopX's own import
+    state. Only the resolved path is memoized; the artifact bytes are re-read on
+    every identity computation so a content mutation still changes the identity.
+    """
+
+    try:
+        completed = subprocess.run(
+            [
+                python_executable,
+                "-I",
+                "-c",
+                _RESOLVE_DECLARED_MODULE_ORIGIN,
+                module_name,
+            ],
+            stdin=subprocess.DEVNULL,
+            capture_output=True,
+            timeout=30,
+            check=False,
+            text=True,
+            encoding="utf-8",
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    if completed.returncode != 0:
+        return None
+    try:
+        payload = json.loads(completed.stdout)
+    except json.JSONDecodeError:
+        return None
+    origin = payload.get("origin") if isinstance(payload, Mapping) else None
+    return origin if isinstance(origin, str) and origin else None
+
+
+def _declared_validator_artifacts(
+    python_executable: str,
+    view_validators: tuple[str, ...],
+) -> dict[str, str | None]:
+    """Bind every declared validator implementation the runtime can resolve.
+
+    A validator that resolves has its implementation bytes bound, so replacing
+    the decision code invalidates the doctor identity even when the launcher,
+    the interpreter and the declared reference are unchanged. A validator the
+    runtime interpreter cannot resolve cannot execute either - the isolated
+    runner resolves the same module with the same interpreter and flags - so its
+    marker records that state instead of refusing the provider runtime
+    readiness. Resolving the module file deliberately does not import it: an
+    implementation that resolves but raises on import is an execution failure at
+    the surface that uses it, not a provider-runtime identity change.
+    """
+
+    artifacts: dict[str, str | None] = {}
+    for reference in view_validators:
+        module_name = reference.split(":", 1)[0]
+        origin = _declared_module_origin(python_executable, module_name)
+        artifact = (
+            None
+            if origin is None
+            else _file_identity(Path(origin), executable=False)
+        )
+        artifacts[module_name] = None if artifact is None else artifact[1]
+    return artifacts
+
+
+def _runtime_executable_identity(
+    entrypoint_identity: str,
+    python_executable: str | None,
+    *,
+    view_validators: tuple[str, ...] = (),
+) -> str | None:
+    """Bind every executable artifact one runtime selects for its provider.
+
+    A runtime that declares no extension-owned validator keeps the identity of
+    its launcher and selected interpreter alone, so an extension whose code did
+    not change does not have to be re-doctored. A runtime that does declare one
+    binds the implementation each reference resolves to, because that code runs
+    in the runtime interpreter and outside LoopX's own verified launcher.
+    """
+
+    if python_executable is None and not view_validators:
+        return entrypoint_identity
+    interpreter_identity: str | None = None
+    if python_executable is not None:
+        interpreter = _file_identity(Path(python_executable), executable=True)
+        if interpreter is None:
+            return None
+        interpreter_identity = interpreter[1]
+    identity_payload: dict[str, Any] = {
+        "schema_version": RUNTIME_EXECUTABLE_IDENTITY_SCHEMA_VERSION,
+        "kind": "executable_runtime",
+        "entrypoint_identity": entrypoint_identity,
+        "python_interpreter_identity": interpreter_identity,
+    }
+    if view_validators:
+        identity_payload["view_validator_artifacts"] = (
+            _declared_validator_artifacts(python_executable, view_validators)
+            if python_executable is not None
+            # No interpreter means no isolated validator can run; record the
+            # declared references so adding one later still changes the identity.
+            else {reference.split(":", 1)[0]: None for reference in view_validators}
+        )
+    serialized = json.dumps(
+        identity_payload,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+
+
 def resolve_runtime_entrypoint(
     runtime: Mapping[str, Any],
+    *,
+    view_validators: tuple[str, ...] = (),
 ) -> ResolvedRuntimeEntrypoint | None:
     python_module = runtime.get("python_module")
     if python_module is None:
         resolved = resolved_entrypoint_identity(str(runtime["entrypoint"]))
         if resolved is None:
             return None
+        python_executable = _python_executable_for_script(resolved[0])
+        identity = _runtime_executable_identity(
+            resolved[1],
+            python_executable,
+            view_validators=view_validators,
+        )
+        if identity is None:
+            return None
         return ResolvedRuntimeEntrypoint(
             argv_prefix=(str(resolved[0]),),
-            identity=resolved[1],
+            identity=identity,
             path_prefix=str(resolved[0].parent),
-            python_executable=_python_executable_for_script(resolved[0]),
+            python_executable=python_executable,
         )
 
     interpreter_path = Path(sys.executable).expanduser()
@@ -172,6 +335,11 @@ def resolve_runtime_entrypoint(
         "module": str(python_module),
         "module_identity": module[1],
     }
+    if view_validators:
+        identity_payload["view_validator_artifacts"] = _declared_validator_artifacts(
+            str(interpreter_path),
+            view_validators,
+        )
     serialized = json.dumps(
         identity_payload,
         sort_keys=True,
@@ -190,7 +358,11 @@ def extension_doctor(
     execute: bool = False,
 ) -> dict[str, Any]:
     runtime = extension_runtime(manifest)
-    identity_before = resolve_runtime_entrypoint(runtime)
+    view_validators = declared_view_validators(manifest)
+    identity_before = resolve_runtime_entrypoint(
+        runtime,
+        view_validators=view_validators,
+    )
     available = identity_before is not None
     doctor_args = [str(value) for value in runtime.get("doctor_args") or []]
     status = "ready" if available else "entrypoint_missing"
@@ -227,7 +399,10 @@ def extension_doctor(
             available = False
             failure_kind = failure_kind or "probe_nonzero_exit"
         else:
-            identity_after = resolve_runtime_entrypoint(runtime)
+            identity_after = resolve_runtime_entrypoint(
+                runtime,
+                view_validators=view_validators,
+            )
             if (
                 identity_after is None
                 or identity_after.identity != identity_before.identity

loopx/extensions/runtime.py

@@ -18,6 +18,7 @@
 from .readiness import (
     EXTENSION_DOCTOR_SCHEMA_VERSION,
     ResolvedRuntimeEntrypoint,
+    declared_view_validators,
     extension_runtime,
     resolve_runtime_entrypoint,
     runtime_process_environment,
@@ -561,7 +562,10 @@ def _verified_entrypoint(
     if not isinstance(manifest, Mapping):
         return None
     runtime = located_runtime(manifest, snapshot.get("entrypoint_path"))
-    identity = resolve_runtime_entrypoint(runtime)
+    identity = resolve_runtime_entrypoint(
+        runtime,
+        view_validators=declared_view_validators(manifest),
+    )
     if identity is None or identity.identity != entry.get(
         "doctor_verified_entrypoint_identity"
     ):