Toolchains

toolchains

Toolchain catalog and host-side provisioning for sandbox profiles.

A sandbox profile's toolchains map names an entry in :data:CATALOG to a version: exact (1.26.5), partial (22, 1.26) where the tool's official index lists releases without an API token, or auto, read from repo files (go.mod, package.json, .nvmrc, .node-version, .python-version, rust-toolchain, rust-toolchain.toml, .tool-versions).

Everything here runs on the host and never executes anything from the repo or from an archive:

1. :func:resolve turns the requested version into an exact release, the archive URL on the tool's fixed official host and the digest published by the tool's official checksum source (a vendored table for protoc versions without one). Repo files are read as data, with a size cap, and a symlink that leads out of the workdir is refused.
2. :func:fetch_archive downloads the archive (size and time limits, HTTPS only, redirects only to the entry's own hosts and, on GitHub, its own release path), verifies it and stores it in the host cache (~/.cache/agentic-ci/toolchains/, keyed by sha256, atomic writes). Nothing unverified is ever kept or installed.
3. :func:provision does both for every toolchain of a profile and hands each verified archive to a backend :class:Installer, which extracts it inside the sandbox; one the installer already has is skipped without a download when its official digest is a sha256. A failure is recorded in the toolchain's :class:ToolchainResult and that toolchain is skipped; the others go on.

:class:ToolchainEnv holds the variables a provisioned toolchain needs (PATH prepends, GOTOOLCHAIN=local, Go and npm caches, the pnpm store, CARGO_HOME), all under /sandbox/.local or /sandbox/.cache so they stay out of the workdir that is downloaded back.

Messages never carry repo-controlled text: they are fixed strings plus validated versions and file names chosen here. Details (exception classes, HTTP hosts) go to the job log.

VERSION_RE = re.compile('[0-9]{1,9}(\\.[0-9]{1,9}){0,2}') module-attribute

A toolchain version (use fullmatch): MAJOR[.MINOR[.PATCH]], at most 9 digits each.

The bound keeps a repo file from putting an arbitrarily long number into URLs, results and messages.

SANDBOX_TOOLCHAIN_ROOT = '/sandbox/.local/toolchains' module-attribute

Where toolchains are extracted inside the sandbox, one <name>-<version> dir each.

CACHE_ENV = 'AGENTIC_CI_TOOLCHAIN_CACHE' module-attribute

Environment variable that moves the host cache (default ~/.cache/agentic-ci/toolchains).

PROTOC_SHA256 = MappingProxyType({'protoc-28.3-linux-x86_64.zip': '0ad949f04a6a174da83cdcbdb36dee0a4925272a5b6d83f79a6bf9852076d53f', 'protoc-28.3-linux-aarch_64.zip': '1de522032a8b194002fe35cab86d747848238b5e4de4f99648372079f5b46f9a', 'protoc-29.5-linux-x86_64.zip': 'a3f094363cd205c6f7af0d1b9305cb4c8517043f265cdb188f098cae93e8b217', 'protoc-29.5-linux-aarch_64.zip': '25eb0848ff13a90a0b2d2a3b4a9d2babc7fbfe158f596c00d8c26a21028dd6f5', 'protoc-30.2-linux-x86_64.zip': '327e9397c6fb3ea2a542513a3221334c6f76f7aa524a7d2561142b67b312a01f', 'protoc-30.2-linux-aarch_64.zip': 'a3173ea338ef91b1605b88c4f8120d6c8ccf36f744d9081991d595d0d4352996', 'protoc-31.1-linux-x86_64.zip': '96553041f1a91ea0efee963cb16f462f5985b4d65365f3907414c360044d8065', 'protoc-31.1-linux-aarch_64.zip': '6c554de11cea04c56ebf8e45b54434019b1cd85223d4bbd25c282425e306ecc2', 'protoc-32.1-linux-x86_64.zip': 'e9c129c176bb7df02546c4cd6185126ca53c89e7d2f09511e209319704b5dd7e', 'protoc-32.1-linux-aarch_64.zip': '4a802ed23d70f7bad7eb19e5a3e724b3aa967250d572cadfd537c1ba939aee6a', 'protoc-33.0-linux-x86_64.zip': 'd99c011b799e9e412064244f0be417e5d76c9b6ace13a2ac735330fa7d57ad8f', 'protoc-33.0-linux-aarch_64.zip': '4b96bc91f8b54d829b8c3ca2207ff1ceb774843321e4fa5a68502faece584272', 'protoc-33.6-linux-x86_64.zip': '9c49962391ff8b5754342509efc3885db91dfe11a358e1e98c9a71518f0ad199', 'protoc-33.6-linux-aarch_64.zip': '4016cd2ad24c6a2405332a2ef7e04b719240e40d811587c6a1c4bbe514cd668c', 'protoc-34.1-linux-x86_64.zip': 'af27ea66cd26938fe48587804ca7d4817457a08350021a1c6e23a27ccc8c6904', 'protoc-34.1-linux-aarch_64.zip': '31c5e9e3c7bf013cf41fb97765ee255c140024a6b175b6cc9b64beddd7c23ba7', 'protoc-35.1-linux-x86_64.zip': '6930ebf62bd4ea607b98fff052596c6ee564b9835b4ce172c75a3f53ae9d91b7', 'protoc-35.1-linux-aarch_64.zip': '01bf9d08808c7f96678b63f4bd8efa559bb4f83d5a7a270d5edaf507f9d5d9cf'}) module-attribute

Vendored, cross-checked sha256 of protoc releases that predate protobuf's checksum manifest.

PYTHON_BUILD_TAGS = MappingProxyType({version: tag for tag, versions in _PYTHON_BUILDS.items() for version in versions.split()}) module-attribute

python-build-standalone release tag by exact CPython version, for pins the newest lacks.

CATALOG = MappingProxyType({e.name: e for e in _ENTRIES}) module-attribute

Every toolchain a sandbox profile may request, by name.

ToolchainError

Bases: Exception

A toolchain could not be resolved, downloaded, verified or installed.

The message is a fixed string plus validated versions and file names, so it can be shown to users and recorded as a result's reason.

CatalogEntry(name, url, hosts, source, bin_dirs, arch, archive='tar.gz', checksum_url='', binary='', node_scripts=dict(), components=(), component_root='', partial=False, auto='', min_parts=3, max_parts=3) dataclass

How to fetch and verify one toolchain.

url is the archive URL template ({version}, {arch} and, for python, {tag}) on a fixed official host; its last path segment, URL-decoded, is the file name the checksum source lists. hosts is every host the entry may contact, redirects included. source names the resolver and checksum source (see :func:resolve). bin_dirs are the directories, relative to the extracted toolchain, prepended to PATH. archive is tar.gz, tar.xz, zip or binary (a single executable stored as <WRAPPER_DIR>/<binary>). components (with {arch}) names the directories under the archive's component_root (with {version} and {arch}) that are merged into the toolchain directory, as the tool's own installer would; the rest of the archive is not installed. node_scripts maps a launcher name to a script in the archive that the launcher runs with node. arch maps the host machine (x86_64, aarch64) to the name the tool uses. partial says whether a partial version resolves. auto names the repo files auto reads (empty: auto is an error). min_parts is the fewest version parts an exact version has, max_parts the most a version may have.

path_prefix property

/<owner>/<repo>/releases/ for an entry on github.com, else "".

Every github.com URL the entry fetches, redirects included, must start with it, so a renamed or transferred repo cannot supply the archive.

Fetcher

Bases: Protocol

Streams an HTTPS resource; the tests replace the real one.

stream(url, *, hosts, limit, timeout, headers=None, path_prefix='')

Yield the body of url in chunks.

Raises :class:ToolchainError when url or a redirect breaks :func:_check_url (with hosts and path_prefix), the status is not 200, the body exceeds limit bytes or the whole transfer takes longer than timeout seconds.

Source code in src/agentic_ci/toolchains.py

def stream(
    self,
    url: str,
    *,
    hosts: frozenset[str],
    limit: int,
    timeout: float,
    headers: Mapping[str, str] | None = None,
    path_prefix: str = "",
) -> Iterator[bytes]:
    """Yield the body of *url* in chunks.

    Raises :class:`ToolchainError` when *url* or a redirect breaks
    :func:`_check_url` (with *hosts* and *path_prefix*), the status is
    not 200, the body exceeds *limit* bytes or the whole transfer takes
    longer than *timeout* seconds.
    """
    ...

HttpFetcher(session=None)

:class:Fetcher over requests, following redirects only within hosts.

Source code in src/agentic_ci/toolchains.py

def __init__(self, session: requests.Session | None = None) -> None:
    if session is None:
        session = requests.Session()
        # No ~/.netrc credentials for the official hosts; proxies and the
        # CA bundle still come from the environment (see _request_settings).
        session.trust_env = False
    self._session = session

close()

Close the HTTP session.

Source code in src/agentic_ci/toolchains.py

def close(self) -> None:
    """Close the HTTP session."""
    self._session.close()

Resolved(name, version, url, filename, algorithm, digest, size=None) dataclass

An exact release and the digest its official checksum source publishes.

digest instance-attribute

Hex digest of the archive in :attr:algorithm (sha256 or sha512).

AutoVersion(version, source, sha512=None) dataclass

A version read from a repo file for auto.

source instance-attribute

The repo file it came from (a fixed name, safe to show).

sha512 = None class-attribute instance-attribute

pnpm's packageManager hash, when it pins one.

InstallPlan(name, version, sha256, archive, target, binary='', node_scripts=dict(), components=(), component_root='') dataclass

What a backend extracts where, for one verified archive.

target instance-attribute

<SANDBOX_TOOLCHAIN_ROOT>/<name>-<version>.

components = () class-attribute instance-attribute

Directories under component_root merged into target; empty: the whole archive.

Installer

Bases: Protocol

Puts verified archives into the sandbox (the OpenShell backend implements it).

is_installed(plan)

Whether plan is already extracted with the same sha256, so it can be skipped.

This records what was installed; it does not verify the installed files.

Source code in src/agentic_ci/toolchains.py

def is_installed(self, plan: InstallPlan) -> bool:
    """Whether *plan* is already extracted with the same sha256, so it can be skipped.

    This records what was installed; it does not verify the installed files.
    """
    ...

install(plan, archive)

Upload archive and extract it to plan.target; raise :class:ToolchainError.

Source code in src/agentic_ci/toolchains.py

def install(self, plan: InstallPlan, archive: Path) -> None:
    """Upload *archive* and extract it to ``plan.target``; raise :class:`ToolchainError`."""
    ...

ToolchainEnv(path=(), variables=dict()) dataclass

Variables for the provisioned toolchains, for the env script and the setup shim.

path is prepended to PATH in order; variables are exported as they are.

script_lines()

export lines for a bash script.

Source code in src/agentic_ci/toolchains.py

def script_lines(self) -> list[str]:
    """``export`` lines for a bash script."""
    lines = []
    if self.path:
        lines.append(f'export PATH={shlex.quote(":".join(self.path))}:"$PATH"')
    for key, value in self.variables.items():
        lines.append(f"export {key}={shlex.quote(value)}")
    return lines

ToolchainResult(name, requested, resolved='', sha256='', status='failed', reason='') dataclass

One toolchain's outcome, as recorded in _run/toolchains.json.

status is installed, present (a reused sandbox already had it) or failed; reason says why it failed.

Provisioned(results, env) dataclass

Result of :func:provision.

host_machine()

The sandbox's machine (the host's, OpenShell runs it natively): x86_64 or aarch64.

Raises :class:ToolchainError on any other machine.

Source code in src/agentic_ci/toolchains.py

def host_machine() -> str:
    """The sandbox's machine (the host's, OpenShell runs it natively): ``x86_64`` or ``aarch64``.

    Raises :class:`ToolchainError` on any other machine.
    """
    machine = _MACHINES.get(platform.machine().lower())
    if machine is None:
        raise ToolchainError("the host machine is not x86_64 or aarch64")
    return machine

auto_version(name, workdir)

Read the version auto means for toolchain name from repo files in workdir.

Source code in src/agentic_ci/toolchains.py

def auto_version(name: str, workdir: Path) -> AutoVersion:
    """Read the version ``auto`` means for toolchain *name* from repo files in *workdir*."""
    entry = CATALOG[name]
    reader = _AUTO.get(entry.auto)
    if reader is None:
        raise ToolchainError(f"{name}: auto is not supported; pin a version")
    return reader(workdir)

resolve(name, requested, *, workdir, fetcher, arch=None)

Resolve requested (a version or auto) for catalog entry name.

Returns the release and, for auto, what was read from the repo. Raises :class:ToolchainError.

Source code in src/agentic_ci/toolchains.py

def resolve(
    name: str, requested: str, *, workdir: Path, fetcher: Fetcher, arch: str | None = None
) -> tuple[Resolved, AutoVersion | None]:
    """Resolve *requested* (a version or ``auto``) for catalog entry *name*.

    Returns the release and, for ``auto``, what was read from the repo.
    Raises :class:`ToolchainError`.
    """
    entry = CATALOG.get(name)
    if entry is None:
        raise ToolchainError("unknown toolchain")
    machine = arch or host_machine()
    auto = None
    version = requested
    if requested == AUTO:
        auto = auto_version(name, workdir)
        version = auto.version
    if not VERSION_RE.fullmatch(version):
        raise ToolchainError(f"{name}: version must be auto or MAJOR[.MINOR[.PATCH]] digits")
    if len(_parts(version)) > entry.max_parts:
        form = ".".join(["MAJOR", "MINOR", "PATCH"][: entry.max_parts])
        raise ToolchainError(f"{name}: {version} has too many parts; use an exact {form} version")
    if len(_parts(version)) < entry.min_parts and not entry.partial:
        _require_exact(entry, version)
    resolved = _RESOLVERS[entry.source](entry, version, entry.arch[machine], fetcher)
    if auto is not None and auto.sha512 is not None and auto.sha512 != resolved.digest:
        raise ToolchainError(
            f"{name} {resolved.version}: the packageManager hash in package.json does not "
            "match the registry"
        )
    return resolved, auto

default_cache_dir()

$AGENTIC_CI_TOOLCHAIN_CACHE, else ~/.cache/agentic-ci/toolchains.

Source code in src/agentic_ci/toolchains.py

def default_cache_dir() -> Path:
    """``$AGENTIC_CI_TOOLCHAIN_CACHE``, else ``~/.cache/agentic-ci/toolchains``."""
    override = os.environ.get(CACHE_ENV)
    if override:
        return Path(override)
    return Path.home() / ".cache" / "agentic-ci" / "toolchains"

write_atomic(path, data)

Write data to path atomically, through a fresh mkstemp name in its directory.

Nothing left at another name in the directory (a symlink, say) is written through; path itself is replaced, never followed.

Source code in src/agentic_ci/toolchains.py

def write_atomic(path: Path, data: bytes) -> None:
    """Write *data* to *path* atomically, through a fresh ``mkstemp`` name in its directory.

    Nothing left at another name in the directory (a symlink, say) is
    written through; *path* itself is replaced, never followed.
    """
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".partial-")
    try:
        with os.fdopen(fd, "wb") as fh:
            fh.write(data)
        os.replace(tmp, path)
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise

fetch_archive(resolved, *, fetcher, cache_dir)

Return the verified archive for resolved from the host cache, downloading it if needed.

The archive is written to a temporary file and renamed to <cache_dir>/<sha256> only once its digest matches the official one; a mismatch deletes it and raises :class:ToolchainError. A digest other than sha256 (pnpm's sha512) gets an alias file <algorithm>/<hex> holding the sha256. Returns the path and the archive's sha256.

Source code in src/agentic_ci/toolchains.py

def fetch_archive(resolved: Resolved, *, fetcher: Fetcher, cache_dir: Path) -> tuple[Path, str]:
    """Return the verified archive for *resolved* from the host cache, downloading it if needed.

    The archive is written to a temporary file and renamed to
    ``<cache_dir>/<sha256>`` only once its digest matches the official one;
    a mismatch deletes it and raises :class:`ToolchainError`. A digest other
    than sha256 (pnpm's sha512) gets an alias file ``<algorithm>/<hex>``
    holding the sha256. Returns the path and the archive's sha256.
    """
    cache_dir.mkdir(parents=True, exist_ok=True)
    hit = _cached(resolved, cache_dir)
    if hit is not None:
        log.info(f"{resolved.name} {resolved.version}: host cache hit")
        return hit
    fd, tmp_name = tempfile.mkstemp(dir=cache_dir, prefix=".partial-")
    tmp = Path(tmp_name)
    try:
        sha256 = hashlib.sha256()
        other = hashlib.new(resolved.algorithm)
        size = 0
        with os.fdopen(fd, "wb") as fh:
            for chunk in fetcher.stream(
                resolved.url,
                hosts=CATALOG[resolved.name].hosts,
                limit=ARCHIVE_MAX_BYTES,
                timeout=ARCHIVE_TIMEOUT_SECONDS,
                path_prefix=CATALOG[resolved.name].path_prefix,
            ):
                fh.write(chunk)
                sha256.update(chunk)
                other.update(chunk)
                size += len(chunk)
        if other.hexdigest() != resolved.digest:
            raise ToolchainError(
                f"{resolved.name} {resolved.version}: checksum mismatch; the download was discarded"
            )
        if resolved.size is not None and size != resolved.size:
            raise ToolchainError(
                f"{resolved.name} {resolved.version}: size differs from the release index; "
                "the download was discarded"
            )
        digest = sha256.hexdigest()
        blob = cache_dir / digest
        os.replace(tmp, blob)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise
    if resolved.algorithm != "sha256":
        (cache_dir / resolved.algorithm).mkdir(exist_ok=True)
        write_atomic(cache_dir / resolved.algorithm / resolved.digest, digest.encode("ascii"))
    log.info(f"{resolved.name} {resolved.version}: downloaded and verified ({size} bytes)")
    return blob, digest

toolchain_env(installed)

The :class:ToolchainEnv for installed toolchains (in the given order).

Source code in src/agentic_ci/toolchains.py

def toolchain_env(installed: Sequence[Resolved]) -> ToolchainEnv:
    """The :class:`ToolchainEnv` for *installed* toolchains (in the given order)."""
    path: list[str] = []
    variables: dict[str, str] = {}
    names = {r.name for r in installed}
    machine = host_machine() if installed else ""
    for resolved in installed:
        entry = CATALOG[resolved.name]
        base = sandbox_dir(resolved.name, resolved.version)
        for template in entry.bin_dirs:
            subdir = template.format(version=resolved.version, arch=entry.arch[machine])
            path.append(posixpath.normpath(posixpath.join(base, subdir)))
    if "go" in names:
        path.append(posixpath.join(SANDBOX_GOPATH, "bin"))
        variables.update(
            {
                "GOTOOLCHAIN": "local",
                "GOPATH": SANDBOX_GOPATH,
                "GOMODCACHE": SANDBOX_GOMODCACHE,
                "GOCACHE": SANDBOX_GOCACHE,
            }
        )
    if names & {"node", "pnpm"}:
        variables["npm_config_cache"] = SANDBOX_NPM_CACHE
    if "pnpm" in names:
        # pnpm 11 reads pnpm_config_*, earlier releases npm_config_*.
        variables["npm_config_store_dir"] = SANDBOX_PNPM_STORE
        variables["pnpm_config_store_dir"] = SANDBOX_PNPM_STORE
    if "rust" in names:
        path.append(posixpath.join(SANDBOX_CARGO_HOME, "bin"))
        # Sparse is the default from cargo 1.70; 1.68 and 1.69 would clone the
        # git index from github.com instead.
        variables.update(
            {"CARGO_HOME": SANDBOX_CARGO_HOME, "CARGO_REGISTRIES_CRATES_IO_PROTOCOL": "sparse"}
        )
    return ToolchainEnv(path=tuple(dict.fromkeys(path)), variables=variables)

provision(toolchains, *, workdir, installer, fetcher=None, cache_dir=None)

Resolve, fetch, verify and install every toolchain in toolchains.

A toolchain that fails is recorded as failed and skipped; the run goes on. Names outside the catalog and versions that are neither auto nor digits are skipped the same way (parse_profile never lets them through).

Source code in src/agentic_ci/toolchains.py

def provision(
    toolchains: Mapping[str, str],
    *,
    workdir: Path,
    installer: Installer,
    fetcher: Fetcher | None = None,
    cache_dir: Path | None = None,
) -> Provisioned:
    """Resolve, fetch, verify and install every toolchain in *toolchains*.

    A toolchain that fails is recorded as ``failed`` and skipped; the run goes
    on. Names outside the catalog and versions that are neither ``auto`` nor
    digits are skipped the same way (``parse_profile`` never lets them
    through).
    """
    own_fetcher = None
    if fetcher is None:
        fetcher = own_fetcher = HttpFetcher()
    try:
        return _provision_all(
            toolchains,
            workdir=workdir,
            installer=installer,
            fetcher=fetcher,
            cache_dir=cache_dir or default_cache_dir(),
        )
    finally:
        if own_fetcher is not None:
            own_fetcher.close()

write_results(path, results)

Write results as a JSON list to path (_run/toolchains.json).

Source code in src/agentic_ci/toolchains.py

def write_results(path: Path, results: Sequence[ToolchainResult]) -> None:
    """Write *results* as a JSON list to *path* (``_run/toolchains.json``)."""
    path.parent.mkdir(parents=True, exist_ok=True)
    data = json.dumps([r.to_dict() for r in results], indent=2, sort_keys=True) + "\n"
    write_atomic(path, data.encode("utf-8"))