Prepare fresh functional wheel targets through the runtime owner

src/agent_comms/build_artifact.py

@@ -0,0 +1,46 @@
+"""Stable file-content observations shared by interpreter and wheel build inputs."""
+
+from __future__ import annotations
+
+import hashlib
+from io import BufferedReader
+from pathlib import Path
+from typing import Annotated
+
+from pydantic import BaseModel, ConfigDict, field_validator
+
+from agent_comms.nominal.field_codec import Sha256Text
+from agent_comms.private_path import FileRevision
+
+
+class BuildArtifact(BaseModel):
+    """A native file identity and its bytes, captured through the same open handle."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+    path: Path
+    revision: FileRevision
+    sha256: Annotated[str, Sha256Text]
+
+    @field_validator("sha256")
+    @classmethod
+    def valid_digest(cls, value: str) -> str:
+        return Sha256Text.decode(value)
+
+    @classmethod
+    def capture(cls, path: Path) -> BuildArtifact:
+        path = path.resolve(strict=True)
+        with path.open("rb") as stream:
+            return cls.from_stream(stream)
+
+    @classmethod
+    def from_stream(cls, stream: BufferedReader) -> BuildArtifact:
+        """Capture the already-open file, including a platform-qualified native image."""
+        path = Path(stream.name).resolve(strict=True)
+        stream.seek(0)
+        revision = FileRevision.from_file(stream)
+        digest = hashlib.file_digest(stream, "sha256").hexdigest()
+        if revision != FileRevision.from_file(stream):
+            raise ValueError(f"Build artifact changed while reading: {path}")
+        if FileRevision.capture(path) != revision:
+            raise ValueError(f"Build artifact was replaced while reading: {path}")
+        return cls(path=path, revision=revision, sha256=digest)

src/agent_comms/build_identity.py

@@ -1,16 +1,17 @@
-"""Installed repository commits, shared by delivery proofs and close checks."""
+"""Installed repository and interpreter identity, shared by proofs and close checks."""
 
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from collections.abc import Iterable
 from typing import TYPE_CHECKING, Annotated, ClassVar
 
 from agent_comms.nominal.declared_family import DeclaredFamily
-from agent_comms.nominal.field_codec import GitCommitText
+from agent_comms.nominal.field_codec import GitCommitText, WireOmitDefault
 from agent_comms.installer.runtime_resolution import RuntimePackage, RuntimeResolution
+from agent_comms.installer.probes.interpreter import InterpreterBuild, RecordedInterpreterBuild
 from agent_comms.launch_runtime import LaunchRuntime
 from agent_comms.repo import Repo
 from agent_comms.repository_name import RepositoryName
@@ -21,7 +22,7 @@
 
 
 class ProofBuildMismatchError(ValueError):
-    """A retained proof does not identify the currently installed repositories."""
+    """A retained proof does not identify the installed repositories and interpreter."""
 
 
 class DeliveredCommitNotInstalledError(ValueError):
@@ -137,7 +138,7 @@ def installation(self, commits: tuple[str, ...], directory: Path) -> CommitInsta
 @dataclass(frozen=True)
 class UnrecordedBuildIdentity(BuildIdentity):
     def recorded(self) -> RecordedBuildIdentity:
-        return RecordedBuildIdentity.installed()
+        return RecordedBuildIdentity.installed().recorded()
 
     def installation(self, commits: tuple[str, ...], directory: Path) -> CommitInstallation:
         installed = RecordedBuildIdentity.installed()
@@ -148,28 +149,43 @@ def installation(self, commits: tuple[str, ...], directory: Path) -> CommitInsta
 
 @dataclass(frozen=True)
 class RecordedBuildIdentity(BuildIdentity, InstalledBuild):
-    """Declared runtime source revisions, each identified by its repository."""
+    """Repository revisions and the interpreter observation owned by their runtime."""
 
     commits: tuple[RepositoryCommit, ...]
+    interpreter: Annotated[
+        RecordedInterpreterBuild | InterpreterBuild, WireOmitDefault()
+    ] = field(default_factory=InterpreterBuild)
 
     def __str__(self) -> str:
-        return ", ".join(
-            f"{record.repository.to_wire()} {record.commit}" for record in self.commits
-        )
+        return ", ".join((
+            *(f"{record.repository.to_wire()} {record.commit}" for record in self.commits),
+            str(self.interpreter),
+        ))
 
     def recorded(self) -> RecordedBuildIdentity:
+        try:
+            self.interpreter.require_recorded()
+        except ValueError as error:
+            raise ProofBuildMismatchError(
+                f"Proof build {self} has no interpreter observation; run and record a new proof."
+            ) from error
         return self
 
     @classmethod
     def installed(cls) -> RecordedBuildIdentity:
-        runtime = LaunchRuntime.installed()
+        return cls.at(LaunchRuntime.installed())
+
+    @classmethod
+    def at(cls, runtime: LaunchRuntime) -> RecordedBuildIdentity:
+        """Read the selected build, never borrow the current pointer for a candidate."""
         runtime.require_available()
         resolution = RuntimeResolution.at(runtime.prefix)
         return cls(
             tuple(
                 RepositoryCommit(member.repository, GitCommitText.decode(member.commit(resolution)))
                 for member in RuntimePackage.members_with(RuntimePackage)
-            )
+            ),
+            interpreter=resolution.interpreter,
         )
 
     def recorded_in(self, repos: Iterable[Repo]) -> tuple[tuple[Repo, RepositoryCommit], ...]:
@@ -201,7 +217,8 @@ def contains(self, commit: str, directory: Path) -> bool:
         return Repo.descends(directory, commit, recorded.commit)
 
     def installation(self, commits: tuple[str, ...], directory: Path) -> CommitInstallation:
-        installed = self.installed()
+        self.recorded()
+        installed = self.installed().recorded()
         if self != installed:
             raise ProofBuildMismatchError(
                 f"Proof build {self} differs from installed {installed}; run the proof again."

src/agent_comms/child_process.py

@@ -1,3 +1,3 @@
 """Identity-bound child ownership, process-group retirement and bounded runs.
 
 Linux pidfd operations are lifted from compaction_child_watchdog. Process
@@ -37,9 +37,10 @@
 from contextvars import copy_context
 from dataclasses import dataclass
 from functools import cache, cached_property, partial
+from io import BufferedReader
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import IO, TYPE_CHECKING, Annotated, Any, ClassVar, NoReturn, Self
+from typing import IO, TYPE_CHECKING, Annotated, Any, BinaryIO, ClassVar, NoReturn, Self
 
 from .nominal.declared_family import DeclaredFamily
 from .errors import RelationViolationError
@@ -529,6 +530,28 @@
         finally:
             os.close(descriptor)
 
+    class NativeImage(ctypes.Structure):
+        """The POSIX loader's native symbol-image ABI."""
+
+        _fields_ = (
+            ("filename", ctypes.c_char_p),
+            ("base", ctypes.c_void_p),
+            ("symbol", ctypes.c_char_p),
+            ("address", ctypes.c_void_p),
+        )
+
+    @contextmanager
+    def native_image(self, address: ctypes.c_void_p) -> Generator[BufferedReader]:
+        """Open the native file providing a symbol; the caller consumes this handle."""
+        lookup = ctypes.CDLL(None).dladdr
+        lookup.argtypes = (ctypes.c_void_p, ctypes.POINTER(self.NativeImage))
+        lookup.restype = ctypes.c_int
+        image = self.NativeImage()
+        if not lookup(address, ctypes.byref(image)) or not image.filename:
+            raise ValueError("The loader did not identify the native symbol image")
+        with Path(os.fsdecode(image.filename)).resolve(strict=True).open("rb") as stream:
+            yield stream
+
     @abstractmethod
     def require_watched_procfs(self, refusal: Exception) -> None: ...
 
@@ -657,6 +680,30 @@
 
 
 class LinuxPlatform(PidfdHandles, Platform):
+    @contextmanager
+    def native_image(self, address: ctypes.c_void_p) -> Generator[BufferedReader]:
+        # glibc's main-image filename aliases argv[0]; setproctitle changes it.
+        # Resolve the actual symbol mapping instead, including shared libpython.
+        # ctypes represents C null as None; kernel mapping addresses are integers.
+        pointer = address.value or 0
+        if pointer == 0:
+            raise ValueError("A null address has no native image")
+        with Path("/proc/self/maps").open() as mappings:
+            for line in mappings:
+                extent, _permissions, _offset, device, inode, *_name = line.split(maxsplit=5)
+                first, stop = (int(value, 16) for value in extent.split("-"))
+                if first <= pointer < stop:
+                    link = Path("/proc/self/map_files") / f"{first:x}-{stop:x}"
+                    with link.resolve(strict=True).open("rb") as stream:
+                        observed = os.fstat(stream.fileno())
+                        major, minor = (int(value, 16) for value in device.split(":"))
+                        # Decode kernel metadata once at this native boundary.
+                        if (observed.st_dev, observed.st_ino) != (os.makedev(major, minor), int(inode)):
+                            raise ValueError("The native image file no longer matches its mapping")
+                        yield stream
+                    return
+        raise ValueError("The kernel did not identify the native symbol image")
+
     def require_watched_procfs(self, refusal: Exception) -> None:
         return None
 

src/agent_comms/cli_verbs/prepare_runtime.py

@@ -0,0 +1,54 @@
+"""Prepare a functional candidate through the existing runtime/preparation owners."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Annotated
+
+from agent_comms.cli_command import CliCommand
+from agent_comms.cli_result import CliResult
+from agent_comms.comms import Comms
+from agent_comms.installer.commands import InstallCommands
+from agent_comms.installer.runtime_stack import WheelRuntime
+from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.nominal.field_codec import Argument, FieldCodec, Normalize, PathText
+from agent_comms.shell_effects import BuildStorageEffect, WriteStorageEffect
+from agent_comms.storage_projection import StorageProjection
+from agent_comms.tool_caller import ThreadCaller
+
+
+@dataclass(frozen=True)
+class PreparedRuntimeResult(CliResult):
+    prefix: Annotated[Path, PathText]
+
+
+@dataclass(frozen=True, kw_only=True)
+class PrepareRuntimeCliCommand(CliCommand, declared_name="prepare-runtime"):
+    help = ("Prepare a NEW owned functional target from an explicit interpreter, "
+            "recorded dependency runtime and exact local wheels; never publish it")
+    caller: Annotated[ThreadCaller, Normalize(FieldCodec.encode),
+                      Argument(flags=("--from",), options={"type": ThreadCaller})]
+    into: Annotated[Path, PathText, Argument(flags=("--into",))]
+    interpreter: Annotated[Path, PathText, Argument(flags=("--python",))]
+    source_runtime: Annotated[Path, PathText, Argument(flags=("--source-runtime",))]
+    checkout: Annotated[Path, PathText, Argument(flags=("--checkout",))]
+    wheels: Annotated[tuple[Annotated[Path, PathText], ...],
+                     Argument(flags=("--wheels",), options={"nargs": "+"})]
+
+    def apply(self, ctx: Comms) -> CliResult:
+        thread = ctx.registry.require(self.caller.own_thread())
+        prefix = self.into.expanduser().absolute()
+        checkout = self.checkout.expanduser().resolve()
+        projection = StorageProjection(ctx.registry.storage_rule, thread, ctx)
+        projection.check((BuildStorageEffect(), WriteStorageEffect(prefix),
+                          WriteStorageEffect(checkout)))
+        with InstallCommands() as commands:
+            target = WheelRuntime(
+                checkout=checkout, commands=commands,
+                runtime=LaunchRuntime.at_prefix(prefix),
+                interpreter=LaunchRuntime.from_interpreter(self.interpreter.expanduser().absolute()),
+                source_runtime=LaunchRuntime.at_prefix(self.source_runtime.expanduser().resolve()),
+                wheels=tuple(path.expanduser().resolve(strict=True) for path in self.wheels),
+            )
+            target.prepare()
+        return PreparedRuntimeResult(prefix)

src/agent_comms/installer/probes/interpreter.py

@@ -0,0 +1,70 @@
+"""The selected interpreter's observed native build, not its repository ancestry."""
+
+from __future__ import annotations
+
+import ctypes
+from pathlib import Path
+import sys
+import sysconfig
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from agent_comms.installer.interpreter_operation import RuntimeProbe, TargetProbe
+from agent_comms.build_artifact import BuildArtifact
+from agent_comms.child_process import Platform
+
+
+class InterpreterBuild(BaseModel):
+    """An existing build record without an interpreter observation is unqualified."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    def __str__(self) -> str:
+        return "unrecorded interpreter"
+
+    def matches(self, actual: RecordedInterpreterBuild) -> bool:
+        """Absent evidence cannot make a cached runtime eligible for reuse."""
+        return False
+
+    def require_recorded(self) -> RecordedInterpreterBuild:
+        raise ValueError("The runtime has no recorded interpreter build")
+
+
+class RecordedInterpreterBuild(InterpreterBuild):
+    """Native artifact identity, with the invoked environment kept separately."""
+
+    invoked: Path
+    version: str = Field(min_length=1)
+    abi: str = Field(min_length=1)
+    stdlib: Path
+    artifacts: tuple[BuildArtifact, ...] = Field(min_length=1)
+
+    def __str__(self) -> str:
+        images = ", ".join(f"{artifact.path} sha256={artifact.sha256}"
+                           for artifact in self.artifacts)
+        return f"interpreter {self.invoked} [{self.abi}]: {images}"
+
+    def matches(self, actual: RecordedInterpreterBuild) -> bool:
+        return self == actual
+
+    def require_recorded(self) -> RecordedInterpreterBuild:
+        return self
+
+
+class InterpreterBuildProbe(TargetProbe, RuntimeProbe[RecordedInterpreterBuild]):
+    def run(self) -> RecordedInterpreterBuild:
+        abi = sysconfig.get_config_var("SOABI")
+        if not isinstance(abi, str) or not abi:
+            raise ValueError("The selected interpreter does not declare its native ABI")
+        address = ctypes.cast(ctypes.pythonapi.Py_GetVersion, ctypes.c_void_p)
+        with Platform.current().native_image(address) as stream:
+            native = BuildArtifact.from_stream(stream)
+        executable = Path(sys.executable).resolve(strict=True)
+        images = {native.path: native}
+        if executable not in images:
+            images[executable] = BuildArtifact.capture(executable)
+        return RecordedInterpreterBuild(
+            invoked=Path(sys.executable), version=sys.version, abi=abi,
+            stdlib=Path(sysconfig.get_path("stdlib")).resolve(strict=True),
+            artifacts=tuple(images[path] for path in dict.fromkeys((executable, native.path))),
+        )