diff --git a/src/agent_comms/build_artifact.py b/src/agent_comms/build_artifact.py
new file mode 100644
index 0000000000..918ed70d48
--- /dev/null
+++ b/src/agent_comms/build_artifact.py
@@ -0,0 +1,45 @@
+"""Stable file-content observations shared by interpreter and wheel build inputs."""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Annotated, BinaryIO
+
+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: BinaryIO) -> 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)
diff --git a/src/agent_comms/build_identity.py b/src/agent_comms/build_identity.py
index 548b393186..0a49479c9e 100644
--- a/src/agent_comms/build_identity.py
+++ b/src/agent_comms/build_identity.py
@@ -1,15 +1,16 @@
-"""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 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
@@ -20,7 +21,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):
@@ -132,7 +133,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()
@@ -143,28 +144,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 for_repository(self, repository: Repo) -> RepositoryCommit:
@@ -192,7 +208,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."
diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py
index 3c5fcd61ca..a86142460d 100644
--- a/src/agent_comms/child_process.py
+++ b/src/agent_comms/child_process.py
@@ -39,7 +39,7 @@
 from functools import cache, cached_property, partial
 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 +529,28 @@ def sync_directory(self, path: Path) -> None:
         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[BinaryIO]:
+        """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 +679,29 @@ class LinuxProcessRecord:
 
 
 class LinuxPlatform(PidfdHandles, Platform):
+    @contextmanager
+    def native_image(self, address: ctypes.c_void_p) -> Generator[BinaryIO]:
+        # glibc's main-image filename aliases argv[0]; setproctitle changes it.
+        # Resolve the actual symbol mapping instead, including shared libpython.
+        pointer = address.value
+        if pointer is None:
+            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
 
diff --git a/src/agent_comms/cli_verbs/prepare_runtime.py b/src/agent_comms/cli_verbs/prepare_runtime.py
new file mode 100644
index 0000000000..82fefd1bbe
--- /dev/null
+++ b/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)
diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py
index 8ae15bf207..cb4bd74f47 100644
--- a/src/agent_comms/commander/order.py
+++ b/src/agent_comms/commander/order.py
@@ -1509,7 +1509,10 @@ class InstallVerb(OrderVerb):
     clears = (InstallDue,)
 
     def run(self, ctx: Commander) -> None:
-        for due in InstallDue.due(ctx.comms, ProgressRead()):
+        pulls = ProgressRead()
+        InstallDue.prepare(ctx.comms, ctx.commander,
+                           HeldOrders.read(ctx.comms, ctx.commander), pulls)
+        for due in InstallDue.due(ctx.comms, pulls):
             print(f"{due.repo.to_wire()} main {due.main[:12]}: {due.launch(ctx.comms)}")
 
 
diff --git a/src/agent_comms/installer/probes/interpreter.py b/src/agent_comms/installer/probes/interpreter.py
new file mode 100644
index 0000000000..bedc64a414
--- /dev/null
+++ b/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))),
+        )
diff --git a/src/agent_comms/installer/probes/packages.py b/src/agent_comms/installer/probes/packages.py
index f924c841d8..0a93dfc131 100644
--- a/src/agent_comms/installer/probes/packages.py
+++ b/src/agent_comms/installer/probes/packages.py
@@ -2,18 +2,25 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import sys
 from dataclasses import dataclass
-from importlib.metadata import distributions
-from pathlib import Path
+from importlib.metadata import Distribution, distributions
+from pathlib import Path, PurePosixPath
+from urllib.parse import unquote, urlsplit
+from zipfile import ZipFile
 from typing import Self
 
-from pydantic import BaseModel, JsonValue, RootModel
+from packaging.version import Version
+from pydantic import BaseModel, Field, JsonValue, RootModel
 from pydantic.json_schema import DEFAULT_REF_TEMPLATE
 
 import agent_comms
-from agent_comms.nominal.field_codec import WireValue
+from agent_comms.build_artifact import BuildArtifact
+from agent_comms.nominal.field_codec import Sha256Text, WireValue
+from agent_comms.private_path import FileRevision
+from agent_comms.launch_runtime import LaunchRuntime
 from agent_comms.installer.interpreter_operation import RuntimeProbe, TargetProbe
 from agent_comms.package_stamp import STAMP
 from agent_comms.runtime_preparation import RuntimePreparation
@@ -42,22 +49,100 @@ class GitPin(PackagePin):
 @dataclass(frozen=True)
 class ArchivePin(PackagePin):
     url: str
+    hashes: tuple[tuple[str, str], ...] = ()
 
 
 class CommitRecord(BaseModel):
     commit_id: str
 
 
-class SourceRecord(BaseModel):
+class WheelSource:
+    """Only an archive source can prove installed bytes against a supplied wheel."""
+
+    def verified_wheel(
+        self, wheel: Path, distribution: Distribution, runtime: LaunchRuntime
+    ) -> ArchiveRecord:
+        raise ValueError("A supplied wheel requires an archive installation origin")
+
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        return ()
+
+
+class SourceRecord(BaseModel, WheelSource):
     url: str
 
     def revision(self) -> str:
         raise ValueError("The installed package does not record a Git revision")
 
 
+class ArchiveInfo(BaseModel):
+    """PEP610's algorithm-to-digest map; an empty map supplies no content proof."""
+
+    hashes: dict[str, str] = Field(default_factory=dict)
+
+    def wheel_inputs(self, url: str) -> tuple[Path, ...]:
+        if "sha256" not in self.hashes:
+            return ()
+        Sha256Text.decode(self.hashes["sha256"])
+        source = urlsplit(url)
+        # Remote archives retain their PEP610 hashes, not a request to fetch them.
+        # Only local wheel inputs can be re-observed without a network operation.
+        if source.scheme != "file" or Path(source.path).suffix != ".whl":
+            return ()
+        if source.netloc not in ("", "localhost"):
+            raise ValueError("Runtime wheel input names a foreign file host")
+        wheel = Path(unquote(source.path))
+        if not wheel.is_absolute():
+            raise ValueError("Verified runtime archive is not an absolute wheel input")
+        return (wheel,)
+
+
 class ArchiveRecord(SourceRecord):
+    archive_info: ArchiveInfo = Field(default_factory=ArchiveInfo)
+
     def pin(self, version: str) -> PackagePin:
-        return ArchivePin(version, self.url)
+        return ArchivePin(version, self.url, tuple(sorted(self.archive_info.hashes.items())))
+
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        return self.archive_info.wheel_inputs(self.url)
+
+    def verified_wheel(
+        self, wheel: Path, distribution: Distribution, runtime: LaunchRuntime
+    ) -> ArchiveRecord:
+        artifact = BuildArtifact.capture(wheel)
+        if self.url != artifact.path.as_uri():
+            raise ValueError(f"Installed archive origin differs from supplied wheel: {wheel}")
+        with ZipFile(artifact.path) as archive:
+            entries = archive.infolist()
+            names = tuple(entry.filename for entry in entries)
+            records = tuple(name for name in names if name.endswith(".dist-info/RECORD"))
+            if len(set(names)) != len(names) or len(records) != 1:
+                raise ValueError("A runtime wheel requires unique entries and one RECORD")
+            (record,) = records
+            for entry in entries:
+                relative = PurePosixPath(entry.filename)
+                if (relative.is_absolute() or ".." in relative.parts
+                        or relative.as_posix() != entry.filename.rstrip("/")):
+                    raise ValueError(f"Unsafe wheel member: {entry.filename}")
+                if relative.parts[0].endswith(".data"):
+                    raise ValueError("Exact runtime inputs require root-layout wheels")
+                if entry.is_dir() or entry.filename == record:
+                    continue
+                installed = distribution.locate_file(entry.filename).resolve(strict=True)
+                if not installed.is_relative_to(runtime.prefix):
+                    raise ValueError(f"Installed wheel member escapes its runtime: {installed}")
+                with archive.open(entry) as source:
+                    expected = hashlib.file_digest(source, "sha256").hexdigest()
+                if BuildArtifact.capture(installed).sha256 != expected:
+                    raise ValueError(f"Installed bytes differ from supplied wheel: {installed}")
+        if FileRevision.capture(artifact.path) != artifact.revision:
+            raise ValueError(f"Supplied wheel changed during verification: {wheel}")
+        known = self.archive_info.hashes
+        if "sha256" in known and known["sha256"] != artifact.sha256:
+            raise ValueError(f"Archive content differs from its recorded hash: {wheel}")
+        return self.model_copy(update={"archive_info": ArchiveInfo(
+            hashes={**known, "sha256": artifact.sha256}
+        )})
 
 
 class GitRecord(SourceRecord):
@@ -70,7 +155,7 @@ def pin(self, version: str) -> PackagePin:
         return GitPin(version, self.vcs_info.commit_id)
 
 
-class RegistryRecord(RootModel[None]):
+class RegistryRecord(RootModel[None], WheelSource):
     """null is the build record's declared registry-package representation."""
 
     def revision(self) -> str:
@@ -101,20 +186,36 @@ def schema(
         return cls.model_json_schema(by_alias=by_alias, ref_template=ref_template)
 
 
-class DistributionVersionsProbe(TargetProbe, RuntimeProbe[dict[str, str]]):
-    def run(self) -> dict[str, str]:
+class DistributionProbe(TargetProbe):
+    """The selected interpreter's distribution inventory, rejecting duplicate identities."""
+
+    @staticmethod
+    def distributions() -> dict[str, Distribution]:
         from packaging.utils import canonicalize_name
 
-        return {canonicalize_name(dist.metadata["Name"]): dist.version for dist in distributions()}
+        installed = tuple(distributions())
+        selected = {canonicalize_name(dist.name): dist for dist in installed}
+        if len(selected) != len(installed):
+            raise ValueError("The runtime contains duplicate distribution identities")
+        return selected
+
+
+class DistributionVersionsProbe(DistributionProbe, RuntimeProbe[dict[str, str]]):
+    def run(self) -> dict[str, str]:
+        return {name: dist.version for name, dist in self.distributions().items()}
 
 
 @dataclass(frozen=True)
-class InstalledPackagesProbe(TargetProbe, RuntimeProbe[dict[str, PackageRecord]]):
+class InstalledPackagesProbe(DistributionProbe, RuntimeProbe[dict[str, PackageRecord]]):
     core: str
 
-    def run(self) -> dict[str, PackageRecord]:
-        from packaging.utils import canonicalize_name
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        """A built donor re-observes its own inputs without changing the request wire."""
+        from agent_comms.installer.runtime_resolution import RuntimeResolution
+
+        return RuntimeResolution.at(LaunchRuntime.running().prefix).wheel_inputs()
 
+    def run(self) -> dict[str, PackageRecord]:
         if not Path(agent_comms.__file__).resolve().is_relative_to(Path(sys.prefix).resolve()):
             raise ValueError("Core is editable, not a built runtime package")
         actual = STAMP.commit()
@@ -123,8 +224,9 @@ def run(self) -> dict[str, PackageRecord]:
                 "The wheel stamp does not name the selected source: "
                 f"installed {actual}, requested {self.core}"
             )
-        return {
-            canonicalize_name(dist.name): PackageRecord.from_wire(
+        selected = self.distributions()
+        records = {
+            name: PackageRecord.from_wire(
                 {
                     "version": dist.version,
                     "source": (
@@ -134,5 +236,28 @@ def run(self) -> dict[str, PackageRecord]:
                     ),
                 }
             )
-            for dist in distributions()
+            for name, dist in selected.items()
         }
+        from agent_comms.installer.runtime_resolution import ProjectMetadata
+
+        for wheel in self.wheel_inputs():
+            metadata = ProjectMetadata.from_wheel(wheel)
+            record = records[metadata.name]
+            if Version(record.version) != metadata.version:
+                raise ValueError(f"Installed version differs from supplied wheel: {wheel}")
+            records[metadata.name] = record.model_copy(update={
+                "source": record.source.verified_wheel(
+                    wheel, selected[metadata.name], LaunchRuntime.running()
+                )
+            })
+        return records
+
+
+@dataclass(frozen=True)
+class WheelPackagesProbe(InstalledPackagesProbe):
+    """Unrecorded targets supply inputs before their first resolution can be saved."""
+
+    wheels: tuple[Path, ...]
+
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        return self.wheels
diff --git a/src/agent_comms/installer/probes/resolution.py b/src/agent_comms/installer/probes/resolution.py
new file mode 100644
index 0000000000..7fefde6838
--- /dev/null
+++ b/src/agent_comms/installer/probes/resolution.py
@@ -0,0 +1,24 @@
+"""One target-owned observation of package inputs and the interpreter running them."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_comms.installer.interpreter_operation import RuntimeProbe, TargetProbe
+from agent_comms.installer.probes.interpreter import InterpreterBuildProbe
+from agent_comms.installer.probes.packages import WheelPackagesProbe
+from agent_comms.installer.runtime_resolution import RuntimeResolution
+
+
+@dataclass(frozen=True)
+class RuntimeResolutionProbe(TargetProbe, RuntimeProbe[RuntimeResolution]):
+    core: str
+    wheels: tuple[Path, ...] = ()
+
+    def run(self) -> RuntimeResolution:
+        return RuntimeResolution(
+            core_commit=self.core,
+            packages=WheelPackagesProbe(self.core, self.wheels).run(),
+            interpreter=InterpreterBuildProbe().run(),
+        )
diff --git a/src/agent_comms/installer/runtime_resolution.py b/src/agent_comms/installer/runtime_resolution.py
index f0f3ed9bb3..1cf43cd765 100644
--- a/src/agent_comms/installer/runtime_resolution.py
+++ b/src/agent_comms/installer/runtime_resolution.py
@@ -31,10 +31,14 @@
 from agent_comms.nominal.field_codec import FieldCodec, GitCommitText, boundary_decoder
 from agent_comms.installer.commands import InstallCommands
 from agent_comms.installer.interpreter_operation import RuntimeProbe
+from agent_comms.installer.probes.interpreter import (
+    InterpreterBuild,
+    InterpreterBuildProbe,
+    RecordedInterpreterBuild,
+)
 from agent_comms.installer.probes.packages import (
     DistributionVersionsProbe,
     GitPin,
-    InstalledPackagesProbe,
     PackagePin,
     PackageRecord,
     PreparePackageProbe,
@@ -636,6 +640,11 @@ class RuntimePackage(DeclaredFamily, affix="Package"):
     def commit(cls, resolution: RuntimeResolution) -> str:
         return resolution.packages[cls.distribution].source.revision()
 
+    @classmethod
+    def pin(cls, resolution: RuntimeResolution) -> PackagePin:
+        """Package origin owns its pin; an archive is not a repository revision."""
+        return resolution.packages[cls.distribution].pin()
+
     @classmethod
     def preparations(cls) -> tuple[RuntimeProbe[None], ...]:
         """Only packages declaring requirements add target-runtime operations."""
@@ -654,6 +663,11 @@ class CorePackage(RuntimePackage, ConsoleScriptPackage):
     def commit(cls, resolution: RuntimeResolution) -> str:
         return resolution.core_commit
 
+    @classmethod
+    def pin(cls, resolution: RuntimeResolution) -> PackagePin:
+        """Core's built wheel carries its source stamp separately from its URL."""
+        return GitPin(resolution.packages[cls.distribution].version, cls.commit(resolution))
+
 
 class ToadPackage(RuntimePackage, ConsoleScriptPackage):
     distribution = "batrachian-toad"
@@ -842,9 +856,13 @@ class UnbuiltRuntime:
     def matches(self, lock: Path, stack: RuntimeStack, core: str) -> bool:
         return False
 
-    def prepare(self, stack: RuntimeStack) -> None:
+    def prepare(self, stack: RuntimeStack, core: str) -> None:
         stack.sync()
 
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        """Only the explicit-input stack can supply wheels before first capture."""
+        return ()
+
     def complete(self, stack: RuntimeStack, core: str) -> None:
         """Link the native package, record the resolution and prove the client imports; a
         failure leaves no prefix, so the next build syncs it afresh."""
@@ -860,6 +878,16 @@ def complete(self, stack: RuntimeStack, core: str) -> None:
 class RuntimeResolution(BaseModel):
     core_commit: str
     packages: dict[str, PackageRecord]
+    interpreter: RecordedInterpreterBuild | InterpreterBuild = Field(
+        default_factory=InterpreterBuild
+    )
+
+    def require_runtime(self, actual: RuntimeResolution) -> None:
+        """Admit observed inputs, never turn an old missing observation into proof."""
+        self.interpreter.require_recorded()
+        actual.interpreter.require_recorded()
+        if self != actual:
+            raise UpgradeRefused("Observed runtime differs from its recorded build inputs")
 
     @classmethod
     def record_path(cls) -> Path:
@@ -879,19 +907,30 @@ def recorded(cls, prefix: Path) -> RuntimeResolution | UnbuiltRuntime:
             return UnbuiltRuntime(prefix)
         return cls.at(prefix)
 
-    def prepare(self, stack: RuntimeStack) -> None:
-        """A recorded runtime is already synced; ``complete`` checks it still matches."""
-
-    def complete(self, stack: RuntimeStack, core: str) -> None:
+    def prepare(self, stack: RuntimeStack, core: str) -> None:
+        """Production-cache reuse must still satisfy the newly resolved source lock."""
         if not self.matches(stack.lock, stack, core):
             raise UpgradeRefused("Existing built runtime differs from its resolved inputs")
+
+    def complete(self, stack: RuntimeStack, core: str) -> None:
+        """Completion proves recorded inputs, not eligibility for a different build."""
+        self.require_runtime(self.capture(stack, core))
         stack.stage_native()
         stack.prepare_runtime()
 
     @classmethod
     def capture(cls, stack: RuntimeStack, core: str) -> RuntimeResolution:
-        packages = stack.probe(InstalledPackagesProbe(core))
-        return cls(core_commit=core, packages=packages)
+        from agent_comms.installer.probes.resolution import RuntimeResolutionProbe
+
+        return stack.probe(RuntimeResolutionProbe(core, stack.wheel_inputs()))
+
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        """Only content-qualified archive records request another wheel observation."""
+        return tuple(
+            wheel
+            for package in self.packages.values()
+            for wheel in package.source.wheel_inputs()
+        )
 
     def save(self, prefix: Path) -> None:
         path = prefix / self.record_path().name
@@ -902,14 +941,13 @@ def pins(self) -> dict[str, PackagePin]:
         result: dict[str, PackagePin] = {
             canonicalize_name(name): record.pin() for name, record in self.packages.items()
         }
-        # Wheel URLs locate build output, not the source revisions recorded by this owner.
         for member in RuntimePackage.members_with(RuntimePackage):
-            result[member.distribution] = GitPin(
-                self.packages[member.distribution].version, member.commit(self)
-            )
+            result[member.distribution] = member.pin(self)
         return result
 
     def matches(self, lock: Path, stack: RuntimeStack, core: str) -> bool:
+        if not self.interpreter.matches(stack.probe(InterpreterBuildProbe())):
+            return False
         resolved = ResolvedLock.read(lock)
         print(f"install: real lock decodes {len(resolved.package)} packages", file=sys.stderr)
         # sync --locked has already selected the platform, groups and extras. Read its
diff --git a/src/agent_comms/installer/runtime_stack.py b/src/agent_comms/installer/runtime_stack.py
index 5a7ba7f77b..e73d72e549 100644
--- a/src/agent_comms/installer/runtime_stack.py
+++ b/src/agent_comms/installer/runtime_stack.py
@@ -13,6 +13,7 @@
 from pathlib import Path
 
 from agent_comms.child_process import ResidentManager
+from agent_comms.build_artifact import BuildArtifact
 from agent_comms.installer.carry import STEPS
 from agent_comms.installer.commands import InstallCommands
 from agent_comms.installer.declared_uv import DeclaredUv
@@ -26,8 +27,11 @@
 from agent_comms.installer.practice import RuntimeHome
 from agent_comms.installer.probes.imports import ClientImportsProbe
 from agent_comms.installer.probes.native import PrepareNativeProbe, PrepareProducerDigestsProbe
+from agent_comms.installer.probes.packages import InstalledPackagesProbe
 from agent_comms.installer.refusals import UpgradeRefused
-from agent_comms.installer.runtime_resolution import ResolvedLock, RuntimePackage, RuntimeResolution
+from agent_comms.installer.runtime_resolution import (
+    ConsoleScriptPackage, ProjectMetadata, ResolvedLock, RuntimePackage, RuntimeResolution,
+)
 from agent_comms.launch_runtime import LaunchRuntime, RestartEnvironment
 from agent_comms.native_build_inputs import NativeBuildInputs
 from agent_comms.native_package import NativePackageStore, pinned_package_for
@@ -114,7 +118,7 @@ def build(checkout: Checkout, commands: InstallCommands, home: Path = HOME) -> T
             built = source.extended(
                 TargetRuntime, runtime=LaunchRuntime.at_prefix(prefix),
                 program=TargetProgram(checkout.path / STEPS.parent))
-            RuntimeResolution.recorded(prefix).prepare(built)
+            RuntimeResolution.recorded(prefix).prepare(built, core)
             return built
 
     def sync(self) -> None:
@@ -131,8 +135,11 @@ def sync(self) -> None:
             raise
 
     def complete(self) -> None:
-        """Finish the synced runtime from inside it: its native package, recorded resolution
-        and import check, or the check that a recorded one still matches its inputs."""
+        """Finish inside this runtime, or re-observe its recorded packages and interpreter.
+
+        Production-cache selection checks the source lock in build/prepare; completion
+        also admits exact explicit-wheel inputs without claiming their Git ancestry.
+        """
         with BuildPhase().running(self.commands.clock):
             core = self.commands.git(self.checkout, "rev-parse", "HEAD").strip()
             RuntimeResolution.recorded(self.runtime.prefix).complete(self, core)
@@ -162,6 +169,10 @@ def probe[Answer](self, probe: RuntimeProbe[Answer]) -> Answer:
     def imports(self) -> None:
         self.probe(ClientImportsProbe())
 
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        """Re-observe retained inputs from the record; unbuilt lock targets have none."""
+        return RuntimeResolution.recorded(self.runtime.prefix).wheel_inputs()
+
     def stage_native(self) -> tuple[Path, Path]:
         """Prepare a missing artifact once; the checkout holds only its store link."""
         manifest = self.checkout / "stack/pi-native.sha256"
@@ -208,3 +219,70 @@ class TargetRuntime(RuntimeStack):
     """A runtime built from the target tree: its worker declares the tree's carry steps."""
 
     program: TargetProgram = field(kw_only=True)
+
+
+@dataclass(frozen=True, kw_only=True)
+class WheelRuntime(RuntimeStack):
+    """A fresh functional target: explicit Python, retained dependencies, exact wheels.
+
+    This never enters the production Core/lock reuse cache. Completion, native
+    preparation, package capture, imports and worker retirement stay on RuntimeStack.
+    Local-wheel origins remain archive origins, not invented repository revisions.
+    """
+
+    interpreter: LaunchRuntime
+    source_runtime: LaunchRuntime
+    wheels: tuple[Path, ...]
+
+    def wheel_inputs(self) -> tuple[Path, ...]:
+        """The new runtime retains its own exact inputs; borrowed originals stay read-only."""
+        return tuple(self.runtime.prefix / "inputs" / wheel.name for wheel in self.wheels)
+
+    def sync(self) -> None:
+        named_prefix = self.runtime.interpreter.parent.parent
+        if named_prefix.exists() or named_prefix.is_symlink():
+            raise UpgradeRefused(f"Preserve existing candidate: {named_prefix}")
+        prefix = self.runtime.prefix
+        self.interpreter.require_available()
+        self.source_runtime.require_available()
+        metadata = tuple(ProjectMetadata.from_wheel(wheel) for wheel in self.wheels)
+        selected = {project.name for project in metadata}
+        registered = {member.distribution for member in RuntimePackage.members_with(RuntimePackage)}
+        required = {member.distribution for member in RuntimePackage.members_with(ConsoleScriptPackage)}
+        if len(selected) != len(metadata) or not required <= selected <= registered:
+            raise UpgradeRefused("Provide one exact wheel for each console-script package; "
+                                 "only declared runtime packages may be replaced")
+        declared = DeclaredUv.of(self.checkout / "pyproject.toml", self.uv)
+        if not declared.accepts(self.commands.run(self.uv, "--version")):
+            raise UpgradeRefused("Candidate preparation requires Core's existing declared uv")
+        source = RuntimeResolution.at(self.source_runtime.prefix)
+        donor = RuntimeStack(self.checkout, self.commands, self.home, runtime=self.source_runtime)
+        actual = donor.probe(InstalledPackagesProbe(source.core_commit))
+        if actual != source.packages:
+            raise UpgradeRefused("Dependency source differs from its recorded packages")
+        artifacts = tuple(BuildArtifact.capture(wheel) for wheel in self.wheels)
+        self.commands.run(self.uv, "--offline", "venv", "--python",
+                          self.interpreter.interpreter, prefix)
+        if self.runtime.interpreter.resolve() != self.interpreter.interpreter.resolve():
+            raise UpgradeRefused("Prepared target did not select the requested interpreter")
+        (prefix / "inputs").mkdir(mode=0o700)
+        for artifact, selected in zip(artifacts, self.wheel_inputs(), strict=True):
+            shutil.copyfile(artifact.path, selected)
+            selected.chmod(0o400)
+            if BuildArtifact.capture(selected).sha256 != artifact.sha256:
+                raise UpgradeRefused(f"Retained wheel differs from its input: {artifact.path}")
+        shutil.copytree(self.source_runtime.purelib, self.runtime.purelib,
+                        dirs_exist_ok=True, symlinks=True)
+        # Prefix data is copied intact; package preparation validates its native assets.
+        shutil.copytree(self.source_runtime.prefix / "share", prefix / "share", symlinks=True)
+        self.commands.run(self.uv, "--offline", "pip", "install", "--no-deps", "--reinstall",
+                          "--python", self.python, *self.wheel_inputs(), env=self.environment())
+        self.commands.run(self.uv, "--offline", "pip", "check", "--python", self.python,
+                          env=self.environment())
+
+    def prepare(self) -> RuntimeResolution:
+        """Prepare once; a retained path is never silently rebound or reused."""
+        with BuildPhase().running(self.commands.clock):
+            self.sync()
+            self.complete()
+            return RuntimeResolution.at(self.runtime.prefix)
diff --git a/src/agent_comms/launch_runtime.py b/src/agent_comms/launch_runtime.py
index c5fbf35524..d9b72d9bb7 100644
--- a/src/agent_comms/launch_runtime.py
+++ b/src/agent_comms/launch_runtime.py
@@ -112,11 +112,15 @@ def at_prefix(cls, prefix: Path) -> Self:
     def prefix(self) -> Path:
         return self.interpreter.parent.parent.resolve()
 
+    @property
+    def purelib(self) -> Path:
+        """The selected runtime's package directory, including before installation."""
+        return Path(sysconfig.get_path("purelib", vars={"base": str(self.prefix)}))
+
     @property
     def package_root(self) -> Path:
         """The Core package installed in this runtime's purelib directory."""
-        purelib = Path(sysconfig.get_path("purelib", vars={"base": str(self.prefix)}))
-        root = purelib / "agent_comms"
+        root = self.purelib / "agent_comms"
         initializer = root / "__init__.py"
         if not initializer.is_file():
             raise FileNotFoundError(f"Installed Core package is missing: {initializer}")
diff --git a/src/agent_comms/nominal/field_codec.py b/src/agent_comms/nominal/field_codec.py
index b271abb0b4..211c3802bd 100644
--- a/src/agent_comms/nominal/field_codec.py
+++ b/src/agent_comms/nominal/field_codec.py
@@ -37,6 +37,7 @@
     get_type_hints,
     overload,
 )
+from urllib.parse import quote
 from uuid import uuid4
 from weakref import WeakKeyDictionary
 
@@ -1114,6 +1115,10 @@ def represented_class(cls, annotation: object) -> type[ValueDecoder[object]]:
         )
         return next(iter(cls._registered(annotation)), scalar)
 
+    def value_schema(self) -> dict[str, JsonValue]:
+        """The compiled decoder owns target-dependent schema as well as conversion."""
+        return self.schema()
+
     @classmethod
     def with_representation(
         cls, annotation: object, target: object, args: tuple[object, ...],
@@ -1450,7 +1455,7 @@ def schema_for(cls, annotation: object) -> dict[str, JsonValue]:
             return FieldCodec._family_schema(annotation)
         if isinstance(annotation, type) and is_dataclass(annotation):
             return FieldCodec.record_schema(annotation)
-        return ValueDecoder.represented_class(annotation).schema()
+        return compile_decoder(annotation).value_schema()
 
 
 class AcceptsNull[T](FieldRepresentation[T]):
@@ -3788,8 +3793,20 @@ def _structural_decoder(annotation: object) -> ValueDecoder[object]:
 class _PydanticDecoder(ValueDecoder[object]):
     """An external model owns its validation and JSON representation."""
 
+    origins = (BaseModel,)
     target: type[BaseModel]
 
+    def value_schema(self) -> dict[str, JsonValue]:
+        # A model's root-relative references belong to that model's resource,
+        # even when its schema is nested inside a tool field or union.
+        identity = quote(f"{self.target.__module__}.{self.target.__qualname__}", safe="")
+        return {
+            "$id": f"urn:python-schema:{identity}",
+            **FieldCodec.decode(dict[str, JsonValue], self.target.model_json_schema(
+                by_alias=True, mode="validation",
+            )),
+        }
+
     @boundary_decoder
     def decode(self, data: object) -> BaseModel:
         with MalformedRecord.refusing(f"Invalid {self.target.__name__}"):
@@ -3811,8 +3828,6 @@ class _MemberDecoders(dict[object, ValueDecoder[object]]):
     def __missing__(self, target: object) -> ValueDecoder[object]:
         if isinstance(target, type) and is_dataclass(target):
             decoder: ValueDecoder[object] = _RecordDecoder(target)
-        elif isinstance(target, type) and issubclass(target, BaseModel):
-            decoder = _PydanticDecoder(target)
         else:
             raise UnsupportedFieldType(f"Unsupported JSON type: {target}")
         self[target] = decoder
diff --git a/src/agent_comms_hooks/shell_commands.py b/src/agent_comms_hooks/shell_commands.py
index 1b652dc1d6..30d8798d71 100644
--- a/src/agent_comms_hooks/shell_commands.py
+++ b/src/agent_comms_hooks/shell_commands.py
@@ -6,6 +6,7 @@
 import shlex
 from collections.abc import Iterable, Iterator
 from dataclasses import dataclass, field, replace
+from itertools import islice
 from pathlib import Path
 from typing import ClassVar, Never, Self
 
@@ -13,7 +14,7 @@
 class ShellText(io.StringIO):
     """shlex skips comments with readline; the shell still needs their newline."""
 
-    def readline(self, size: int = -1) -> str:
+    def readline(self, size: int | None = -1) -> str:
         line = super().readline(size)
         if line.endswith('\n'):
             self.seek(self.tell() - 1)
@@ -23,6 +24,23 @@ def document_lines(self) -> Iterator[str]:
         while line := super().readline():
             yield line
 
+    def read_at(self, position: int, size: int) -> str:
+        """Read only the requested raw span, without invoking the lexer or moving it."""
+        cursor = self.tell()
+        try:
+            self.seek(position)
+            return super().read(size)
+        finally:
+            self.seek(cursor)
+
+    def single_line(self) -> bool:
+        position = 0
+        while chunk := self.read_at(position, io.DEFAULT_BUFFER_SIZE):
+            if '\n' in chunk:
+                return False
+            position += len(chunk)
+        return True
+
 
 @dataclass(frozen=True)
 class Redirection:
@@ -53,9 +71,12 @@ def line_value(self, line: str) -> str:
     def check_line(self, line: str) -> None:
         pass
 
+    def body_lines(self, lexer: ShellLexer) -> Iterator[str]:
+        return lexer.document_lines()
+
     def read_body(self, lexer: ShellLexer) -> None:
         lexer.resume_at_boundary()
-        for line in lexer.text.document_lines():
+        for line in self.body_lines(lexer):
             if self.line_value(line) == self.target:
                 return
             self.check_line(line)
@@ -63,6 +84,21 @@ def read_body(self, lexer: ShellLexer) -> None:
 
 
 class ExpandedHereDocument(HereDocument):
+    def body_lines(self, lexer: ShellLexer) -> Iterator[str]:
+        """Join unquoted continuations before both termination and expansion checks."""
+        parts: list[str] = []
+        for line in super().body_lines(lexer):
+            content = line.removesuffix('\n')
+            backslashes = len(content) - len(content.rstrip('\\'))
+            if line.endswith('\n') and backslashes % 2:
+                parts.append(line[:-2])
+                continue
+            parts.append(line)
+            yield ''.join(parts)
+            parts.clear()
+        if parts:
+            yield ''.join(parts)
+
     def check_line(self, line: str) -> None:
         if '$(' in line or '`' in line:
             raise ValueError('Command substitution in an expanded here-document is not admitted')
@@ -79,6 +115,8 @@ class TabStrippedExpandedHereDocument(TabStrippedHereDocument, ExpandedHereDocum
 
 @dataclass(frozen=True)
 class ShellWord:
+    """Logical spelling retains quotes/escapes; start/end retain physical offsets."""
+
     value: str
     start: int
     end: int
@@ -92,7 +130,9 @@ def __init_subclass__(cls, *, spellings: tuple[str, ...] = ()) -> None:
 
     @classmethod
     def decoded(cls, value: str, lexer: ShellLexer, end: int) -> ShellWord:
-        spelling = lexer.command[lexer.word_start:end]
+        source = lexer.read_at(lexer.word_start, end - lexer.word_start)
+        spelling = ''.join(source[position - lexer.word_start]
+                           for position in lexer.word_positions(end))
         kind = cls.members.get(spelling, cls)
         return kind(value, lexer.word_start, end, spelling)
 
@@ -138,7 +178,7 @@ def tabbed_document(self) -> HereDocument:
         return TabStrippedHereDocument(self.value)
 
 
-class ShellLexer(shlex.shlex):
+class ShellLexer(shlex.shlex, ShellText):
     """Retain shlex's own punctuation decision, before POSIX quote removal erases it.
 
     No second quoting scanner: native state transitions distinguish punctuation from
@@ -146,45 +186,94 @@ class ShellLexer(shlex.shlex):
     """
 
     def __init__(self, command: str):
-        self.text = ShellText(command)
-        self.command = command
+        ShellText.__init__(self, command)
         self._state: str | None = None
         self.word_start = 0
         self.word_type: type[ShellWord] = ShellWord
+        self.continuations: list[int] = []
         self.documents: list[HereDocument] = []
-        super().__init__(self.text, posix=True, punctuation_chars=';&|><()\n')
+        self.target = ArgumentTarget()
+        super().__init__(self, posix=True, punctuation_chars=';&|><()\n')
         self.whitespace = ' \t\r'
         self.whitespace_split = True
 
+    def continuation_end(self, position: int) -> int:
+        """Skip active continuations; the caller's native state establishes activity."""
+        while self.read_at(position, 2) == '\\\n':
+            position += 2
+        return position
+
+    def read(self, size: int | None = -1) -> str:
+        # Let native shlex see the joined input before it classifies words or
+        # punctuation. Escaped backslashes and single-quoted text remain literal.
+        if self.state not in ("'", '\\'):
+            start = self.tell()
+            end = self.continuation_end(start)
+            if end != start:
+                self.continuations.extend(range(start, end, 2))
+                self.seek(end)
+        character = super().read(size)
+        # Punctuation lookahead belongs to the next word, whose target is not
+        # known yet. Its first character is checked on native word entry below.
+        if self.state in ('a', '"'):
+            self.check_substitution(character, self.tell())
+        return character
+
+    def check_substitution(self, character: str, following: int) -> None:
+        """Inspect active source characters, before native quote/escape removal."""
+        if character == '`':
+            self.target.require_expansion()
+        elif character == '$' and self.read_at(self.continuation_end(following), 1) == '(':
+            self.target.require_expansion()
+
     @property
     def state(self) -> str | None:
         return self._state
 
     @state.setter
     def state(self, value: str | None) -> None:
+        if self._state == '\\':
+            self.word_type = QuotedWord
+            # shlex retains the escape before $/` inside double quotes; an inner
+            # -c must receive the actual shell text, not that retained backslash.
+            if value == '"' and self.token[-1:] in ('$', '`'):
+                self.token = self.token[:-2] + self.token[-1:]
         if self._state == ' ' and value not in (' ', None):
             self.word_start = self.position - 1
+            self.check_substitution(self.read_at(self.word_start, 1),
+                                    self.word_start + 1)
         if value == 'c':
             self.word_type = Operator
-        elif value in ("'", '"', '\\'):
+        elif value in ("'", '"'):
             self.word_type = QuotedWord
         self._state = value
 
     @property
     def position(self) -> int:
-        return self.text.tell() - len(self._pushback_chars)
+        return self.tell() - len(self._pushback_chars)
+
+    def word_positions(self, end: int) -> Iterator[int]:
+        """Physical positions of the logical spelling native shlex just consumed."""
+        start = self.word_start
+        for continuation in self.continuations:
+            if continuation < start:
+                continue
+            if continuation >= end:
+                break
+            yield from range(start, continuation)
+            start = continuation + 2
+        yield from range(start, end)
 
     def take_adjacent(self, character: str) -> bool:
-        if self.command[self.position:self.position + 1] != character:
-            return False
-        if self._pushback_chars:
-            self._pushback_chars.pop()
-        else:
-            self.text.read(1)
-        return True
+        position = self.position
+        self.resume_at_boundary(position)
+        if self.read(1) == character:
+            return True
+        self.resume_at_boundary(position)
+        return False
 
     def resume_at_boundary(self, position: int | None = None) -> None:
-        self.text.seek(self.position if position is None else position)
+        self.seek(self.position if position is None else position)
         self._pushback_chars.clear()
         self.state = ' '
 
@@ -196,13 +285,15 @@ def read_documents(self) -> None:
         for document in documents:
             document.read_body(self)
 
-    def next_word(self) -> ShellWord | None:
+    def next_word(self, target: ArgumentTarget) -> ShellWord | None:
+        self.target = target
         self.word_type = ShellWord
+        self.continuations.clear()
         value = super().read_token()
         if value is None:
             return None
         end = self.position
-        if self.state == ' ' and self.command[end - 1:end] in self.whitespace:
+        if self.state == ' ' and self.read_at(end - 1, 1) in self.whitespace:
             end -= 1
         return self.word_type.decoded(value, self, end)
 
@@ -216,7 +307,7 @@ def decoded(cls, value: str, lexer: ShellLexer, end: int) -> ShellWord:
                          if value.startswith(key)), '')
         if not spelling:
             raise ValueError(f'Unsupported shell operator {value!r}')
-        end = lexer.word_start + len(spelling)
+        end = next(islice(lexer.word_positions(end), len(spelling) - 1, None)) + 1
         selected = cls.members[spelling](spelling, lexer.word_start, end, spelling)
         # Admit one operator before recursive parsing; native lookahead stays here.
         lexer.resume_at_boundary(end)
@@ -273,10 +364,13 @@ def bind(self, word: ShellWord) -> Redirection:
     def consume_descriptor(self, parser: ShellParser) -> None:
         pass
 
+    def operand(self) -> RedirectionTarget:
+        return RedirectionTarget(self)
+
     def apply(self, parser: ShellParser) -> Iterator[ShellCommand]:
         parser.target.require_word()
         self.consume_descriptor(parser)
-        parser.target = RedirectionTarget(self)
+        parser.target = self.operand()
         return iter(())
 
 
@@ -318,11 +412,14 @@ class DocumentRedirect(NumberedRedirect, spellings=('<<',)):
     def bind(self, word: ShellWord) -> Redirection:
         return word.document()
 
+    def operand(self) -> HereDocumentTarget:
+        return HereDocumentTarget(self)
+
     def apply(self, parser: ShellParser) -> Iterator[ShellCommand]:
         super().apply(parser)
         if parser.lexer.take_adjacent('-'):
-            parser.target = RedirectionTarget(TabbedDocumentRedirect(
-                '<<-', self.start, self.end + 1, '<<-'))
+            parser.target = TabbedDocumentRedirect(
+                '<<-', self.start, parser.lexer.position, '<<-').operand()
         return iter(())
 
 
@@ -332,6 +429,9 @@ def bind(self, word: ShellWord) -> Redirection:
 
 
 class ArgumentTarget:
+    def require_expansion(self) -> None:
+        raise NotStandalone('Command substitution is not admitted by the bounded shell grammar')
+
     def take(self, parser: ShellParser, word: ShellWord) -> None:
         word.argument(parser)
 
@@ -353,6 +453,13 @@ def require_word(self) -> None:
         raise ValueError('Shell redirection requires a word operand')
 
 
+class HereDocumentTarget(RedirectionTarget):
+    """The shell removes delimiter quotes but never expands its word."""
+
+    def require_expansion(self) -> None:
+        pass
+
+
 class ShellParser:
     def __init__(self, lexer: ShellLexer):
         self.lexer = lexer
@@ -393,7 +500,7 @@ def flush(self) -> Iterator[ShellCommand]:
         return iter(completed)
 
     def commands(self) -> Iterator[ShellCommand]:
-        while word := self.lexer.next_word():
+        while word := self.lexer.next_word(self.target):
             yield from word.consume(self)
         self.finish()
         self.lexer.read_documents()
@@ -423,8 +530,13 @@ class NotStandalone(ValueError):
 
 
 class LiteralShellParser(ShellParser):
+    def __init__(self, lexer: ShellLexer):
+        if not lexer.single_line():
+            raise NotStandalone('Recovery requires literal, single-line words')
+        super().__init__(lexer)
+
     def word(self, word: ShellWord) -> None:
-        if set(word.spelling) & {'$', '`', '\n'}:
+        if set(word.spelling) & {'$', '`'}:
             raise NotStandalone('Recovery requires literal, single-line words')
         super().word(word)
 
diff --git a/stack/uv.lock b/stack/uv.lock
index 192af8fec1..0d843a4fe6 100644
--- a/stack/uv.lock
+++ b/stack/uv.lock
@@ -210,7 +210,7 @@ wheels = [
 [[package]]
 name = "batrachian-toad"
 version = "0.6.20"
-source = { git = "https://github.com/OpenHCSDev/toad.git?branch=main#03f32f270ec4ba52dc9ba559b43c4246fd5f945a" }
+source = { git = "https://github.com/OpenHCSDev/toad.git?branch=main#d47f216e51bc5413b55212cf848f12a698012614" }
 dependencies = [
     { name = "agent-client-protocol" },
     { name = "agent-comms" },
@@ -1029,7 +1029,7 @@ wheels = [
 [[package]]
 name = "textual"
 version = "8.2.8"
-source = { git = "https://github.com/OpenHCSDev/textual.git?branch=main#9a0501c98c8494999a7f031508e6e2dc6d1e7e0d" }
+source = { git = "https://github.com/OpenHCSDev/textual.git?branch=main#7c62867cce97d2b0c521c869750413132a315a77" }
 dependencies = [
     { name = "markdown-it-py", extra = ["linkify"] },
     { name = "mdit-py-plugins" },
diff --git a/tests/conftest.py b/tests/conftest.py
index 6181681da7..0276d04edc 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -58,6 +58,7 @@
 from agent_comms.commander.order import Commander
 from agent_comms.repo import Repo
 from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.installer.probes.interpreter import InterpreterBuildProbe, RecordedInterpreterBuild
 from source_programs import SourcePrograms
 from source_revision_fixture import SourceRepository
 from agent_comms.threads import Thread
@@ -445,7 +446,19 @@ def configure(bus: Comms, *check_repositories: Repo) -> Commander:
 
 
 @pytest.fixture
-def installed_runtime(durable_home: Path, tmp_path: Path) -> LaunchRuntime:
+def interpreter_build() -> RecordedInterpreterBuild:
+    """The native implementation tests execute under, also symlinked by fixture runtimes.
+
+    This observes the real current interpreter; it does not claim a process ran
+    inside the synthetic installation's package environment.
+    """
+    return InterpreterBuildProbe().run()
+
+
+@pytest.fixture
+def installed_runtime(
+    durable_home: Path, tmp_path: Path, interpreter_build: RecordedInterpreterBuild,
+) -> LaunchRuntime:
     """A temporary installation record, executable and pointer, never the real home."""
     from agent_comms.installer.probes.packages import CommitRecord, GitRecord, PackageRecord
     from agent_comms.installer.runtime_resolution import RuntimePackage, RuntimeResolution
@@ -462,7 +475,7 @@ def installed_runtime(durable_home: Path, tmp_path: Path) -> LaunchRuntime:
         )
     return record_installed_runtime(tmp_path / 'installed-runtime', RuntimeResolution(
         core_commit=repository.git(repository.checkout, 'rev-parse', 'HEAD').strip(),
-        packages=packages,
+        packages=packages, interpreter=interpreter_build,
     ))
 
 
diff --git a/tests/test_field_codec.py b/tests/test_field_codec.py
index 6487bead89..a56c58c8fb 100644
--- a/tests/test_field_codec.py
+++ b/tests/test_field_codec.py
@@ -6,7 +6,7 @@
 from typing import Annotated, ClassVar, Self
 
 import pytest
-from pydantic import JsonValue
+from pydantic import BaseModel, JsonValue
 from typing_extensions import TypeForm
 
 from agent_comms.nominal.declared_family import DeclaredFamily, UnknownFamilyMember
@@ -104,6 +104,28 @@ class NoSchema:
         FieldCodec.value_schema(NoSchema)
 
 
+def test_external_model_schema_and_conversion_use_the_same_decoder() -> None:
+    class Child(BaseModel):
+        count: int
+
+    class Model(BaseModel):
+        child: Child
+
+    @dataclass
+    class Envelope:
+        value: Model
+
+    value = Envelope(Model(child=Child(count=3)))
+    assert FieldCodec.decode(Envelope, FieldCodec.encode(value)) == value
+    with pytest.raises(MalformedRecord):
+        FieldCodec.decode(Envelope, {"value": {"child": {"count": "wrong"}}})
+    schema = FieldCodec.value_schema(Model)
+    assert schema["$id"]
+    assert {key: item for key, item in schema.items() if key != "$id"} == Model.model_json_schema()
+    assert FieldCodec.record_schema(Envelope)["properties"]["value"] == schema
+    assert compile_decoder(Model).target is Model
+
+
 def test_transient_plan_excludes_fields_before_decoder_or_value_access() -> None:
     class Opaque:
         pass
diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py
index 5404952800..faf053ee93 100644
--- a/tests/test_proof_build.py
+++ b/tests/test_proof_build.py
@@ -14,6 +14,7 @@
     ProofBuildMismatchError,
     RecordedBuildIdentity,
     UninstalledRepositoryError,
+    UnrecordedBuildIdentity,
 )
 from agent_comms.check_root import CheckRoot
 from agent_comms.commander.order import Commander
@@ -22,6 +23,8 @@
 from agent_comms.nominal.field_codec import FieldCodec, GoalId
 from agent_comms.goal_actions import CreateProblemGoalAction
 from agent_comms.goals import CountCheckResult, OrderCheck
+from agent_comms.held_orders import HeldOrders
+from agent_comms.installer.probes.interpreter import RecordedInterpreterBuild
 from agent_comms.installer.probes.packages import ArchiveRecord, CommitRecord, GitRecord, PackageRecord
 from agent_comms.installer.runtime_resolution import CorePackage, RuntimePackage, RuntimeResolution
 from agent_comms.installer.logs import InstallLogs
@@ -69,7 +72,7 @@ class ProofBuildCheckout:
 
 @pytest.fixture
 def proof_build_checkouts(
-    tmp_path: Path, durable_home: Path
+    tmp_path: Path, durable_home: Path, interpreter_build: RecordedInterpreterBuild,
 ) -> tuple[ProofBuildCheckout, ...]:
     """Own each package's distinct commit graph and the installed manifest; durable_home is the
     HOME the installation records under."""
@@ -113,7 +116,9 @@ def proof_build_checkouts(
         version="1", source=ArchiveRecord(url="file:///owned/core-wheel.whl")
     )
     record_installed_runtime(
-        tmp_path / "runtime", RuntimeResolution(core_commit=core.installed, packages=packages))
+        tmp_path / "runtime", RuntimeResolution(
+            core_commit=core.installed, packages=packages, interpreter=interpreter_build,
+        ))
     return tuple(checkouts)
 
 
@@ -164,7 +169,7 @@ def test_merge_in_head_cannot_close_until_the_recorded_build_contains_it(
 
     runtime = LaunchRuntime.installed()
     resolution = RuntimeResolution.at(runtime.prefix)
-    RuntimeResolution(core_commit=graph.newer, packages=resolution.packages).save(runtime.prefix)
+    resolution.model_copy(update={"core_commit": graph.newer}).save(runtime.prefix)
     accepted = run.after_merge(todo, merged)
     assert accepted.state is after_install
     # P553: bf328ceb5 sat in the installed build under "Merged, waiting on install".
@@ -192,9 +197,50 @@ def test_runtime_package_family_owns_installed_commits(proof_build_checkouts):
         )
         assert recorded.repository == package.repository
         assert recorded.commit == graph.installed
+    assert installed.interpreter == RuntimeResolution.at(LaunchRuntime.installed().prefix).interpreter
     assert FieldCodec.decode(RecordedBuildIdentity, FieldCodec.encode(installed)) == installed
 
 
+def test_same_commits_do_not_reuse_a_proof_from_another_interpreter(proof_build_checkouts):
+    original = RecordedBuildIdentity.installed().recorded()
+    runtime = LaunchRuntime.installed()
+    resolution = RuntimeResolution.at(runtime.prefix)
+    interpreter = resolution.interpreter.require_recorded()
+    changed_image = interpreter.artifacts[0].model_copy(update={"sha256": "0" * 64})
+    changed = interpreter.model_copy(update={
+        "artifacts": (changed_image, *interpreter.artifacts[1:]),
+    })
+    resolution.model_copy(update={"interpreter": changed}).save(runtime.prefix)
+    current = RecordedBuildIdentity.installed().recorded()
+    assert original.commits == current.commits
+    assert original != current
+    assert FieldCodec.decode(RecordedBuildIdentity, FieldCodec.encode(current)) == current
+    graph = proof_build_checkouts[0]
+    with pytest.raises(ProofBuildMismatchError, match="differs from installed"):
+        original.installation((), graph.repository.checkout)
+    # Ancestry still answers its own question; it is not an interpreter proof.
+    assert original.contains(graph.installed, graph.repository.checkout)
+    assert current.contains(graph.installed, graph.repository.checkout)
+
+
+def test_missing_interpreter_evidence_is_readable_but_cannot_certify_a_proof(proof_build_checkouts):
+    installed = RecordedBuildIdentity.installed().recorded()
+    unobserved = RecordedBuildIdentity(installed.commits)
+    retained = FieldCodec.decode(RecordedBuildIdentity, FieldCodec.encode(unobserved))
+    assert retained.commits == installed.commits
+    with pytest.raises(ProofBuildMismatchError, match="no interpreter observation"):
+        retained.recorded()
+    with pytest.raises(ProofBuildMismatchError, match="no interpreter observation"):
+        retained.installation((), proof_build_checkouts[0].repository.checkout)
+    runtime = LaunchRuntime.installed()
+    resolution = RuntimeResolution.at(runtime.prefix)
+    resolution.model_copy(update={"interpreter": retained.interpreter}).save(runtime.prefix)
+    with pytest.raises(ProofBuildMismatchError, match="no interpreter observation"):
+        UnrecordedBuildIdentity().recorded()
+    with pytest.raises(ProofBuildMismatchError, match="no interpreter observation"):
+        installed.installation((), proof_build_checkouts[0].repository.checkout)
+
+
 @pytest.mark.parametrize("package", RuntimePackage.members_with(RuntimePackage))
 def test_containment_selects_the_checkouts_installed_repository(package, proof_build_checkouts):
     installed = RecordedBuildIdentity.installed()
@@ -250,10 +296,13 @@ def test_installed_proof_closes_and_older_build_is_refused_on_copied_bus(
     checked, contained, missing = graph.installed, graph.installed, graph.newer
     github_cli.configure(graph.repository.github)
     for number, commit in ((1, checked), (2, checked), (3, contained), (4, missing)):
-        github_cli.put(graph.repository.github, MergedListedPullRequest(
+        row = MergedListedPullRequest(
             number, f'fix/{number}', commit, updated='2026-10-25T12:00:00Z', base=WorkBranches.main,
             reviewable_times=('2026-10-25T12:00:00Z',), merge=PullRequestCommit(commit),
-            merged='2026-10-25T12:00:00Z'))
+            merged='2026-10-25T12:00:00Z')
+        graph.repository.git(checkout, 'push', 'origin',
+                             f'{row.head}:refs/pull/{row.number}/head')
+        github_cli.put(graph.repository.github, row)
     caller = ThreadCaller("lead")
     works = "Open the installed application and confirm the copied goal keeps its source location."
     artifact = tmp_path / "proof.txt"
@@ -277,7 +326,7 @@ def test_installed_proof_closes_and_older_build_is_refused_on_copied_bus(
                 DeliveredCommitNotInstalledError,
             ),
         )
-        for problem, delivery, expected_error in cases:
+        for number, (problem, delivery, expected_error) in enumerate(cases, 1):
             build = delivery.proof_build
             order = PendingProblemOrder(
                 scope=FIX_SCOPE,
@@ -372,7 +421,9 @@ def test_install_due_compares_fetched_main_to_installed_build(
     Commander.on_root("lead", registered, bus.root)
 
     github_cli.configure(graph.repository.github)
-    due = InstallDue.due(bus, ProgressRead())
+    pulls = ProgressRead()
+    InstallDue.prepare(bus, "lead", HeldOrders.read(bus, "lead"), pulls)
+    due = InstallDue.due(bus, pulls)
 
     assert Repo.git(checkout, "rev-parse", "HEAD") == before
     assert Repo.git(checkout, "rev-parse", "origin/main").strip() == (
@@ -391,8 +442,10 @@ def test_install_due_compares_fetched_main_to_installed_build(
     # An install advances the canonical manifest, not a clock or bus schedule.
     runtime = LaunchRuntime.installed().prefix
     resolution = RuntimeResolution.at(runtime)
-    RuntimeResolution(core_commit=graph.newer, packages=resolution.packages).save(runtime)
-    after_install = InstallDue.due(bus, ProgressRead())
+    resolution.model_copy(update={"core_commit": graph.newer}).save(runtime)
+    pulls = ProgressRead()
+    InstallDue.prepare(bus, "lead", HeldOrders.read(bus, "lead"), pulls)
+    after_install = InstallDue.due(bus, pulls)
     assert after_install == ()
     assert Repo.git(checkout, "rev-parse", "HEAD") == before
     configured = bus.registry.snapshot().commander_project.require().repos
@@ -433,7 +486,9 @@ def test_a_one_minute_old_merge_is_due_and_names_its_request_without_launching(
     merged = int(Repo.git(checkout, "log", "-1", "--format=%ct", merge))
     monkeypatch.setattr(time, "time", lambda: merged + 60)
 
-    (due,) = InstallDue.due(bus, ProgressRead())
+    pulls = ProgressRead()
+    InstallDue.prepare(bus, "lead", HeldOrders.read(bus, "lead"), pulls)
+    (due,) = InstallDue.due(bus, pulls)
     assert due.main == merge
     assert "#41 Install what merged" in due.line() and "#42 fix/next" in due.line()
     assert "order install" in due.line().rsplit(" -> ", 1)[1]
diff --git a/tests/test_runtime_resolution.py b/tests/test_runtime_resolution.py
index a47f89d73d..e6a0c7cc60 100644
--- a/tests/test_runtime_resolution.py
+++ b/tests/test_runtime_resolution.py
@@ -1,14 +1,29 @@
+import ctypes
+import hashlib
+import json
 import os
+import sys
 import tomllib
+from dataclasses import dataclass, fields
+from importlib.metadata import PathDistribution
 from pathlib import Path
+from zipfile import ZipFile
 
 import pytest
 
+from agent_comms.build_artifact import BuildArtifact
+from agent_comms.child_process import Platform
 from agent_comms.nominal.field_codec import FieldCodec
 from agent_comms.installer.refusals import UpgradeRefused
 from agent_comms.installer.commands import InstallCommands
-from agent_comms.installer.probes.packages import InstalledPackagesProbe
-from agent_comms.installer.runtime_resolution import ResolvedLock, RuntimeResolution
+from agent_comms.installer.probes.interpreter import InterpreterBuildProbe
+from agent_comms.installer.probes.packages import (
+    ArchivePin, ArchiveRecord, GitPin, InstalledPackagesProbe, PackageRecord,
+)
+from agent_comms.installer.runtime_resolution import (
+    CorePackage, ResolvedLock, RuntimeResolution, ToadPackage,
+)
+from agent_comms.installer.runtime_stack import RuntimeStack
 from agent_comms.launch_runtime import LaunchRuntime
 
 
@@ -22,6 +37,172 @@ def test_package_probe_entry_uses_one_declaration_before_refusing_build_identity
         )
 
 
+@dataclass(frozen=True)
+class InstalledWheel:
+    runtime: LaunchRuntime
+    wheel: Path
+    distribution: PathDistribution
+    source: ArchiveRecord
+
+@pytest.fixture
+def installed_wheel(tmp_path: Path) -> InstalledWheel:
+    prefix = tmp_path / "runtime"
+    purelib = LaunchRuntime.at_prefix(prefix).purelib
+    metadata = "sample-1.0.dist-info"
+    contents = {
+        "sample/__init__.py": b"value = 1\n",
+        f"{metadata}/METADATA": b"Metadata-Version: 2.4\nName: sample\nVersion: 1.0\n",
+        f"{metadata}/WHEEL": b"Wheel-Version: 1.0\nRoot-Is-Purelib: true\nTag: py3-none-any\n",
+        f"{metadata}/RECORD": b"installer replaces this record\n",
+    }
+    wheel = tmp_path / "sample-1.0-py3-none-any.whl"
+    with ZipFile(wheel, "w") as archive:
+        for name, data in contents.items():
+            archive.writestr(name, data)
+            target = purelib / name
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_bytes(data)
+    source = ArchiveRecord(url=wheel.as_uri())
+    (purelib / metadata / "direct_url.json").write_text(json.dumps({
+        "url": source.url, "archive_info": {},
+    }))
+    return InstalledWheel(LaunchRuntime.at_prefix(prefix), wheel,
+                          PathDistribution(purelib / metadata), source)
+
+
+def test_verified_archive_preserves_content_without_inventing_git(installed_wheel: InstalledWheel):
+    original = installed_wheel.distribution.read_text("direct_url.json")
+    verified = installed_wheel.source.verified_wheel(
+        installed_wheel.wheel, installed_wheel.distribution, installed_wheel.runtime
+    )
+    assert verified.archive_info.hashes == {
+        "sha256": hashlib.sha256(installed_wheel.wheel.read_bytes()).hexdigest()
+    }
+    assert verified.wheel_inputs() == (installed_wheel.wheel,)
+    assert installed_wheel.distribution.read_text("direct_url.json") == original
+    with pytest.raises(ValueError, match="Git revision"):
+        verified.revision()
+    record = RuntimeResolution.model_validate({
+        "core_commit": "unqualified-fixture",
+        "packages": {"sample": {"version": "1.0", "source": verified.model_dump()}},
+    })
+    assert RuntimeResolution.model_validate_json(record.model_dump_json()) == record
+
+
+@pytest.mark.parametrize("changed", [False, True])
+def test_donor_wheel_selection_uses_its_recorded_prefix(
+    installed_wheel: InstalledWheel, monkeypatch: pytest.MonkeyPatch, changed: bool
+):
+    """Model the invoked prefix; this does not launch or qualify a donor worker."""
+    original = installed_wheel.distribution.read_text("direct_url.json")
+    verified = installed_wheel.source.verified_wheel(
+        installed_wheel.wheel, installed_wheel.distribution, installed_wheel.runtime
+    )
+    recorded = RuntimeResolution.model_validate({
+        "core_commit": "unqualified-fixture",
+        "packages": {"sample": {"version": "1.0", "source": verified.model_dump()}},
+    })
+    recorded.save(installed_wheel.runtime.prefix)
+    # sys.executable is the foreign launch input; no production method is patched.
+    monkeypatch.setattr(sys, "executable", str(installed_wheel.runtime.interpreter))
+    probe = InstalledPackagesProbe(recorded.core_commit)
+    assert tuple(field.name for field in fields(probe)) == ("core",)
+    assert probe.wheel_inputs() == recorded.wheel_inputs() == (installed_wheel.wheel,)
+    assert original is not None
+    raw = ArchiveRecord.model_validate_json(original)
+    if changed:
+        installed_wheel.distribution.locate_file("sample/__init__.py").write_bytes(b"changed\n")
+        with pytest.raises(ValueError, match="Installed bytes differ"):
+            raw.verified_wheel(probe.wheel_inputs()[0], installed_wheel.distribution,
+                               installed_wheel.runtime)
+    else:
+        assert raw.verified_wheel(probe.wheel_inputs()[0], installed_wheel.distribution,
+                                  installed_wheel.runtime) == verified
+    assert installed_wheel.distribution.read_text("direct_url.json") == original
+
+
+def test_runtime_stack_reobserves_its_resolution_wheel_inputs(installed_wheel: InstalledWheel):
+    with InstallCommands() as commands:
+        stack = RuntimeStack(Path(__file__).resolve().parents[1], commands,
+                             runtime=installed_wheel.runtime)
+        assert stack.wheel_inputs() == ()
+        verified = installed_wheel.source.verified_wheel(
+            installed_wheel.wheel, installed_wheel.distribution, installed_wheel.runtime
+        )
+        recorded = RuntimeResolution(
+            core_commit="unqualified-fixture",
+            packages={"sample": PackageRecord(version="1.0", source=verified)},
+        )
+        recorded.save(installed_wheel.runtime.prefix)
+        assert stack.wheel_inputs() == (installed_wheel.wheel,)
+
+
+def test_changed_installed_member_refuses_archive_proof(installed_wheel: InstalledWheel):
+    installed_wheel.distribution.locate_file("sample/__init__.py").write_bytes(b"value = 2\n")
+    with pytest.raises(ValueError, match="Installed bytes differ"):
+        installed_wheel.source.verified_wheel(
+            installed_wheel.wheel, installed_wheel.distribution, installed_wheel.runtime
+        )
+
+
+def test_root_wheel_domain_does_not_silently_skip_data_members(installed_wheel: InstalledWheel):
+    with ZipFile(installed_wheel.wheel, "a") as archive:
+        archive.writestr("sample-1.0.data/scripts/other", b"unverified\n")
+    with pytest.raises(ValueError, match="root-layout"):
+        installed_wheel.source.verified_wheel(
+            installed_wheel.wheel, installed_wheel.distribution, installed_wheel.runtime
+        )
+
+
+def test_archive_proof_binds_both_origin_and_runtime(installed_wheel: InstalledWheel, tmp_path: Path):
+    wrong_origin = ArchiveRecord(url=(tmp_path / "different.whl").as_uri())
+    with pytest.raises(ValueError, match="origin differs"):
+        wrong_origin.verified_wheel(installed_wheel.wheel, installed_wheel.distribution,
+                                    installed_wheel.runtime)
+    with pytest.raises(ValueError, match="escapes its runtime"):
+        installed_wheel.source.verified_wheel(
+            installed_wheel.wheel, installed_wheel.distribution,
+            LaunchRuntime.at_prefix(tmp_path / "unrelated"),
+        )
+
+
+def test_native_image_artifact_uses_its_open_handle():
+    address = ctypes.cast(ctypes.pythonapi.Py_GetVersion, ctypes.c_void_p)
+    with Platform.current().native_image(address) as stream:
+        observed = BuildArtifact.from_stream(stream)
+        assert observed == BuildArtifact.capture(observed.path)
+        assert not stream.closed
+    assert stream.closed
+
+
+def test_unrecorded_interpreter_is_not_a_runtime_cache_match(tmp_path: Path):
+    recorded = manifest()
+    observed = InterpreterBuildProbe().run()
+    assert not recorded.interpreter.matches(observed)
+    assert observed.matches(observed)
+    changed = observed.model_copy(update={"abi": observed.abi + "-different"})
+    assert not observed.matches(changed)
+    with InstallCommands() as commands:
+        stack = RuntimeStack(Path(__file__).resolve().parents[1], commands,
+                             home=tmp_path, runtime=LaunchRuntime.running())
+        assert not recorded.matches(stack.lock, stack, recorded.core_commit)
+
+
+def test_runtime_identity_refuses_missing_and_changed_interpreter():
+    with pytest.raises(ValueError, match="no recorded interpreter"):
+        manifest().require_runtime(manifest())
+    observed = InterpreterBuildProbe().run()
+    first = manifest().model_copy(update={"interpreter": observed})
+    assert RuntimeResolution.model_validate_json(first.model_dump_json()) == first
+    first.require_runtime(first)
+    changed_image = observed.artifacts[0].model_copy(update={"sha256": "0" * 64})
+    second = first.model_copy(update={"interpreter": observed.model_copy(update={
+        "artifacts": (changed_image, *observed.artifacts[1:]),
+    })})
+    with pytest.raises(UpgradeRefused, match="recorded build inputs"):
+        first.require_runtime(second)
+
+
 TOAD_COMMIT = "a" * 40
 TEXTUAL_COMMIT = "b" * 40
 
@@ -97,6 +278,28 @@ def test_recorded_sources_match_selected_lock_not_wheel_locations():
     assert lock().pins({**selected, "unselected": "5"}, "core") != built.pins()
 
 
+@pytest.mark.parametrize("digest", ["0" * 64, "1" * 64])
+def test_explicit_archive_keeps_its_pin_and_refuses_repository_ancestry(digest: str):
+    built = manifest()
+    archive = ArchiveRecord.model_validate({
+        "url": "file:///owned/inputs/batrachian_toad-2-py3-none-any.whl",
+        "archive_info": {"hashes": {"sha256": digest}},
+    })
+    built = built.model_copy(update={"packages": {
+        **built.packages,
+        ToadPackage.distribution: PackageRecord(version="2", source=archive),
+    }})
+    pins = built.pins()
+    assert pins[CorePackage.distribution] == GitPin("1", "core")
+    assert pins[ToadPackage.distribution] == ArchivePin(
+        "2", archive.url, (("sha256", digest),)
+    )
+    selected = {name: record.version for name, record in built.packages.items()}
+    assert pins != lock().pins(selected, built.core_commit)
+    with pytest.raises(ValueError, match="Git revision"):
+        ToadPackage.commit(built)
+
+
 def test_missing_and_ambiguous_lock_evidence_refuses():
     with pytest.raises(ValueError):
         lock().pins({"missing": "1"}, "core")
diff --git a/tests/test_shell_decoding.py b/tests/test_shell_decoding.py
index 74bc6b442d..83bc04179a 100644
--- a/tests/test_shell_decoding.py
+++ b/tests/test_shell_decoding.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+import io
 import shlex
 import unittest
 
@@ -10,13 +11,39 @@
     shell_storage_effects,
 )
 from agent_comms_hooks.bounded_waits import WaitCommand
-from agent_comms_hooks.shell_commands import ConnectCommand, HereDocument, ShellCommand
+from agent_comms_hooks.shell_commands import ConnectCommand, HereDocument, ShellCommand, ShellText
 
 
 CWD = Path('/shell-decoder-control')
 
 
 class ShellDecoding(unittest.TestCase):
+    def test_raw_source_reads_preserve_the_cursor(self):
+        source = ShellText('before λ\\\nafter')
+        source.seek(3)
+        self.assertEqual(source.read_at(7, 3), 'λ\\\n')
+        self.assertEqual(source.tell(), 3)
+        self.assertEqual(source.read_at(100, 2), '')
+        self.assertEqual(source.tell(), 3)
+        with self.assertRaises(ValueError):
+            source.read_at(-1, 1)
+        self.assertEqual(source.tell(), 3)
+
+    def test_physical_single_line_scan_restores_the_cursor(self):
+        for text, expected in (
+            ('', True),
+            ('literal', True),
+            ('x' * (io.DEFAULT_BUFFER_SIZE * 2 + 1), True),
+            ('x' * (io.DEFAULT_BUFFER_SIZE - 1) + '\n', False),
+            ('x' * io.DEFAULT_BUFFER_SIZE + '\n', False),
+            ('\\\ncontinued', False),
+        ):
+            with self.subTest(length=len(text), expected=expected):
+                source = ShellText(text)
+                source.seek(min(3, len(text)))
+                self.assertEqual(source.single_line(), expected)
+                self.assertEqual(source.tell(), min(3, len(text)))
+
     def test_words_are_never_promoted_to_operators(self):
         cases = (
             ("echo '>' picked", ('echo', '>', 'picked')),
@@ -31,6 +58,11 @@ def test_words_are_never_promoted_to_operators(self):
             (r'echo \{ \}', ('echo', '{', '}')),
             ('echo { }', ('echo', '{', '}')),
             ("'{' literal", ('{', 'literal')),
+            ("'{'\\\n literal", ('{', 'literal')),
+            ('"\\\n{" literal', ('{', 'literal')),
+            ('echo ">\\\n>"', ('echo', '>>')),
+            ("echo '>\\\n>'", ('echo', '>\\\n>')),
+            ('echo \\>\\\n\\>', ('echo', '>>')),
             ('echo file{brace}', ('echo', 'file{brace}')),
         )
         for text, expected in cases:
@@ -39,6 +71,60 @@ def test_words_are_never_promoted_to_operators(self):
                 self.assertEqual(commands[0].words, expected)
                 self.assertEqual(shell_storage_effects(text, CWD), ())
 
+    def test_active_substitutions_refuse_before_effects_at_every_shell_depth(self):
+        cases = (
+            'printf "%s" "$(touch target)"',
+            'printf %s `touch target`',
+            'printf "%s" "`touch target`"',
+            'printf %s before"$(touch target)"after',
+            "printf %s 'literal'$(touch target)",
+            'printf done;$(touch target)',
+            'printf done;`touch target`',
+            'printf %s >"$(touch target)"',
+            'printf %s "${unused:-$(touch target)}"',
+            'printf %s "$\\\n(touch target)"',
+            r'printf %s "\\$(touch target)"',
+        )
+        for script in cases:
+            for text in (script, shlex.join(('bash', '-c', script))):
+                with self.subTest(text=text), self.assertRaises(StorageDenied):
+                    shell_storage_effects(text, CWD)
+        for text in (
+            r'bash -c "printf %s \$(touch target)"',
+            r'bash -c "printf %s \`touch target\`"',
+        ):
+            with self.subTest(text=text), self.assertRaises(StorageDenied):
+                shell_storage_effects(text, CWD)
+
+    def test_literal_substitution_text_retains_native_quote_and_escape_meaning(self):
+        cases = (
+            ("printf %s '$(touch target)'", '$(touch target)'),
+            ("printf %s '`touch target`'", '`touch target`'),
+            (r'printf %s "\$(touch target)"', '$(touch target)'),
+            (r'printf %s "\`touch target\`"', '`touch target`'),
+            (r'printf %s \`literal\`', '`literal`'),
+            ("printf %s '$'\"(touch target)\"", '$(touch target)'),
+            (r'printf %s "one\q"', r'one\q'),
+            ('printf %s fi\\\nle', 'file'),
+            ('printf %s "fi\\\nle"', 'file'),
+            ("printf %s '$\\\n(literal)'", '$\\\n(literal)'),
+        )
+        for text, value in cases:
+            with self.subTest(text=text):
+                command, = ShellCommand.parsed(text)
+                self.assertEqual(command.words, ('printf', '%s', value))
+                self.assertEqual(shell_storage_effects(text, CWD), ())
+
+    def test_document_delimiter_and_opaque_body_do_not_execute_substitutions(self):
+        for text in (
+            "cat <<'EOF'\n$(touch target)\nEOF\nmkdir after",
+            "cat <<'EOF'\n`touch target`\nEOF\nmkdir after",
+            'cat <<"$(tag)"\nopaque\n$(tag)\nmkdir after',
+            'cat <<-"`tag`"\n\topaque\n\t`tag`\nmkdir after',
+        ):
+            with self.subTest(text=text):
+                self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'after'),))
+
     def test_quoted_operator_cannot_hide_a_forbidden_following_command(self):
         for text in (
             "echo '<<' 'EOF'\npython -m venv env",
@@ -59,6 +145,9 @@ def test_real_documents_remain_opaque_and_following_commands_survive(self):
             "cat <<-'A'\n\t'not shell\n\tA\nmkdir target",
             "cat <<A\nordinary text\nA\nmkdir target",
             "cat <<E\"OF\"\n$(uv sync)\nEOF\nmkdir target",
+            "cat <<E\"\"\\\nOF\n$(touch hidden)\nEOF\nmkdir target",
+            "cat <<\\E\\\nOF\n$(touch hidden)\nEOF\nmkdir target",
+            'cat <<"E\\\nOF"\n$(touch hidden)\nEOF\nmkdir target',
             "cat <<'EOF'\n>>> '''not shell\nEOF\nmkdir target",
             "cat <<'EOF'; echo header\n>& 'not shell\nEOF\nmkdir target",
             shlex.join(('bash', '-c', "python - <<'PY'\nprint(\"it's > picked\")\nPY\nmkdir target")),
@@ -67,10 +156,32 @@ def test_real_documents_remain_opaque_and_following_commands_survive(self):
             with self.subTest(text=text):
                 self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'target'),))
 
+    def test_document_logical_termination_preserves_following_commands(self):
+        for text in (
+            'cat <<EOF\nE\\\nOF\nmkdir target',
+            'cat <<-EOF\n\tE\\\nOF\nmkdir target',
+            "cat <<'EOF'\n$\\\n(touch hidden)\nEOF\nmkdir target",
+            "cat <<-'EOF'\n\t$\\\n(touch hidden)\n\tEOF\nmkdir target",
+            "cat <<'EOF'\nE\\\nOF\nEOF\nmkdir target",
+            'cat <<EOF\nliteral' + '\\' * 2 + '\nEOF\nmkdir target',
+            'cat <<EOF\nnot-end\\\nEOF\nstill body\nEOF\nmkdir target',
+        ):
+            with self.subTest(text=text):
+                self.assertEqual(tuple(command.words for command in ShellCommand.parsed(text)),
+                                 (('cat',), ('mkdir', 'target')))
+                self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'target'),))
+
     def test_document_scope_and_unsupported_expansion_are_explicit(self):
         for text in (
             'cat <<EOF\n$(uv sync)\nEOF\n',
             'cat <<EOF\n`uv sync`\nEOF\n',
+            'cat <<EOF\n$\\\n(touch target)\nEOF\n',
+            'cat <<-EOF\n\t$\\\n(touch target)\n\tEOF\n',
+            'cat <<EOF\n$\\\n\\\n(touch target)\nEOF\n',
+            "cat <<EOF\n'$\\\n(touch target)'\nEOF\n",
+            'cat <<EOF\nnot-end\\\nEOF\n$\\\n(touch target)\nEOF\n',
+            'cat <<E\\\nOF\n$(touch target)\nEOF\n',
+            'cat <<-E\\\nOF\n\t`touch target`\n\tEOF\n',
             "cat <<'EOF'\nno terminator",
             "cat <<'EOF'\n\tEOF\n",
             'echo >',
@@ -79,9 +190,48 @@ def test_document_scope_and_unsupported_expansion_are_explicit(self):
                 with self.assertRaises(StorageDenied):
                     shell_storage_effects(text, CWD)
 
+    def test_assignments_use_logical_spelling_without_erasing_real_quoting(self):
+        for prefix in (
+            'FO\\\nO=value',
+            'FOO=va\\\nlue',
+            'FO\\\nO="va\\\nlue"',
+            "FO\\\nO='va\\\nlue'",
+            '\\\nFOO=value',
+            "FOO=va''lue",
+        ):
+            text = prefix + ' mkdir target'
+            with self.subTest(text=text):
+                command, = ShellCommand.parsed(text)
+                self.assertEqual(command.words, ('mkdir', 'target'))
+                self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'target'),))
+        for prefix in (r'F\OO=value', r'FOO\=value', "'FOO'=value", "FOO''=value"):
+            text = prefix + ' mkdir target'
+            with self.subTest(text=text):
+                command, = ShellCommand.parsed(text)
+                self.assertEqual(command.words, ('FOO=value', 'mkdir', 'target'))
+
+    def test_continued_operators_keep_one_physical_source_boundary(self):
+        for text in (
+            'echo value >\\\n> target',
+            'echo value &\\\n> target',
+            'echo value &>\\\n> target',
+            'echo value >\\\n| target',
+        ):
+            with self.subTest(text=text):
+                command, = ShellCommand.parsed(text)
+                self.assertEqual(command.words, ('echo', 'value'))
+                self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'target'),))
+        text = 'cat <\\\n<\\\n-EOF\n\tliteral\n\tEOF\nmkdir target'
+        self.assertEqual(tuple(command.words for command in ShellCommand.parsed(text)),
+                         (('cat',), ('mkdir', 'target')))
+        self.assertEqual(shell_storage_effects(text, CWD), (WriteStorageEffect(CWD / 'target'),))
+
     def test_reserved_brace_groups_cannot_hide_the_first_executable(self):
         for text in (
             '{ mkdir target; }',
+            '\\\n{ mkdir target; \\\n}',
+            '{\\\n mkdir target; }',
+            'A=literal \\\n{ mkdir target; }',
             '{ python -m venv env; }',
             'mkdir before && { (mkdir inner) & printf ready; }',
             'A=literal { mkdir target; }',
@@ -118,12 +268,15 @@ def test_redirection_operands_are_not_arguments(self):
             ('echo &>> target', ('echo',), ('target',)),
         )
         for number in ('2', '700'):
-            retained = (f"'{number}'", f'"{number}"', '\\' + number, number + ' ')
+            continued = number[:1] + '\\\n' + number[1:]
+            retained = (f"'{number}'", f'"{number}"', '\\' + number, number + ' ',
+                        f"'{number}'\\\n", number[:1] + "''\\\n" + number[1:])
             for operator in ('&>', '&>>'):
                 cases += tuple((f'mkdir {prefix}{operator}/owned/log', ('mkdir', number),
-                                ('/owned/log', number)) for prefix in (number, *retained))
+                                ('/owned/log', number)) for prefix in (number, continued, *retained))
             for operator in ('>', '>>', '>|', '<>'):
-                cases += ((f'mkdir {number}{operator}/owned/log', ('mkdir',), ('/owned/log',)),)
+                cases += tuple((f'mkdir {prefix}{operator}/owned/log', ('mkdir',), ('/owned/log',))
+                               for prefix in (number, continued))
                 cases += tuple((f'mkdir {prefix}{operator}/owned/log', ('mkdir', number),
                                 ('/owned/log', number)) for prefix in retained)
         for text, words, paths in cases:
@@ -182,7 +335,8 @@ def test_literal_recovery_uses_the_same_operator_boundary(self):
         command, = ShellCommand.standalone(valid)
         self.assertEqual(command.args[-1], 'quoted;path')
         for text in ('echo ok; echo no', 'echo "$HOME"', 'echo `date`', 'echo ok > file',
-                     'echo ok\necho no', '(echo ok)'):
+                     'echo ok\necho no', '(echo ok)', '\\\n' + valid,
+                     valid.replace('watch', 'wa\\\ntch')):
             with self.subTest(text=text):
                 self.assertEqual(tuple(ShellCommand.standalone(text)), ())
         self.assertTrue(ConnectCommand.permits('agent-comms-claude-bridge connect'))
@@ -208,6 +362,8 @@ def test_waits_follow_source_scope_and_runner_ownership(self):
         script = 'while kill -0 123; do sleep 1; done'
         for text, bounded in (
             (script, False),
+            ('whi\\\nle kill -0 123; do sleep 1; done', False),
+            ('un\\\ntil kill -0 123; do sleep 1; done', False),
             ('(' + script + ')', False),
             (shlex.join(('bash', '-o', 'pipefail', '-c', script)), False),
             (shlex.join(('timeout', '-s', 'INT', '5', 'bash', '-c', script)), True),
@@ -217,7 +373,13 @@ def test_waits_follow_source_scope_and_runner_ownership(self):
                 waits = tuple(WaitCommand(text).waits())
                 self.assertEqual(len(waits), 1)
                 self.assertEqual(waits[0].bounded, bounded)
-        self.assertEqual(tuple(WaitCommand('"while" kill -0 123').waits()), ())
+        for text in (
+            '"while" kill -0 123',
+            'whi""\\\nle kill -0 123',
+            '\\whi\\\nle kill -0 123',
+        ):
+            with self.subTest(text=text):
+                self.assertEqual(tuple(WaitCommand(text).waits()), ())
         polling = shlex.join(('timeout', '5', 'bash', '-c',
                              'while env -u OLD pgrep -f worker; do sleep 1; done'))
         self.assertTrue(tuple(WaitCommand(polling).waits())[0].polls_pgrep_f())