artifact.py

Go to the documentation of this file.

"""Content-addressed PineScript strategy artifact builder."""
 
from __future__ import annotations
 
from contextlib import AbstractContextManager
import copy
import ctypes
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import time
from typing import Any, Callable, Literal, Mapping, Sequence
 
from .error import HpoError
 
from .transpile import (
    REQUEST_FEED_INVENTORY_SCHEMA,
    CodegenIdentity,
    TranspileDiagnostic,
    TranspileFailure,
    codegen_identity,
    codegen_supports_request_inventory,
    legacy_input_kind_version,
    transpile_source,
    wrapper_codegen_refusal,
)
from .window_plan import (
    WindowPlanError,
    WindowPlanIdentity,
    _associate_window_plan_path,
    _check_compile_identity,
    _require_engine_layout_identity,
    associate_window_plan,
    derive_window_plan_path,
    require_window_plan_block,
    resolve_window_plan_path,
    verify_window_plan,
)
 
 
 
 
CANONICAL_COMPILE_FLAGS = (
    "-std=c++17",
    "-O2",
    "-ffp-contract=off",
    "-fPIC",
    "-shared",
)
# Engine 1.0 requires Clang builds outside CMake to lift Clang's 256-level bracket
# limit: generated C++ for long Pine chains nests deeper. GCC has no such flag.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
class ArtifactBuildError(HpoError, RuntimeError):
    """A configuration, transpile, compile, inventory, or publication failure."""
 
    def __init__(
        self,
        stage: str,
        message: str,
        *,
        command: Sequence[str] = (),
        stdout: str = "",
        stderr: str = "",
        diagnostics: Sequence[TranspileDiagnostic] = (),
        actual_sha256: str | None = None,
        expected_sha256: str | None = None,
    ) -> None:
        self.stage = stage
        self.command = tuple(command)
        self.stdout = stdout
        self.stderr = stderr
        self.diagnostics = tuple(diagnostics)
        self.actual_sha256 = actual_sha256
        self.expected_sha256 = expected_sha256
        super().__init__(message)
        code, args = {
            "configuration": ("hpo_toolchain_unavailable", {"reason": "configuration"}),
            "transpile": ("hpo_transpile_failed", {}),
            "compile": ("hpo_toolchain_unavailable", {"reason": "compile"}),
            "inventory": (
                "hpo_toolchain_unavailable",
                {"reason": "transpile_internal"},
            ),
            "plugin_validation": ("hpo_plugin_invalid", {"reason": "validation"}),
            "cache": ("hpo_output_io_failed", {}),
            "generated_cpp_hash": ("hpo_generated_cpp_hash_mismatch", {}),
        }.get(stage, ("hpo_unclassified_error", {}))
        self.with_failure(code, args)
        if stage == "transpile":
            if any(item.phase == "CONFIGURATION" for item in self.diagnostics):
                self.with_failure(
                    "hpo_toolchain_unavailable", {"reason": "configuration"}
                )
            elif any(item.phase == "INTERNAL" for item in self.diagnostics):
                self.with_failure(
                    "hpo_toolchain_unavailable", {"reason": "transpile_internal"}
                )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
@dataclass(frozen=True)
class StrategyArtifact:
    """Paths and metadata consumed by the native strategy-plugin loader.
 
    A deferred build can return a pending artifact. Read ``plugin_validation``
    in the recorded manifest and provenance: ``{"status": "deferred"}`` means
    pending until ``ArtifactBuilder.validate_built_plugin`` succeeds. Successful
    validation records ``abi_version`` and ``required_symbols`` instead.
    """
 
    artifact_key: str
    request_key: str
    plugin_path: Path
    generated_cpp_path: Path
    manifest_path: Path
    provenance_path: Path
    cache_hit: bool
    inputs: tuple[dict[str, Any], ...]
    strategy_params: Mapping[str, Any]
 
    @property
    def library_path(self) -> Path:
        """Compatibility alias for callers that call plugins libraries."""
 
        return self.plugin_path
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
class ArtifactBuilder:
    """Transpile and compile a Pine strategy once per compatible request identity.
 
    Compilation has no time limit by default. Set ``compile_timeout_seconds`` to
    a positive number to limit it; ``None`` means no limit. Artifact lock waits
    default to 120 seconds; a lock that never frees is a deadlock to report.
    ``plugin_load="deferred"`` skips loading the compiled plugin during a build;
    call ``validate_built_plugin`` later in the process chosen by the caller.
    ``wrapper_launched=True`` requires stable codegen >= 1.10.0 and refuses
    source-build cache returns that skip transpilation or compilation.
    """
 
    def __init__(
        self,
        *,
        engine_root: str | os.PathLike[str],
        cache_dir: str | os.PathLike[str] | None = None,
        compiler: str | os.PathLike[str] | None = None,
        eigen_include: str | os.PathLike[str] | None = None,
        plugin_validator: Callable[[Path, int], Mapping[str, Any]] | None = None,
        plugin_load: Literal["build", "deferred"] = "build",
        lock_timeout_seconds: float = 120.0,
        compile_timeout_seconds: float | None = None,
        wrapper_launched: bool = False,
    ) -> None:
        self.engine_root = Path(engine_root).expanduser().resolve()
        self.cache_dir = (
            Path(cache_dir).expanduser().resolve()
            if cache_dir
            else _default_cache_dir()
        )
        self.compiler = os.fspath(compiler) if compiler is not None else None
        self.eigen_include = (
            Path(eigen_include).expanduser().resolve()
            if eigen_include is not None
            else None
        )
        self.plugin_validator = plugin_validator or self._validate_plugin
        self.plugin_load = plugin_load
        self.lock_timeout_seconds = lock_timeout_seconds
        self.compile_timeout_seconds = compile_timeout_seconds
        if type(wrapper_launched) is not bool:
            raise ArtifactBuildError("configuration", "wrapper_launched must be a bool")
        self.wrapper_launched = wrapper_launched
 
 
 
 
 
 
 
 
 
 
 
 
    def build(self, pine_source: str, *, filename: str = "<input>") -> StrategyArtifact:
        if not isinstance(pine_source, str):
            raise TypeError("pine_source must be a string")
        self._check_build_configuration()
        checked_codegen = (
            _require_wrapper_codegen_identity() if self.wrapper_launched else None
        )
        engine = self._resolve_engine_layout()
        compiler = self._resolve_compiler_identity()
        eigen = self._resolve_eigen_identity(engine)
        try:
            codegen = checked_codegen or codegen_identity()
        except ModuleNotFoundError as error:
            raise ArtifactBuildError(
                "configuration",
                "pineforge-codegen is required to compile PineScript; install "
                "pineforge-hpo[transpile] or provide a precompiled artifact",
            ) from error
        output_extension, link_mode = self._platform_link_mode()
        request_identity = self._request_identity(
            source_sha256=hashlib.sha256(pine_source.encode("utf-8")).hexdigest(),
            codegen=codegen,
            engine=engine,
            compiler=compiler,
            eigen=eigen,
            output_extension=output_extension,
            link_mode=link_mode,
            inventory_schema=(
                REQUEST_FEED_INVENTORY_SCHEMA
                if codegen_supports_request_inventory()
                else None
            ),
        )
        request_key = _json_sha256(request_identity)
 
        hit = self._lookup_request(
            request_key,
            request_identity,
            output_extension,
            allow_deferred=self.plugin_load == "deferred",
        )
        if hit is not None:
            if self.wrapper_launched:
                raise _wrapper_bypass_error("request_cache")
            return hit
 
        lock_path = self.cache_dir / "locks" / f"{request_key}.lock"
        with _KeyLock(lock_path, self.lock_timeout_seconds):
            hit = self._lookup_request(
                request_key, request_identity, output_extension, allow_deferred=True
            )
            if hit is not None:
                if self.wrapper_launched:
                    raise _wrapper_bypass_error("request_cache_locked")
                if self.plugin_load == "build":
                    return self._validate_built_plugin_locked(hit)
                return hit
 
            transpile_result = transpile_source(pine_source, filename=filename)
            try:
                generated_cpp = transpile_result.require_success()
            except TranspileFailure as error:
                raise ArtifactBuildError(
                    "transpile",
                    str(error),
                    diagnostics=transpile_result.diagnostics,
                ) from error
 
            if transpile_result.input_kind_schema == 1 and legacy_input_kind_version(
                request_identity["codegen"]["version"]
            ):
                raise ArtifactBuildError(
                    "transpile",
                    "input_kind_schema: 1 contradicts recorded codegen version; "
                    "codegen >= 1.1.0 is required",
                )
 
            generated_cpp_sha256 = hashlib.sha256(
                generated_cpp.encode("utf-8")
            ).hexdigest()
            artifact_identity = {
                "request_identity": request_identity,
                "generated_cpp_sha256": generated_cpp_sha256,
            }
            artifact_key = _json_sha256(artifact_identity)
            existing = self._load_artifact(
                artifact_key=artifact_key,
                request_key=request_key,
                request_identity=request_identity,
                output_extension=output_extension,
                cache_hit=True,
                allow_deferred=True,
            )
            if existing is not None:
                if self.wrapper_launched:
                    raise _wrapper_bypass_error("artifact_cache")
                if self.plugin_load == "build":
                    existing = self._validate_built_plugin_locked(existing)
                self._write_request_index(request_key, artifact_key)
                return existing
 
            artifact = self._compile_and_publish(
                artifact_key=artifact_key,
                request_key=request_key,
                request_identity=request_identity,
                output_extension=output_extension,
                generated_cpp=generated_cpp,
                generated_cpp_sha256=generated_cpp_sha256,
                inputs=transpile_result.inputs,
                input_kind_schema=transpile_result.input_kind_schema,
                strategy_params=transpile_result.strategy_params,
                diagnostics=transpile_result.diagnostics,
                request_feed_inventory=transpile_result.request_feed_inventory,
                source_name=filename,
                compiler=compiler,
                engine=engine,
                eigen=eigen,
            )
            self._write_request_index(request_key, artifact_key)
            return artifact
 
    def build_cpp(
        self,
        generated_cpp: bytes,
        *,
        expected_sha256: str,
        inputs: Sequence[Mapping[str, Any]],
        input_kind_schema: int | None,
        strategy_params: Mapping[str, Any],
        request_feed_inventory: Mapping[str, Any] | None,
        diagnostics: Sequence[TranspileDiagnostic] = (),
        filename: str = "<input>",
    ) -> StrategyArtifact:
        """Compile checked UTF-8 C++ bytes with the caller's emission metadata.
 
        Verify the raw bytes before resolving a compiler or publishing anything.
        This entry never transpiles and never returns before a real compile.
        Supply metadata from the checked emission; no inventory is inferred.
        """
 
        if not isinstance(generated_cpp, bytes):
            raise _checked_cpp_error("bytes", "generated_cpp must be raw bytes")
        if (
            not isinstance(expected_sha256, str)
            or re.fullmatch(r"[0-9a-fA-F]{64}", expected_sha256) is None
        ):
            raise _checked_cpp_error(
                "expected_sha256", "expected_sha256 must contain exactly 64 hex digits"
            )
        actual_sha256 = hashlib.sha256(generated_cpp).hexdigest()
        if actual_sha256 != expected_sha256.lower():
            raise ArtifactBuildError(
                "generated_cpp_hash",
                "checked generated C++ SHA-256 mismatch: "
                f"expected {expected_sha256}, actual {actual_sha256}",
                actual_sha256=actual_sha256,
                expected_sha256=expected_sha256,
            )
        try:
            generated_text = generated_cpp.decode("utf-8")
        except UnicodeDecodeError as error:
            raise _checked_cpp_error(
                "utf8", f"checked generated C++ is not strict UTF-8: {error}"
            ) from error
 
        self._check_build_configuration()
        try:
            codegen = (
                _require_wrapper_codegen_identity()
                if self.wrapper_launched
                else codegen_identity()
            )
        except ModuleNotFoundError as error:
            raise ArtifactBuildError(
                "configuration",
                "checked C++ requires its emitting codegen installation",
            ) from error
        if (
            not isinstance(inputs, (list, tuple))
            or any(not isinstance(item, Mapping) for item in inputs)
            or not isinstance(strategy_params, Mapping)
            or not (
                input_kind_schema is None
                or (type(input_kind_schema) is int and input_kind_schema == 1)
            )
            or not isinstance(filename, str)
            or (
                request_feed_inventory is not None
                and (
                    not isinstance(request_feed_inventory, Mapping)
                    or request_feed_inventory.get("schema")
                    != REQUEST_FEED_INVENTORY_SCHEMA
                    or request_feed_inventory.get("source_sha256") != actual_sha256
                )
            )
            or (request_feed_inventory is None and codegen_supports_request_inventory())
        ):
            raise _checked_cpp_error(
                "metadata",
                "checked emission requires its real inputs, input-kind schema, "
                "strategy parameters and source-matching request-feed inventory "
                "(when the loaded producer supports one)",
            )
        if not isinstance(diagnostics, (list, tuple)) or any(
            not isinstance(item, TranspileDiagnostic) for item in diagnostics
        ):
            raise _checked_cpp_error(
                "diagnostics", "checked diagnostics must be TranspileDiagnostic objects"
            )
        if any(item.severity == "error" for item in diagnostics):
            raise ArtifactBuildError(
                "transpile",
                "checked emission has error diagnostics",
                diagnostics=diagnostics,
            )
        if input_kind_schema == 1 and legacy_input_kind_version(codegen.version):
            raise _checked_cpp_error(
                "metadata", "input_kind_schema: 1 requires codegen >= 1.1.0"
            )
        # Freeze caller-owned metadata before deriving its identity or compiling.
        try:
            emission = copy.deepcopy(
                {
                    "inputs": [dict(item) for item in inputs],
                    "input_kind_schema": input_kind_schema,
                    "strategy_params": dict(strategy_params),
                    "request_feed_inventory": (
                        dict(request_feed_inventory)
                        if request_feed_inventory is not None
                        else None
                    ),
                    "diagnostics": [item.to_dict() for item in diagnostics],
                    "source_name": filename,
                }
            )
            emission_sha256 = _json_sha256(emission)
        except (TypeError, ValueError) as error:
            raise _checked_cpp_error(
                "metadata",
                f"checked emission metadata is not JSON-serializable: {error}",
            ) from error
        engine = self._resolve_engine_layout()
        compiler = self._resolve_compiler_identity()
        eigen = self._resolve_eigen_identity(engine)
        output_extension, link_mode = self._platform_link_mode()
        request_identity = self._request_identity(
            source_sha256=actual_sha256,
            codegen=codegen,
            engine=engine,
            compiler=compiler,
            eigen=eigen,
            output_extension=output_extension,
            link_mode=link_mode,
            inventory_schema=(
                REQUEST_FEED_INVENTORY_SCHEMA
                if request_feed_inventory is not None
                else None
            ),
        )
        request_identity["source_kind"] = "checked_cpp"
        request_identity["emission_metadata_sha256"] = emission_sha256
        request_key = _json_sha256(request_identity)
        artifact_key = _json_sha256(
            {
                "request_identity": request_identity,
                "generated_cpp_sha256": actual_sha256,
            }
        )
        lock_path = self.cache_dir / "locks" / f"{request_key}.lock"
        with _KeyLock(lock_path, self.lock_timeout_seconds):
            artifact = self._compile_and_publish(
                artifact_key=artifact_key,
                request_key=request_key,
                request_identity=request_identity,
                output_extension=output_extension,
                generated_cpp=generated_text,
                generated_cpp_sha256=actual_sha256,
                inputs=emission["inputs"],
                input_kind_schema=input_kind_schema,
                strategy_params=emission["strategy_params"],
                diagnostics=diagnostics,
                request_feed_inventory=emission["request_feed_inventory"],
                source_name=filename,
                compiler=compiler,
                engine=engine,
                eigen=eigen,
            )
            self._write_request_index(request_key, artifact_key)
            return artifact
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
    def validate_built_plugin(self, artifact: StrategyArtifact) -> StrategyArtifact:
        """Load and validate a pending source-cache plugin in the calling process.
 
        Call this in a fresh process to separate the first plugin load from
        compilation. This method starts no process and needs neither codegen nor
        a local engine/compiler installation. It checks the recorded engine ABI
        and content hashes under the request lock before publishing success.
        Already validated artifacts are checked without loading or rewriting.
        The caller supplies process isolation; this method does not install a jail.
        """
 
        if self.lock_timeout_seconds <= 0:
            raise ArtifactBuildError("configuration", "build timeouts must be positive")
        keys = (artifact.request_key, artifact.artifact_key)
        if any(
            len(key) != 64
            or any(character not in "0123456789abcdef" for character in key)
            for key in keys
        ):
            raise ArtifactBuildError(
                "cache", "validation requires valid source-cache keys"
            )
        artifact_dir = self.cache_dir / "artifacts" / artifact.artifact_key
        output_extension, _ = self._platform_link_mode()
        expected_paths = (
            artifact_dir / "manifest.json",
            artifact_dir / "provenance.json",
            artifact_dir / "generated.cpp",
            artifact_dir / f"strategy{output_extension}",
        )
        actual_paths = (
            artifact.manifest_path,
            artifact.provenance_path,
            artifact.generated_cpp_path,
            artifact.plugin_path,
        )
        if actual_paths != expected_paths or any(
            path.resolve().parent != artifact_dir for path in actual_paths
        ):
            raise ArtifactBuildError(
                "cache", "only this builder's source-cache artifacts may be validated"
            )
        lock_path = self.cache_dir / "locks" / f"{artifact.request_key}.lock"
        with _KeyLock(lock_path, self.lock_timeout_seconds):
            return self._validate_built_plugin_locked(artifact)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
    def enrich_window_plan(
        self,
        artifact: StrategyArtifact,
        *,
        explicit_path: str | os.PathLike[str] | None = None,
    ) -> dict[str, Any]:
        """Admit a source-cache artifact for selected execution under its request lock.
 
        Ordinary builds do not call this method. An existing block is re-resolved
        and verified without changing its expected hash. An interrupted two-file
        enrichment leaves selected admission refused until enrichment is re-run.
        A retry repairs missing or stale metadata, including a stale request_identity,
        from the validating side under the same lock. A validated old cache with a
        matching local engine may acquire metadata, never new plugin bytes or keys.
        An explicit path also supports the non-derivable third archive layout.
        Arbitrary precompiled artifacts cannot acquire an identity here.
        """
 
        keys = (artifact.request_key, artifact.artifact_key)
        if any(
            len(key) != 64
            or any(character not in "0123456789abcdef" for character in key)
            for key in keys
        ):
            raise WindowPlanError(
                "window_plan_identity",
                "enrichment requires valid source-cache keys",
            )
        artifact_dir = self.cache_dir / "artifacts" / artifact.artifact_key
        output_extension, _ = self._platform_link_mode()
        expected_paths = (
            artifact_dir / "manifest.json",
            artifact_dir / "provenance.json",
            artifact_dir / "generated.cpp",
            artifact_dir / f"strategy{output_extension}",
        )
        actual_paths = (
            artifact.manifest_path,
            artifact.provenance_path,
            artifact.generated_cpp_path,
            artifact.plugin_path,
        )
        if actual_paths != expected_paths or any(
            path.resolve().parent != artifact_dir for path in actual_paths
        ):
            raise WindowPlanError(
                "window_plan_identity",
                "only this builder's validated source-cache artifacts may be enriched",
            )
        lock_path = self.cache_dir / "locks" / f"{artifact.request_key}.lock"
        with _KeyLock(lock_path, self.lock_timeout_seconds):
            try:
                manifest = json.loads(
                    artifact.manifest_path.read_text(encoding="utf-8"),
                )
                provenance = json.loads(
                    artifact.provenance_path.read_text(encoding="utf-8")
                )
            except (OSError, ValueError) as error:
                raise WindowPlanError(
                    "window_plan_manifest_invalid",
                    f"cannot read cached manifest/provenance: {artifact_dir}: {error}",
                ) from error
            if not isinstance(manifest, dict) or not isinstance(provenance, dict):
                raise WindowPlanError(
                    "window_plan_manifest_invalid",
                    "source-cache manifest and provenance must be JSON objects",
                )
            # Shallow copies suffice: repairs replace top-level values and never
            # mutate shared nested objects in place.
            originals = (dict(manifest), dict(provenance))
            block = None
            request_identity = manifest.get("request_identity")
            if "window_plan" in manifest or "window_plan" in provenance:
                candidates = []
                # Repair only from a candidate whose block and file-local
                # request_identity validate.
                for location, document in (
                    ("manifest", manifest),
                    ("provenance", provenance),
                ):
                    if "window_plan" not in document:
                        continue
                    try:
                        identity = WindowPlanIdentity.from_dict(document["window_plan"])
                        _check_compile_identity(
                            identity,
                            document.get("request_identity"),
                            location,
                        )
                    except WindowPlanError:
                        continue
                    candidates.append((identity, document))
                if not candidates:
                    raise WindowPlanError(
                        "window_plan_manifest_invalid",
                        "no cached window_plan block validates against its "
                        "file's request_identity",
                    )
                identity = candidates[0][0]
                if any(candidate != identity for candidate, _ in candidates[1:]):
                    raise WindowPlanError(
                        "window_plan_identity",
                        "manifest.window_plan differs from provenance.window_plan",
                    )
                block = identity.to_dict()
                # A valid identity in a file without a valid block cannot rescue a
                # block stored with a stale identity; ordinary builds treat that
                # stale request identity as a cache miss.
                request_identity = next(
                    (
                        document["request_identity"]
                        for _, document in candidates
                        if _json_sha256(document["request_identity"])
                        == artifact.request_key
                    ),
                    None,
                )
            if (
                not isinstance(request_identity, Mapping)
                or _json_sha256(request_identity) != artifact.request_key
            ):
                raise WindowPlanError(
                    "window_plan_identity",
                    "source-cache request_identity differs from its request_key",
                )
            if block is not None:
                for document in (manifest, provenance):
                    document["window_plan"] = block
                    if document.get("request_identity") != request_identity:
                        document["request_identity"] = request_identity
                        if document is provenance:
                            document["compile"] = request_identity["compile"]
            # Validate the proposed repair before publishing either metadata file.
            cached = self._load_artifact(
                artifact_key=artifact.artifact_key,
                request_key=artifact.request_key,
                request_identity=request_identity,
                output_extension=output_extension,
                cache_hit=True,
                metadata=(manifest, provenance),
                allow_deferred=True,
            )
            if cached is None:
                raise ArtifactBuildError(
                    "cache",
                    f"cannot enrich an invalid source-cache artifact: {artifact_dir}",
                )
            if manifest["plugin_validation"] == {"status": "deferred"}:
                raise ArtifactBuildError(
                    "plugin_validation",
                    "artifact plugin validation is pending; "
                    "call validate_built_plugin first",
                )
            try:
                engine = self._resolve_engine_layout()
            except (ArtifactBuildError, OSError, RuntimeError, ValueError) as error:
                raise WindowPlanError(
                    "window_plan_identity",
                    f"cannot verify local engine identity: {self.engine_root}: {error}",
                ) from error
            _require_engine_layout_identity(engine, request_identity)
            if block is not None:
                path = resolve_window_plan_path(explicit_path, self.engine_root, block)
                verify_window_plan(block, path)
            elif explicit_path is not None:
                block = _associate_window_plan_path(engine, explicit_path)
            else:
                block = associate_window_plan(engine)
                if block is None:
                    if derive_window_plan_path(engine) is None:
                        raise WindowPlanError(
                            "window_plan_path",
                            "the selected archive requires an explicit absolute "
                            "window_plan path for source-cache enrichment",
                        )
                    raise WindowPlanError(
                        "window_plan_missing",
                        "the selected engine archive has no window_plan helper sibling",
                    )
            manifest["window_plan"] = block
            provenance["window_plan"] = block
            require_window_plan_block(manifest, provenance)
            try:
                for destination, payload, original in zip(
                    (artifact.manifest_path, artifact.provenance_path),
                    (manifest, provenance),
                    originals,
                ):
                    # Equal metadata is not rewritten just to canonicalize key
                    # order; readers access fields by key.
                    if payload != original:
                        self._write_window_plan_metadata(destination, payload)
            except OSError as error:
                raise ArtifactBuildError(
                    "cache",
                    f"cannot publish window_plan metadata: {artifact_dir}: {error}",
                ) from error
            return block
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
def build_strategy_artifact(
    pine_source: str,
    *,
    engine_root: str | os.PathLike[str],
    cache_dir: str | os.PathLike[str] | None = None,
    compiler: str | os.PathLike[str] | None = None,
    eigen_include: str | os.PathLike[str] | None = None,
    plugin_load: Literal["build", "deferred"] = "build",
    wrapper_launched: bool = False,
    filename: str = "<input>",
) -> StrategyArtifact:
    """Convenience entry point for a one-off CLI artifact build."""
 
    builder = ArtifactBuilder(
        engine_root=engine_root,
        cache_dir=cache_dir,
        compiler=compiler,
        eigen_include=eigen_include,
        plugin_load=plugin_load,
        wrapper_launched=wrapper_launched,
    )
    return builder.build(pine_source, filename=filename)