fix(extensions): isolate projection validator runtime

loopx/extensions/presentation.py

@@ -15,10 +15,13 @@
 
 from ..file_lock import exclusive_file_lock
 from .runtime import (
+    MAX_EXTENSION_RESPONSE_BYTES,
     _resolved_active_extension,
     extension_catalog_entries,
     run_standalone_extension,
 )
+from .process_runtime import run_capped_process
+from .readiness import ResolvedRuntimeEntrypoint, runtime_process_environment
 from .manifest import validate_extension_id
 
 
@@ -73,6 +76,39 @@
     "secret",
     "token",
 }
+_CORE_VIEW_VALIDATORS = {
+    "loopx.extensions.presentation:validate_opaque_presentation_view",
+}
+
+_ISOLATED_VIEW_VALIDATOR = """\
+import importlib
+import json
+import sys
+
+request = json.load(sys.stdin)
+reference = request["reference"]
+module_name, attribute_name = reference.split(":", 1)
+try:
+    module = importlib.import_module(module_name)
+except (ImportError, ModuleNotFoundError):
+    response = {"ok": False, "status": "unavailable"}
+else:
+    validator = getattr(module, attribute_name, None)
+    if not callable(validator):
+        response = {"ok": False, "status": "not_callable"}
+    elif request["operation"] == "resolve":
+        response = {"ok": True, "status": "ready"}
+    else:
+        try:
+            response = {"ok": True, "view": validator(request["view"])}
+        except Exception as exc:
+            response = {
+                "ok": False,
+                "status": "rejected",
+                "error": str(exc)[:1000],
+            }
+json.dump(response, sys.stdout, ensure_ascii=False, separators=(",", ":"))
+"""
 
 
 def _is_forbidden_key_name(value: Any) -> bool:
@@ -288,13 +324,99 @@ def validate_opaque_presentation_view(value: Any) -> Any:
 
 def load_presentation_view_validator(
     declared_surface: Mapping[str, Any],
+    *,
+    runtime_entrypoint: ResolvedRuntimeEntrypoint | None = None,
+    timeout_seconds: int = 30,
 ) -> Callable[[Any], Any]:
     """Load the exact validator declared by one active presentation surface."""
 
     reference = declared_surface.get("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:
+        python_executable = runtime_entrypoint.python_executable
+        if python_executable is None:
+            raise ValueError(
+                f"presentation surface view_validator `{reference}` is unavailable"
+            )
+
+        def invoke_runtime_validator(
+            operation: str, *, value: Any = None
+        ) -> Mapping[str, Any]:
+            try:
+                request = json.dumps(
+                    {"operation": operation, "reference": reference, "view": value},
+                    ensure_ascii=False,
+                    separators=(",", ":"),
+                ).encode("utf-8")
+            except (TypeError, ValueError) as exc:
+                raise ValueError(
+                    "presentation surface view is not JSON serializable"
+                ) from exc
+            environment = dict(
+                runtime_process_environment(runtime_entrypoint.path_prefix) or os.environ
+            )
+            environment.pop("PYTHONPATH", None)
+            try:
+                completed = run_capped_process(
+                    [python_executable, "-I", "-c", _ISOLATED_VIEW_VALIDATOR],
+                    stdin=request,
+                    timeout_seconds=timeout_seconds,
+                    output_limit_bytes=MAX_EXTENSION_RESPONSE_BYTES,
+                    env=environment,
+                )
+            except OSError as exc:
+                raise ValueError(
+                    f"presentation surface view_validator `{reference}` is unavailable"
+                ) from exc
+            if completed.returncode != 0 or completed.failure_kind is not None:
+                raise ValueError(
+                    f"presentation surface view_validator `{reference}` is unavailable"
+                )
+            try:
+                response = json.loads(completed.stdout.decode("utf-8"))
+            except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+                raise ValueError(
+                    f"presentation surface view_validator `{reference}` returned invalid output"
+                ) from exc
+            if not isinstance(response, Mapping):
+                raise ValueError(
+                    f"presentation surface view_validator `{reference}` returned invalid output"
+                )
+            return response
+
+        resolved = invoke_runtime_validator("resolve")
+        if resolved.get("status") == "not_callable":
+            raise ValueError(
+                f"presentation surface view_validator `{reference}` is not callable"
+            )
+        if resolved.get("ok") is not True:
+            raise ValueError(
+                f"presentation surface view_validator `{reference}` is unavailable"
+            )
+
+        def validate_in_runtime(value: Any) -> Any:
+            response = invoke_runtime_validator("validate", value=value)
+            if response.get("ok") is True:
+                return response.get("view")
+            if response.get("status") == "not_callable":
+                raise ValueError(
+                    f"presentation surface view_validator `{reference}` is not callable"
+                )
+            if response.get("status") == "rejected":
+                error = response.get("error")
+                raise ValueError(
+                    error
+                    if isinstance(error, str) and error.strip()
+                    else "presentation surface view was rejected"
+                )
+            raise ValueError(
+                f"presentation surface view_validator `{reference}` is unavailable"
+            )
+
+        return validate_in_runtime
+
     try:
         module = import_module(module_name)
     except (ImportError, ModuleNotFoundError) as exc:
@@ -309,6 +431,22 @@ def load_presentation_view_validator(
     return validator
 
 
+def _runtime_presentation_view_validator(
+    declared_surface: Mapping[str, Any],
+    *,
+    runtime_entrypoint: ResolvedRuntimeEntrypoint,
+    manifest: Mapping[str, Any],
+) -> Callable[[Any], Any]:
+    runtime = manifest.get("runtime")
+    if not isinstance(runtime, Mapping):
+        raise ValueError("extension active manifest is incomplete")
+    return load_presentation_view_validator(
+        declared_surface,
+        runtime_entrypoint=runtime_entrypoint,
+        timeout_seconds=int(runtime["timeout_seconds"]),
+    )
+
+
 def validate_provider_presentation_projection(
     value: Mapping[str, Any],
     *,
@@ -502,6 +640,7 @@ def _validate_persisted_envelope(
     extension_id: str,
     revision: str,
     declared_surface: Mapping[str, Any],
+    view_validator: Callable[[Any], Any] | None = None,
 ) -> dict[str, Any]:
     record = _record(
         value,
@@ -547,6 +686,7 @@ def _validate_persisted_envelope(
             "view": record.get("view"),
         },
         declared_surface=declared_surface,
+        view_validator=view_validator,
     )
     envelope = {
         "schema_version": EXTENSION_PROJECTION_SURFACE_SCHEMA_VERSION,
@@ -635,7 +775,11 @@ def publish_extension_projection(
     if not execute:
         return receipt
 
-    view_validator = load_presentation_view_validator(surface)
+    view_validator = _runtime_presentation_view_validator(
+        surface,
+        runtime_entrypoint=verified_entrypoint,
+        manifest=manifest,
+    )
     runtime_receipt = run_standalone_extension(
         safe_extension_id,
         state_file=state_path,
@@ -725,6 +869,7 @@ def publish_extension_projection(
             extension_id=safe_extension_id,
             revision=active_revision,
             declared_surface=surface,
+            view_validator=view_validator,
         )
         if readback != envelope:
             raise ValueError("published projection readback does not match exact payload")
@@ -822,6 +967,14 @@ def collect_active_extension_presentation_surfaces(
             continue
         extension_id = str(provider.get("id") or "")
         revision = str(provider.get("active_revision") or "")
+        try:
+            resolved_revision, runtime_entrypoint, active_manifest = (
+                _resolved_active_extension(extension_id, state_file=state_path)
+            )
+        except ValueError:
+            continue
+        if resolved_revision != revision:
+            continue
         surfaces = manifest.get("presentation_surfaces")
         if not isinstance(surfaces, list):
             continue
@@ -870,11 +1023,17 @@ def collect_active_extension_presentation_surfaces(
                 )
                 continue
             try:
+                view_validator = _runtime_presentation_view_validator(
+                    surface,
+                    runtime_entrypoint=runtime_entrypoint,
+                    manifest=active_manifest,
+                )
                 envelope = _validate_persisted_envelope(
                     raw,
                     extension_id=extension_id,
                     revision=revision,
                     declared_surface=surface,
+                    view_validator=view_validator,
                 )
             except ValueError as exc:
                 items.append(
@@ -938,7 +1097,7 @@ def read_extension_projection(
         payload_sha256,
         context="payload_sha256",
     )
-    active_revision, _entrypoint, manifest = _resolved_active_extension(
+    active_revision, runtime_entrypoint, manifest = _resolved_active_extension(
         safe_extension_id,
         state_file=state_path,
     )
@@ -973,6 +1132,11 @@ def read_extension_projection(
         extension_id=safe_extension_id,
         revision=active_revision,
         declared_surface=declared_surface,
+        view_validator=_runtime_presentation_view_validator(
+            declared_surface,
+            runtime_entrypoint=runtime_entrypoint,
+            manifest=manifest,
+        ),
     )
     if envelope["payload_sha256"] != requested_hash:
         raise ValueError("payload_sha256 does not match the published projection")

loopx/extensions/readiness.py

@@ -4,6 +4,7 @@
 import importlib.util
 import json
 import os
+import shlex
 import shutil
 import subprocess
 import sys
@@ -21,6 +22,60 @@ class ResolvedRuntimeEntrypoint:
     argv_prefix: tuple[str, ...]
     identity: str
     path_prefix: str | None = None
+    python_executable: str | None = None
+
+
+def _python_executable_for_script(path: Path) -> str | None:
+    """Resolve the interpreter selected by one verified console script."""
+
+    try:
+        first_line = path.open("rb").readline(4096).decode("utf-8")
+    except (OSError, UnicodeDecodeError):
+        first_line = ""
+    selected: Path | None = None
+    if first_line.startswith("#!"):
+        try:
+            command = shlex.split(first_line[2:].strip())
+        except ValueError:
+            command = []
+        if command:
+            executable = command[0]
+            if Path(executable).name == "env":
+                candidates = [
+                    item for item in command[1:] if not item.startswith("-")
+                ]
+                executable = (
+                    shutil.which(
+                        candidates[0],
+                        path=str(path.parent)
+                        + os.pathsep
+                        + os.environ.get("PATH", os.defpath),
+                    )
+                    if candidates
+                    else None
+                ) or ""
+            selected = Path(executable).expanduser()
+            if not selected.is_absolute():
+                selected = Path(os.path.abspath(selected))
+
+    candidates = [selected] if selected is not None else []
+    # Windows console-script launchers are executable wrappers rather than
+    # text shebang scripts. Their venv interpreter remains a sibling in the
+    # same Scripts directory; the same fallback is safe for opaque POSIX
+    # launchers and fails closed for non-Python runtimes.
+    candidates.extend(
+        path.parent / name
+        for name in ("python.exe", "python3.exe", "python", "python3")
+    )
+    for candidate in candidates:
+        identified = _file_identity(candidate, executable=True)
+        if identified is None or not candidate.name.lower().startswith("python"):
+            continue
+        # Preserve the selected venv launcher path. Resolving the symlink to
+        # the base interpreter would discard pyvenv.cfg and load the wrong
+        # packages.
+        return str(candidate)
+    return None
 
 
 def runtime_process_environment(
@@ -97,6 +152,7 @@ def resolve_runtime_entrypoint(
             argv_prefix=(str(resolved[0]),),
             identity=resolved[1],
             path_prefix=str(resolved[0].parent),
+            python_executable=_python_executable_for_script(resolved[0]),
         )
 
     interpreter_path = Path(sys.executable).expanduser()
@@ -124,6 +180,7 @@ def resolve_runtime_entrypoint(
     return ResolvedRuntimeEntrypoint(
         argv_prefix=(str(interpreter_path), "-m", str(python_module)),
         identity=hashlib.sha256(serialized.encode("utf-8")).hexdigest(),
+        python_executable=str(interpreter_path),
     )