lib

Reusable functions for parsing @nix protocol and running nix commands.

ArchChecks dataclass

Checks discovered for a single architecture.

Attributes:

Name	Type	Description
arch	str	Architecture identifier (e.g. "x86_64-linux").
checks	list[CheckResult]	One CheckResult per check in this arch.
error	str | None	Error message if the entire arch eval failed, else None.

Source code in src/junix/lib.py

@dataclass
class ArchChecks:
    """Checks discovered for a single architecture.

    Attributes:
        arch: Architecture identifier (e.g. ``"x86_64-linux"``).
        checks: One CheckResult per check in this arch.
        error: Error message if the entire arch eval failed, else None.
    """

    arch: str
    checks: list[CheckResult] = field(default_factory=list)
    error: str | None = None

CheckResult dataclass

Result of evaluating a single check.

Attributes:

Name	Type	Description
arch	str	Architecture identifier (e.g. "x86_64-linux").
name	str	Check name (e.g. "passing").
drv_path	str | None	Derivation path (or None if eval failed).
error	str | None	Error message (or None if eval succeeded).

Source code in src/junix/lib.py

@dataclass
class CheckResult:
    """Result of evaluating a single check.

    Attributes:
        arch: Architecture identifier (e.g. ``"x86_64-linux"``).
        name: Check name (e.g. ``"passing"``).
        drv_path: Derivation path (or ``None`` if eval failed).
        error: Error message (or ``None`` if eval succeeded).
    """

    arch: str
    name: str
    drv_path: str | None = None
    error: str | None = None

EvalResult dataclass

Result of evaluating flake checks across architectures.

Attributes:

Name	Type	Description
arches	list[ArchChecks]	Per-architecture results, one entry per evaluated arch.
path	str	Normalised flake path used for building URIs.

Source code in src/junix/lib.py

@dataclass
class EvalResult:
    """Result of evaluating flake checks across architectures.

    Attributes:
        arches: Per-architecture results, one entry per evaluated arch.
        path: Normalised flake path used for building URIs.
    """

    arches: list[ArchChecks] = field(default_factory=list)
    path: str = ""

    @property
    def all_attrs(self) -> list[str]:
        """Flatten all successful check attrs across all arches.

        Returns:
            List of flake attribute URIs (e.g. ``[".#checks.x86_64-linux.foo"]``).
        """
        return [
            f"{self.path}#checks.{cr.arch}.{cr.name}"
            for ac in self.arches
            if ac.error is None
            for cr in ac.checks
            if cr.error is None and cr.drv_path is not None
        ]

    @property
    def has_errors(self) -> bool:
        """Whether any arch evaluation failed.

        Returns:
            True if at least one arch has an error.
        """
        return any(ac.error is not None for ac in self.arches)

all_attrs property

Flatten all successful check attrs across all arches.

Returns:

Type	Description
list[str]	List of flake attribute URIs (e.g. [".#checks.x86_64-linux.foo"]).

has_errors property

Whether any arch evaluation failed.

Returns:

Type	Description
bool	True if at least one arch has an error.

_apply_color_mode(mode)

Apply the resolved color mode to plumbum's ANSIStyle.

Parameters:

Name	Type	Description	Default
mode	str	One of "auto", "yes", "no". For "auto", the decision follows sys.stderr.isatty(); for "yes" and "no" the value is forced regardless of TTY.	required

Source code in src/junix/lib.py

def _apply_color_mode(mode: str) -> None:
    """Apply the resolved color mode to plumbum's ``ANSIStyle``.

    Args:
        mode: One of ``"auto"``, ``"yes"``, ``"no"``. For ``"auto"``, the
            decision follows ``sys.stderr.isatty()``; for ``"yes"`` and
            ``"no"`` the value is forced regardless of TTY.
    """
    if mode == "yes":
        ANSIStyle.use_color = 1
    elif mode == "no":
        ANSIStyle.use_color = 0
    else:  # "auto"
        ANSIStyle.use_color = 1 if sys.stderr.isatty() else 0

_attribute_eval_time(arches, eval_elapsed, re_eval_seconds)

Split the eval phase wall-clock across per-arch suite times.

Failing checks triggered a per-check nix eval --show-trace round trip; those round trips are measured and assigned to the arch where the check lives. The remaining bulk-eval time is split evenly across the healthy checks in all arches.

Parameters:

Name	Type	Description	Default
arches	list[ArchChecks]	Per-architecture check lists, post-recovery.	required
eval_elapsed	float	Wall-clock of the bulk nix eval call.	required
re_eval_seconds	dict[tuple[str, str], float]	Map of (arch, name) to wall-clock of each failing check's re-eval round trip.	required

Returns:

Type	Description
dict[str, float]	Per-arch seconds to assign to that arch's eval suite time
dict[str, float]	attribute.

Source code in src/junix/lib.py

def _attribute_eval_time(
    arches: list[ArchChecks],
    eval_elapsed: float,
    re_eval_seconds: dict[tuple[str, str], float],
) -> dict[str, float]:
    """Split the eval phase wall-clock across per-arch suite times.

    Failing checks triggered a per-check ``nix eval --show-trace``
    round trip; those round trips are measured and assigned to the
    arch where the check lives. The remaining bulk-eval time is
    split evenly across the healthy checks in all arches.

    Args:
        arches: Per-architecture check lists, post-recovery.
        eval_elapsed: Wall-clock of the bulk ``nix eval`` call.
        re_eval_seconds: Map of ``(arch, name)`` to wall-clock of each
            failing check's re-eval round trip.

    Returns:
        Per-arch seconds to assign to that arch's eval suite ``time``
        attribute.
    """
    total_healthy = sum(1 for ac in arches for cr in ac.checks if cr.error is None)
    re_eval_total = sum(re_eval_seconds.values())
    remaining = max(0.0, eval_elapsed - re_eval_total)
    per_healthy_seconds = remaining / total_healthy if total_healthy else 0.0
    out: dict[str, float] = {}
    for ac in arches:
        arch_re_eval = sum(
            re_eval_seconds.get((cr.arch, cr.name), 0.0) for cr in ac.checks
        )
        arch_healthy = sum(1 for cr in ac.checks if cr.error is None)
        out[ac.arch] = arch_re_eval + (per_healthy_seconds * arch_healthy)
    return out

_current_nix_system() cached

Return the Nix system identifier for the current host machine.

Delegates to nix eval --impure --json --expr builtins.currentSystem. The result is cached because it never changes during a process lifetime.

Returns:

Type	Description
str	Nix system string (e.g. "x86_64-linux").

Source code in src/junix/lib.py

@cache
def _current_nix_system() -> str:
    """Return the Nix system identifier for the current host machine.

    Delegates to `nix eval --impure --json --expr builtins.currentSystem`.
    The result is cached because it never changes during a process lifetime.

    Returns:
        Nix system string (e.g. `"x86_64-linux"`).
    """
    return cast(
        str,
        json.loads(
            local["nix"](
                "eval", "--impure", "--json", "--expr", "builtins.currentSystem"
            )
        ),
    )

_ensure_builds(builds, process_successful, default_name='nix build', expected_count=0, expected_names=None, expected_pairs=None, unbuildable_drv_paths=None, real_failure_drv_paths=None)

Ensure the build list has at least expected_count entries.

Nix only emits actBuild protocol activities when it actually compiles something. Cached/substituted builds produce no actBuild events, so the build list may be shorter than the number of attributes requested. This helper pads with synthetic skipped builds so the JUnit report reflects every requested attribute.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	Collected build events from the protocol handler.	required
process_successful	bool	Whether the overall Nix process succeeded.	required
default_name	str	Name to use for the synthetic build when no builds exist and no expected_count is given.	'nix build'
expected_count	int	Minimum number of builds to return. When 0 (default), at least one build is guaranteed.	0
expected_names	list[str] | None	Names for synthetic entries. When provided, padded entries use these names instead of default_name #N.	None
expected_pairs	list[tuple[str, str]] | None	(name, drv_path) for each expected target. When provided, synthetic entries use the real drv_path (so JUnit classname is the actual store path, not the human-readable name) and unbuildable_drv_paths can be applied per drv. Mutually exclusive with expected_names; if both are given, expected_pairs wins.	None
unbuildable_drv_paths	set[str] | None	Drv paths that Nix explicitly reported as "Cannot build" (either real failure or dep-failed). Synthetic entries whose drv_path is in this set are marked as failures with the drv path in log_lines, so the JUnit report does not falsely label them as cached/passed. None (default) means the caller has no info about which drv paths failed — every synthetic stays cached.	None
real_failure_drv_paths	list[str] | None	Drv paths of builds that failed for their own reasons (e.g. builder exit code, output rejected). When unbuildable_drv_paths causes a synthetic to be marked as failed, this list is used to attribute the failure to a specific root cause: the first entry is used as BuildEvent.failed_dep_drv_path so the JUnit <failure message="..."> can name the dep that broke. With a single failing dep the message is precise; with multiple, the earliest one in the Nix stderr stream is used as best-effort default. None (default) means the caller has no info about which drv paths failed for their own reasons.	None

Returns:

Type	Description
list[BuildEvent]	Build list padded to at least expected_count entries.

Source code in src/junix/lib.py

def _ensure_builds(  # noqa: C901  (pairs+unbuildable+root-cause logic is intrinsically branchy)
    builds: list[BuildEvent],
    process_successful: bool,
    default_name: str = "nix build",
    expected_count: int = 0,
    expected_names: list[str] | None = None,
    expected_pairs: list[tuple[str, str]] | None = None,
    unbuildable_drv_paths: set[str] | None = None,
    real_failure_drv_paths: list[str] | None = None,
) -> list[BuildEvent]:
    """Ensure the build list has at least ``expected_count`` entries.

    Nix only emits ``actBuild`` protocol activities when it actually
    compiles something.  Cached/substituted builds produce no ``actBuild``
    events, so the build list may be shorter than the number of
    attributes requested.  This helper pads with synthetic skipped
    builds so the JUnit report reflects every requested attribute.

    Args:
        builds: Collected build events from the protocol handler.
        process_successful: Whether the overall Nix process succeeded.
        default_name: Name to use for the synthetic build when no
            builds exist and no expected_count is given.
        expected_count: Minimum number of builds to return.  When 0
            (default), at least one build is guaranteed.
        expected_names: Names for synthetic entries.  When provided,
            padded entries use these names instead of ``default_name #N``.
        expected_pairs: ``(name, drv_path)`` for each expected target.
            When provided, synthetic entries use the real ``drv_path``
            (so JUnit ``classname`` is the actual store path, not the
            human-readable name) and ``unbuildable_drv_paths`` can be
            applied per drv.  Mutually exclusive with ``expected_names``;
            if both are given, ``expected_pairs`` wins.
        unbuildable_drv_paths: Drv paths that Nix explicitly reported
            as "Cannot build" (either real failure or dep-failed).
            Synthetic entries whose ``drv_path`` is in this set are
            marked as failures with the drv path in ``log_lines``, so
            the JUnit report does not falsely label them as
            cached/passed.  ``None`` (default) means the caller has no
            info about which drv paths failed — every synthetic stays
            cached.
        real_failure_drv_paths: Drv paths of builds that failed for
            their own reasons (e.g. builder exit code, output
            rejected).  When ``unbuildable_drv_paths`` causes a
            synthetic to be marked as failed, this list is used to
            attribute the failure to a specific root cause: the
            first entry is used as ``BuildEvent.failed_dep_drv_path``
            so the JUnit ``<failure message="...">`` can name the
            dep that broke.  With a single failing dep the message
            is precise; with multiple, the earliest one in the
            Nix stderr stream is used as best-effort default.
            ``None`` (default) means the caller has no info about
            which drv paths failed for their own reasons.

    Returns:
        Build list padded to at least ``expected_count`` entries.
    """
    if not builds and expected_count == 0:
        builds = [
            BuildEvent(
                name=default_name,
                drv_path=default_name,
                started=True,
                stopped=True,
                success=process_successful,
                cached=True,
            )
        ]
    # Pick the richest expected list: expected_pairs wins when
    # non-empty (i.e. the caller has at least one resolved drv path
    # to use); otherwise fall back to expected_names.  An empty
    # ``expected_pairs`` (resolution failed for every attr) is
    # treated as "no pairs" and we use the names as drv paths
    # instead.
    pairs: list[tuple[str, str]] | None
    if expected_pairs:
        pairs = list(expected_pairs)
    elif expected_names is not None:
        pairs = [(name, name) for name in expected_names]
    else:
        pairs = None

    # Best-effort: when there's exactly one real failure, every
    # dep-failed target in the run is almost certainly caused by it.
    # With multiple real failures, the first one Nix reported is
    # usually the root cause (Nix builds in topological order).
    root_cause_drv: str | None = None
    if real_failure_drv_paths:
        if len(real_failure_drv_paths) == 1:
            root_cause_drv = real_failure_drv_paths[0]
        else:
            root_cause_drv = real_failure_drv_paths[0]

    if pairs:
        # Pad with named entries for missing builds.  When ``pairs`` is
        # provided we always add one synthetic per missing pair — the
        # caller already knows the exact set of expected targets, so
        # ``expected_count`` is informational only and is *not* used as
        # an early-exit cap.  Without this, a small ``expected_count``
        # (e.g. ``len(build_attrs)``) would cut off the loop before all
        # dep-failed targets are reported.
        existing_names = {b.name for b in builds}
        for name, drv_path in pairs:
            if name in existing_names:
                continue
            is_unbuildable = (
                unbuildable_drv_paths is not None and drv_path in unbuildable_drv_paths
            )
            if is_unbuildable:
                # Dep-failed target: not built because its dep failed.
                # We use the drv path in the message so the JUnit
                # <system-out> shows the actual store path; the
                # log_lines (which become <system-out>) intentionally
                # do not duplicate "Cannot build '...'." since that
                # string is already in <system-err>.
                builds.append(
                    BuildEvent(
                        name=name,
                        drv_path=drv_path,
                        started=True,
                        stopped=True,
                        success=False,
                        cached=False,
                        log_lines=[drv_path],
                        failed_dep_drv_path=root_cause_drv,
                    )
                )
            else:
                builds.append(
                    BuildEvent(
                        name=name,
                        drv_path=drv_path,
                        started=True,
                        stopped=True,
                        success=True,
                        cached=True,
                    )
                )
    else:
        while len(builds) < expected_count:
            builds.append(
                BuildEvent(
                    name=f"{default_name} #{len(builds) + 1}",
                    drv_path=default_name,
                    started=True,
                    stopped=True,
                    success=True,
                    cached=True,
                )
            )
    return builds

_eval_check_trace(path, arch, name) async

Re-evaluate a single failing check to capture its Nix error trace.

builtins.tryEval swallows the error message, so the bulk nix eval --json ...#checks call only ever reports a generic "evaluation failed" placeholder for failing checks. This helper runs nix eval --show-trace <path>#checks.<arch>.<name>.drvPath and returns the stderr verbatim so the JUnit report can show the real trace to the user. The round-trip wall-clock is also returned so the CLI can include it in the per-arch suite time — this is the only honest per-check timing available, since @nix carries no timestamps.

Parameters:

Name	Type	Description	Default
path	str	Flake path the user passed (e.g. "." or "./tests/failing-flake").	required
arch	str	Architecture (e.g. "x86_64-linux").	required
name	str	Check name (e.g. "eval-fail").	required

Returns:

Type	Description
str | None	Tuple of (Nix error trace or None, elapsed_seconds). The trace
float	is the stderr stripped, or None if the recovery call
tuple[str | None, float]	produced no useful output. Callers should keep the original
tuple[str | None, float]	bulk error in that case.

Source code in src/junix/lib.py

async def _eval_check_trace(
    path: str, arch: str, name: str
) -> tuple[str | None, float]:
    """Re-evaluate a single failing check to capture its Nix error trace.

    `builtins.tryEval` swallows the error message, so the bulk
    `nix eval --json ...#checks` call only ever reports a generic
    ``"evaluation failed"`` placeholder for failing checks.  This helper
    runs ``nix eval --show-trace <path>#checks.<arch>.<name>.drvPath``
    and returns the stderr verbatim so the JUnit report can show the
    real trace to the user.  The round-trip wall-clock is also returned
    so the CLI can include it in the per-arch suite time — this is
    the only honest per-check timing available, since ``@nix`` carries
    no timestamps.

    Args:
        path: Flake path the user passed (e.g. ``"."`` or
            ``"./tests/failing-flake"``).
        arch: Architecture (e.g. ``"x86_64-linux"``).
        name: Check name (e.g. ``"eval-fail"``).

    Returns:
        Tuple of (Nix error trace or None, elapsed_seconds).  The trace
        is the stderr stripped, or ``None`` if the recovery call
        produced no useful output.  Callers should keep the original
        bulk error in that case.
    """
    started = time.monotonic()
    result = await _nix.run(
        ["eval", "--show-trace", f"{path}#checks.{arch}.{name}.drvPath"],
        retcode=None,
    )
    elapsed = time.monotonic() - started
    if result.returncode == 0:
        return None, elapsed
    return result.stderr.strip() or None, elapsed

_overall_success(builds)

Return True when every build completed successfully.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	List of build events to check.	required

Returns:

Type	Description
bool	True if all builds stopped and succeeded.

Source code in src/junix/lib.py

def _overall_success(builds: list[BuildEvent]) -> bool:
    """Return True when every build completed successfully.

    Args:
        builds: List of build events to check.

    Returns:
        True if all builds stopped and succeeded.
    """
    return all(b.stopped and b.success for b in builds)

_parse_nix_stderr(stderr, handler, log_level=0)

Feed stderr lines to handler, forwarding non-@nix lines to stderr.

Parameters:

Name	Type	Description	Default
stderr	str	Raw stderr text from a nix invocation.	required
handler	NixEventHandler	The event handler to feed @nix lines to.	required
log_level	int	Maximum Nix log level to forward to stderr (default 0 = errors only; higher = more verbose).	0

Source code in src/junix/lib.py

def _parse_nix_stderr(
    stderr: str,
    handler: NixEventHandler,
    log_level: int = 0,
) -> None:
    """Feed stderr lines to handler, forwarding non-`@nix` lines to stderr.

    Args:
        stderr: Raw stderr text from a nix invocation.
        handler: The event handler to feed `@nix` lines to.
        log_level: Maximum Nix log level to forward to stderr
            (default 0 = errors only; higher = more verbose).
    """
    for line in stderr.splitlines(keepends=True):
        line = line.rstrip("\n").rstrip("\r")
        if not handler.handle_line(line):
            print(line, file=sys.stderr)

    for msg in handler.get_messages(threshold=log_level):
        print(msg, file=sys.stderr)

_phase_count(events, overall_success, label)

Build the "N evals" or "F/T evals" fragment for one phase.

Parameters:

Name	Type	Description	Default
events	list[BuildEvent]	Phase's BuildEvents.	required
overall_success	bool	Whether everything passed.	required
label	str	The phase label ("evals" or "builds").	required

Returns:

Type	Description
str	A string like "4 evals" or "1/4 evals".

Source code in src/junix/lib.py

def _phase_count(events: list[BuildEvent], overall_success: bool, label: str) -> str:
    """Build the ``"N evals"`` or ``"F/T evals"`` fragment for one phase.

    Args:
        events: Phase's BuildEvents.
        overall_success: Whether everything passed.
        label: The phase label (``"evals"`` or ``"builds"``).

    Returns:
        A string like ``"4 evals"`` or ``"1/4 evals"``.
    """
    total = len(events)
    failed = sum(1 for b in events if not b.success)
    if overall_success:
        return f"{total} {label}"
    return f"{failed}/{total} {label}"

_print_build_list(builds, downloaded=0, header=None)

Print the downloaded line, a header (if any), and one line per build.

Used by subcommands that have no per-arch sections (i.e. build). For multi-arch flows (check), the per-arch :func:_print_build_results and :func:_print_eval_results already print the listing.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	Build events to list individually.	required
downloaded	int	Number of paths downloaded from binary caches.	0
header	str | None	Optional info line to print before the listing, e.g. "build . ... 2 build(s)". Printed in the same color style as the per-arch headers in _print_eval_results / _print_build_results.	None

Source code in src/junix/lib.py

def _print_build_list(
    builds: list[BuildEvent],
    downloaded: int = 0,
    header: str | None = None,
) -> None:
    """Print the downloaded line, a header (if any), and one line per build.

    Used by subcommands that have no per-arch sections (i.e. ``build``).
    For multi-arch flows (``check``), the per-arch :func:`_print_build_results`
    and :func:`_print_eval_results` already print the listing.

    Args:
        builds: Build events to list individually.
        downloaded: Number of paths downloaded from binary caches.
        header: Optional info line to print before the listing, e.g.
            ``"build . ... 2 build(s)"``.  Printed in the same color
            style as the per-arch headers in ``_print_eval_results`` /
            ``_print_build_results``.
    """
    if header is not None:
        print(str(colors.info | f"  {header}"), file=sys.stderr)

    if downloaded:
        print(
            str(colors.info | f"  \u2b07 {downloaded} path(s) downloaded"),
            file=sys.stderr,
        )

    # Note: we intentionally do NOT print a separate
    # ``~ N build(s) cached`` line here.  The per-build ``✓ … (cached)``
    # markers already convey that information, and the header line
    # already states how many builds there are.  A separate count
    # line on top of that is just visual noise — and a misleading
    # one, since it only fires when there are cached builds and
    # otherwise looks like something is missing.
    for b in builds:
        if b.cached and b.success:
            print(
                str(colors.dim & colors.success | f"  \u2713 {b.name} (cached)"),
                file=sys.stderr,
            )
        elif b.success:
            print(str(colors.success | f"  \u2713 {b.name}"), file=sys.stderr)
        elif b.failed_dep_drv_path:
            # Same dim+red + "(dep failed)" suffix as the per-arch
            # build results.  See _print_build_results for the
            # rationale (visual distinction from a direct build
            # failure: dim instead of bold).
            print(
                str(colors.dim & colors.fatal | f"  \u2717 {b.name} (dep failed)"),
                file=sys.stderr,
            )
        else:
            print(str(colors.fatal | f"  \u2717 {b.name}"), file=sys.stderr)

_print_build_results(arch, build_builds, color_mode='auto')

Print build phase results for one architecture.

Parameters:

Name	Type	Description	Default
arch	str	Architecture identifier.	required
build_builds	list[BuildEvent]	Build events (real + synthetic) for this arch.	required
color_mode	str	One of "auto", "yes", "no" — applied via :func:_apply_color_mode.	'auto'

Source code in src/junix/lib.py

def _print_build_results(
    arch: str,
    build_builds: list[BuildEvent],
    color_mode: str = "auto",
) -> None:
    """Print build phase results for one architecture.

    Args:
        arch: Architecture identifier.
        build_builds: Build events (real + synthetic) for this arch.
        color_mode: One of ``"auto"``, ``"yes"``, ``"no"`` — applied via
            :func:`_apply_color_mode`.
    """
    _apply_color_mode(color_mode)
    print(
        str(colors.info | f"build {arch} ... {len(build_builds)} checks"),
        file=sys.stderr,
    )
    for b in build_builds:
        if not b.success:
            if b.failed_dep_drv_path:
                # Dep-failed target: dim+red (different from the
                # bold-red of a direct build failure) plus a
                # "(dep failed)" suffix so the user can see at a
                # glance that the target itself didn't break — a
                # transitive dep did.
                print(
                    str(colors.dim & colors.fatal | f"  \u2717 {b.name} (dep failed)"),
                    file=sys.stderr,
                )
            else:
                print(str(colors.fatal | f"  \u2717 {b.name}"), file=sys.stderr)
        elif b.cached:
            print(
                str(colors.dim & colors.success | f"  \u2713 {b.name} (cached)"),
                file=sys.stderr,
            )
        else:
            print(str(colors.success | f"  \u2713 {b.name}"), file=sys.stderr)

_print_check_summary(evals=None, builds=None, overall_success=True, color_mode='auto')

Print the final summary line used by every junix subcommand.

The format is consistent across commands: PASSED (...) or FAILED (...) with separate counts for evals and builds. Each phase is only mentioned if it has entries, in the order evals, builds.

The per-phase listing (with ✓/✗/(cached)) is the caller's responsibility — check uses :func:_print_eval_results and :func:_print_build_results per arch, and build uses :func:_print_build_list directly. This keeps the visual layout consistent with how each subcommand presents its per-arch sections.

Parameters:

Name	Type	Description	Default
evals	list[BuildEvent] | None	Synthetic BuildEvents from the eval phase. Pass an empty list (or None) for subcommands that don't eval.	None
builds	list[BuildEvent] | None	BuildEvents from the build phase. Pass an empty list (or None) for subcommands that don't build.	None
overall_success	bool	Whether everything passed. Drives the PASS/FAIL prefix.	True
color_mode	str	One of "auto", "yes", "no" — applied via :func:_apply_color_mode.	'auto'

Source code in src/junix/lib.py

def _print_check_summary(
    evals: list[BuildEvent] | None = None,
    builds: list[BuildEvent] | None = None,
    overall_success: bool = True,
    color_mode: str = "auto",
) -> None:
    """Print the final summary line used by every junix subcommand.

    The format is consistent across commands: ``PASSED (...)`` or
    ``FAILED (...)`` with separate counts for evals and builds.  Each
    phase is only mentioned if it has entries, in the order
    ``evals, builds``.

    The per-phase listing (with ``✓``/``✗``/``(cached)``) is the caller's
    responsibility — ``check`` uses :func:`_print_eval_results` and
    :func:`_print_build_results` per arch, and ``build`` uses
    :func:`_print_build_list` directly.  This keeps the visual layout
    consistent with how each subcommand presents its per-arch sections.

    Args:
        evals: Synthetic BuildEvents from the eval phase.  Pass an empty
            list (or ``None``) for subcommands that don't eval.
        builds: BuildEvents from the build phase.  Pass an empty list
            (or ``None``) for subcommands that don't build.
        overall_success: Whether everything passed.  Drives the PASS/FAIL
            prefix.
        color_mode: One of ``"auto"``, ``"yes"``, ``"no"`` — applied via
            :func:`_apply_color_mode`.
    """
    _apply_color_mode(color_mode)
    evals = evals or []
    builds = builds or []

    parts: list[str] = []
    if evals:
        parts.append(_phase_count(evals, overall_success, "evals"))
    if builds:
        parts.append(_phase_count(builds, overall_success, "builds"))

    body = ", ".join(parts)
    if overall_success:
        print(
            str(colors.bold & colors.success | f"PASSED ({body})"),
            file=sys.stderr,
        )
    else:
        print(
            str(colors.bold & colors.fatal | f"FAILED ({body})"),
            file=sys.stderr,
        )

_print_eval_results(eval_result, color_mode='auto')

Print eval phase results in pytest style.

Each check is printed with the exact nix eval command that would reproduce it, so CI failures are copy-pasteable.

Parameters:

Name	Type	Description	Default
eval_result	EvalResult	The result from discover_checks().	required
color_mode	str	One of "auto", "yes", "no" — applied via :func:_apply_color_mode.	'auto'

Source code in src/junix/lib.py

def _print_eval_results(
    eval_result: EvalResult,
    color_mode: str = "auto",
) -> None:
    """Print eval phase results in pytest style.

    Each check is printed with the exact ``nix eval`` command that would
    reproduce it, so CI failures are copy-pasteable.

    Args:
        eval_result: The result from discover_checks().
        color_mode: One of ``"auto"``, ``"yes"``, ``"no"`` — applied via
            :func:`_apply_color_mode`.
    """
    _apply_color_mode(color_mode)
    path = eval_result.path
    for ac in eval_result.arches:
        if ac.error is not None:
            print(
                str(colors.fatal | f"eval {ac.arch} ... ERROR"),
                file=sys.stderr,
            )
            print(str(colors.fatal | f"  {ac.error}"), file=sys.stderr)
            continue
        print(
            str(colors.info | f"eval {ac.arch} ... {len(ac.checks)} checks"),
            file=sys.stderr,
        )
        for cr in ac.checks:
            name = f"nix eval {path}#checks.{cr.arch}.{cr.name}"
            if cr.error is not None:
                print(str(colors.fatal | f"  \u2717 {name}"), file=sys.stderr)
            else:
                print(str(colors.success | f"  \u2713 {name}"), file=sys.stderr)

_rename_builds(builds_list, attrs, drv_to_name, expected_names, expected_drv_paths)

Rename actBuild events and build the expected (name, drv) pairs.

The renames happen on the BuildEvent objects in place (mutating b.name) so _ensure_builds sees the right names and does not create duplicate synthetic entries.

Parameters:

Name	Type	Description	Default
builds_list	list[BuildEvent]	Collected actBuild events from the handler.	required
attrs	list[str]	Original attrs the user passed to nix build.	required
drv_to_name	dict[str, str] | None	Caller-supplied drv_path → name map (e.g. from discover_checks). Wins over the other pairing modes when non-empty.	required
expected_names	list[str] | None	User-facing testcase names parallel to attrs.	required
expected_drv_paths	list[str] | list[str | None] | None	Real /nix/store/...drv paths parallel to expected_names, as resolved by the caller. May contain None entries when resolution failed.	required

Returns:

Type	Description
list[tuple[str, str]]	expected_pairs for _ensure_builds: parallel to
list[tuple[str, str]]	expected_names/expected_drv_paths with the None
list[tuple[str, str]]	entries stripped out (since those attrs have no drv path to
list[tuple[str, str]]	match against unbuildable_drv_paths). Empty when no
list[tuple[str, str]]	resolved drv paths are available, so the caller falls back
list[tuple[str, str]]	to positional pairing.

Source code in src/junix/lib.py

def _rename_builds(  # noqa: C901  (4 pairing modes; a single if/elif is clearer than dispatching)
    builds_list: list[BuildEvent],
    attrs: list[str],
    drv_to_name: dict[str, str] | None,
    expected_names: list[str] | None,
    expected_drv_paths: list[str] | list[str | None] | None,
) -> list[tuple[str, str]]:
    """Rename actBuild events and build the expected (name, drv) pairs.

    The renames happen on the ``BuildEvent`` objects in place
    (mutating ``b.name``) so ``_ensure_builds`` sees the right names
    and does not create duplicate synthetic entries.

    Args:
        builds_list: Collected ``actBuild`` events from the handler.
        attrs: Original attrs the user passed to ``nix build``.
        drv_to_name: Caller-supplied drv_path → name map (e.g. from
            ``discover_checks``).  Wins over the other pairing modes
            when non-empty.
        expected_names: User-facing testcase names parallel to
            ``attrs``.
        expected_drv_paths: Real ``/nix/store/...drv`` paths parallel
            to ``expected_names``, as resolved by the caller.  May
            contain ``None`` entries when resolution failed.

    Returns:
        ``expected_pairs`` for ``_ensure_builds``: parallel to
        ``expected_names``/``expected_drv_paths`` with the ``None``
        entries stripped out (since those attrs have no drv path to
        match against ``unbuildable_drv_paths``).  Empty when no
        resolved drv paths are available, so the caller falls back
        to positional pairing.
    """
    # If every drv-path resolution failed, there is nothing to
    # match against — fall back to positional pairing (and the
    # dep-failed detection is best-effort: the handler's
    # ``unbuildable_drv_paths`` still applies to the actBuild events
    # we *do* have, but synthetic entries for missing targets stay
    # cached instead of being marked as failures).
    if (
        expected_drv_paths is not None
        and expected_names is not None
        and len(expected_names) == len(expected_drv_paths)
        and not any(drv is not None for drv in expected_drv_paths)
    ):
        expected_drv_paths = None

    if drv_to_name:
        for b in builds_list:
            if b.drv_path in drv_to_name:
                b.name = drv_to_name[b.drv_path]
    elif (
        expected_drv_paths
        and expected_names
        and len(expected_names) == len(expected_drv_paths)
    ):
        # Drv_path-based rename: the caller resolved the drv paths up
        # front (e.g. `check` reads them from `discover_checks`,
        # `build` calls `_resolve_drv_paths`). Skip ``None`` entries
        # (resolution failed for that attr).
        drv_to_name_runtime: dict[str, str] = {
            drv: name
            for drv, name in zip(expected_drv_paths, expected_names, strict=True)
            if drv is not None
        }
        for b in builds_list:
            if b.drv_path in drv_to_name_runtime:
                b.name = drv_to_name_runtime[b.drv_path]
    elif expected_names and len(expected_names) == len(attrs):
        # Positional rename: when the caller doesn't know drv paths
        # (e.g. drv-path resolution failed in `_resolve_drv_paths`),
        # assume Nix emits `actBuild` events in the same order as
        # the attrs were passed and pair them up.  Note: when a dep
        # fails first, the dependent targets' actBuild events never
        # fire, so the positional pairing here misaligns.  The
        # dep-failed detection via ``expected_pairs`` still marks the
        # missing expected names as failures.
        for i, b in enumerate(builds_list):
            if i < len(expected_names):
                b.name = expected_names[i]

    if (
        expected_drv_paths
        and expected_names
        and len(expected_names) == len(expected_drv_paths)
    ):
        return [
            (name, drv)
            for name, drv in zip(expected_names, expected_drv_paths, strict=True)
            if drv is not None
        ]
    return []

_resolve_color_mode(cli_value)

Resolve the effective color mode from the CLI value.

This is a no-op pass-through: the precedence (CLI > JUNIX_COLOR > "auto") is handled by the --color switch's default= in cli.py, which feeds the resolved value here. This function exists only for historical symmetry with the call sites.

Parameters:

Name	Type	Description	Default
cli_value	str | None	The raw CLI value (one of "auto", "yes", "no", or None when the parent has not resolved yet — treated as "auto").	required

Returns:

Type	Description
str	The resolved color mode, always one of "auto", "yes", "no".

Source code in src/junix/lib.py

def _resolve_color_mode(cli_value: str | None) -> str:
    """Resolve the effective color mode from the CLI value.

    This is a no-op pass-through: the precedence (CLI > ``JUNIX_COLOR`` >
    ``"auto"``) is handled by the ``--color`` switch's ``default=`` in
    ``cli.py``, which feeds the resolved value here. This function exists
    only for historical symmetry with the call sites.

    Args:
        cli_value: The raw CLI value (one of ``"auto"``, ``"yes"``, ``"no"``,
            or ``None`` when the parent has not resolved yet — treated as
            ``"auto"``).

    Returns:
        The resolved color mode, always one of ``"auto"``, ``"yes"``, ``"no"``.
    """
    return cli_value or "auto"

_resolve_drv_paths(attrs) async

Resolve the drv path of each requested attr via nix eval.

Used by junix build so run_nix_build can pair actBuild events with the user's attrs by drv path (more reliable than positional pairing) and mark dep-failed targets as failures instead of cached/skipped.

Parameters:

Name	Type	Description	Default
attrs	list[str]	Flake attribute paths the user passed to junix build (e.g. ["nixpkgs#hello", "."]).	required

Returns:

Type	Description
list[str | None] | None	List parallel to attrs: each entry is the drv path string
list[str | None] | None	(resolved successfully) or None (eval failed for that
list[str | None] | None	attr — e.g. not a flake attr, or no drvPath). Returns
list[str | None] | None	None (no list) if all resolutions failed — in that case
list[str | None] | None	run_nix_build falls back to its positional pairing.

Source code in src/junix/lib.py

async def _resolve_drv_paths(attrs: list[str]) -> list[str | None] | None:
    """Resolve the drv path of each requested attr via ``nix eval``.

    Used by ``junix build`` so ``run_nix_build`` can pair
    ``actBuild`` events with the user's attrs by drv path (more
    reliable than positional pairing) and mark dep-failed targets as
    failures instead of cached/skipped.

    Args:
        attrs: Flake attribute paths the user passed to
            ``junix build`` (e.g. ``["nixpkgs#hello", "."]``).

    Returns:
        List parallel to ``attrs``: each entry is the drv path string
        (resolved successfully) or ``None`` (eval failed for that
        attr — e.g. not a flake attr, or no ``drvPath``).  Returns
        ``None`` (no list) if all resolutions failed — in that case
        ``run_nix_build`` falls back to its positional pairing.
    """
    if not attrs:
        return []

    import asyncio

    async def _resolve_one(attr: str) -> str | None:
        # Use `nix eval --raw <attr>.drvPath`.  This works for flake
        # attrs (`.#foo`, `nixpkgs#foo`, etc.).  For non-flake args
        # (raw ``/nix/store/...`` paths) it fails, and we return None
        # for that entry — the caller falls back to the drv path
        # already present in the actBuild event.
        try:
            result = await _nix.run(["eval", "--raw", f"{attr}.drvPath"], retcode=None)
        except Exception:  # noqa: BLE001
            return None
        if result.returncode != 0:
            return None
        out = result.stdout.strip() if result.stdout else ""
        return out or None

    results = await asyncio.gather(*(_resolve_one(a) for a in attrs))
    if all(r is None for r in results):
        return None
    return list(results)

_write_report(xml, output_path)

Write the JUnit XML to output_path or stdout.

Parameters:

Name	Type	Description	Default
xml	str	The JUnit XML string to write.	required
output_path	str | None	Path to write to, or None for stdout.	required

Source code in src/junix/lib.py

def _write_report(xml: str, output_path: str | None) -> None:
    """Write the JUnit XML to output_path or stdout.

    Args:
        xml: The JUnit XML string to write.
        output_path: Path to write to, or None for stdout.
    """
    if output_path:
        with open(output_path, "w") as f:
            f.write(xml)
    else:
        sys.stdout.write(xml)

discover_checks(path, eval_arch=None, on_trace_recovery=None) async

Discover flake check attribute URIs for path.

Uses a single nix eval invocation with --apply to evaluate each check with builtins.tryEval, returning a list of objects with arch, name, drvPath and error fields. When eval_arch is not provided, all architectures are returned.

Per-arch wall-clock attribution (seconds) is returned for the CLI to populate <testsuite time="...">. Each arch's value includes the wall-clock of any per-check nix eval --show-trace round trips fired for failing checks plus a proportional share of the bulk nix eval for its healthy checks.

Parameters:

Name	Type	Description	Default
path	str	Flake path (e.g. "." or "github:owner/repo").	required
eval_arch	list[str] | None	Architectures to evaluate (default: all).	None
on_trace_recovery	Callable[[str], None] | None	Optional callback invoked once per failing check just before the second nix eval --show-trace round trip fires. Receives the failing check's arch/name (e.g. "x86_64-linux/eval-fail"). Used by the CLI to advertise the retry to users running with -v: without it, the user would not know junix spent extra nix eval invocations recovering the underlying trace. None (default) silently skips the notification.	None

Returns:

Type	Description
EvalResult	Tuple of (EvalResult, eval_elapsed_per_arch). The
dict[str, float]	EvalResult carries the per-arch check lists; the second
tuple[EvalResult, dict[str, float]]	tuple element is a {arch: seconds} map.

Source code in src/junix/lib.py

async def discover_checks(
    path: str,
    eval_arch: list[str] | None = None,
    on_trace_recovery: Callable[[str], None] | None = None,
) -> tuple[EvalResult, dict[str, float]]:
    """Discover flake check attribute URIs for path.

    Uses a single ``nix eval`` invocation with ``--apply`` to evaluate
    each check with ``builtins.tryEval``, returning a list of objects
    with ``arch``, ``name``, ``drvPath`` and ``error`` fields.  When
    ``eval_arch`` is not provided, all architectures are returned.

    Per-arch wall-clock attribution (seconds) is returned for the
    CLI to populate ``<testsuite time="...">``. Each arch's value
    includes the wall-clock of any per-check ``nix eval --show-trace``
    round trips fired for failing checks plus a proportional share
    of the bulk ``nix eval`` for its healthy checks.

    Args:
        path: Flake path (e.g. ``"."`` or ``"github:owner/repo"``).
        eval_arch: Architectures to evaluate (default: all).
        on_trace_recovery: Optional callback invoked once per failing
            check just before the second ``nix eval --show-trace``
            round trip fires.  Receives the failing check's
            ``arch/name`` (e.g. ``"x86_64-linux/eval-fail"``).  Used
            by the CLI to advertise the retry to users running with
            ``-v``: without it, the user would not know junix spent
            extra ``nix eval`` invocations recovering the underlying
            trace.  ``None`` (default) silently skips the
            notification.

    Returns:
        Tuple of (EvalResult, eval_elapsed_per_arch). The
        ``EvalResult`` carries the per-arch check lists; the second
        tuple element is a ``{arch: seconds}`` map.
    """
    # The path is passed verbatim to nix — no normalisation, no surprises.
    # Users get nix's own error message for malformed paths, not ours.

    # Nix expression that uses builtins.tryEval per check so one failure
    # does not abort the entire eval.  Returns a list of attrsets:
    # [{ arch = "..."; name = "..."; drvPath = "..."; error = null; }]
    if eval_arch:
        nix_systems = "[ " + " ".join(f'"{a}"' for a in eval_arch) + " ]"
        systems_arg = nix_systems
    else:
        systems_arg = "(builtins.attrNames c)"

    apply_expr = (
        "c: builtins.concatMap "
        "(arch: builtins.concatMap "
        "(name: let v = builtins.tryEval "
        "(builtins.getAttr name (builtins.getAttr arch c)); in "
        "if v.success then "
        "let drv = builtins.tryEval v.value.drvPath; in "
        "if drv.success then "
        "[{ arch = arch; name = name; drvPath = drv.value; error = null; }] "
        "else "
        '[{ arch = arch; name = name; drvPath = null; error = "evaluation failed"; }]'
        " else "
        '[{ arch = arch; name = name; drvPath = null; error = "evaluation failed"; }]'
        ") (builtins.attrNames (builtins.getAttr arch c))) "
        f"{systems_arg}"
    )

    started = time.monotonic()
    result = await _nix.run(
        ["eval", "--json", f"{path}#checks", "--apply", apply_expr],
        retcode=None,
    )
    eval_elapsed = time.monotonic() - started

    if result.returncode != 0:
        return (
            EvalResult(
                arches=[ArchChecks(arch="*", error=result.stderr.strip())],
                path=path,
            ),
            {"*": eval_elapsed},
        )

    try:
        entries: list[dict[str, Any]] = json.loads(result.stdout)
    except (json.JSONDecodeError, TypeError):
        return (
            EvalResult(
                arches=[ArchChecks(arch="*", error="Invalid JSON from nix eval")],
                path=path,
            ),
            {"*": eval_elapsed},
        )

    # Group entries by arch.
    by_arch: dict[str, list[CheckResult]] = {}
    for entry in entries:
        arch = entry.get("arch", "")
        name = entry.get("name", "")
        drv_path = entry.get("drvPath")
        error = entry.get("error")
        cr = CheckResult(arch=arch, name=name, drv_path=drv_path, error=error)
        by_arch.setdefault(arch, []).append(cr)

    # Per-check trace recovery: `builtins.tryEval` only exposes
    # `{success, value}` — it never surfaces the error message.  So when a
    # check fails, the bulk call's "evaluation failed" placeholder is
    # useless: re-evaluate the failing attr directly to capture the real
    # Nix error trace.  Only failing checks pay the round trip. The
    # round-trip wall-clock is also recorded so the CLI can include it
    # in the per-arch suite time.
    re_eval_seconds: dict[tuple[str, str], float] = {}
    for arch_checks in by_arch.values():
        for cr in arch_checks:
            if cr.error is None or cr.drv_path is not None:
                continue
            if on_trace_recovery is not None:
                on_trace_recovery(f"{cr.arch}/{cr.name}")
            trace, re_elapsed = await _eval_check_trace(path, cr.arch, cr.name)
            if trace:
                cr.error = trace
            re_eval_seconds[(cr.arch, cr.name)] = re_elapsed

    arches = [
        ArchChecks(arch=arch, checks=checks) for arch, checks in sorted(by_arch.items())
    ]
    eval_elapsed_per_arch = _attribute_eval_time(arches, eval_elapsed, re_eval_seconds)
    return (
        EvalResult(arches=arches, path=path),
        eval_elapsed_per_arch,
    )

discover_checks_individually(path, eval_arch=None, on_progress=None) async

Discover flake checks by running one nix eval per check.

The bulk :func:discover_checks evaluates every check × arch of the flake in a single Nix process, which is memory-efficient for small flakes but blows the RAM budget of CI runners on large ones (e.g. flakes that pull in Odoo's full closure across four architectures).

This helper trades that for per-check isolation: each check runs in its own nix eval <path>#checks.<arch>.<name>.drvPath process, so the RAM peak per process is bounded by the closure of that single check — not the sum of all checks. Cost: one Nix process per check (~hundreds of milliseconds each), so a flake with hundreds of checks pays noticeable wall-clock overhead. Use this only when the bulk path OOMs.

Algorithm (cheap listing, then per-check eval):

1. List arches. One cheap nix eval --json <path>#checks --apply builtins.attrNames to enumerate the flake's check architectures. Skipped if eval_arch is provided by the caller.
2. List checks per arch. One nix eval --json <path>#checks.<arch> --apply builtins.attrNames per arch. This is the second-cheapest Nix call (just attribute names, no closure evaluation).
3. Per-check drvPath. One nix eval --raw <path>#checks.<arch>.<name>.drvPath per check. This is the step that actually loads the check's closure — but each call only sees its own check.

Each per-check call surfaces the real Nix error trace in stderr on failure, so unlike the bulk path no second nix eval --show-trace round trip is needed to recover the trace.

Total nix eval invocations: 1 + A + A·C (no -e) or A + A·C (-e given), where A = arches, C = checks per arch. Bulk path: 1 + F (bulk + per-failing-check trace). Per-check path is slower in wall-clock but bounded in memory by the largest single check.

Parameters:

Name	Type	Description	Default
path	str	Flake path (e.g. "." or "github:owner/repo").	required
eval_arch	list[str] | None	Architectures to evaluate (default: all). When provided, the arch-listing step is skipped.	None
on_progress	Callable[[str], None] | None	Optional callback invoked once per check just before its per-check eval fires, receiving "arch/name". Used by the CLI to advertise progress under -v.	None

Returns:

Name	Type	Description
	EvalResult	Tuple of (EvalResult, eval_elapsed_per_arch) — same shape
as	dict[str, float]	func:discover_checks so the rest of the pipeline
	tuple[EvalResult, dict[str, float]]	(suite structure, JUnit, summary) is unchanged.

Source code in src/junix/lib.py

async def discover_checks_individually(  # noqa: C901  (per-arch × per-check loops + 3-step branching are intrinsic)
    path: str,
    eval_arch: list[str] | None = None,
    on_progress: Callable[[str], None] | None = None,
) -> tuple[EvalResult, dict[str, float]]:
    """Discover flake checks by running one ``nix eval`` per check.

    The bulk :func:`discover_checks` evaluates every check × arch of
    the flake in a single Nix process, which is memory-efficient for
    small flakes but blows the RAM budget of CI runners on large ones
    (e.g. flakes that pull in Odoo's full closure across four
    architectures).

    This helper trades that for **per-check isolation**: each check
    runs in its own ``nix eval <path>#checks.<arch>.<name>.drvPath``
    process, so the RAM peak per process is bounded by the closure
    of that single check — not the sum of all checks.  Cost: one Nix
    process per check (~hundreds of milliseconds each), so a flake
    with hundreds of checks pays noticeable wall-clock overhead.  Use
    this only when the bulk path OOMs.

    Algorithm (cheap listing, then per-check eval):

    1.  **List arches.** One cheap ``nix eval --json
        <path>#checks --apply builtins.attrNames`` to enumerate the
        flake's check architectures.  Skipped if ``eval_arch`` is
        provided by the caller.
    2.  **List checks per arch.** One ``nix eval --json
        <path>#checks.<arch> --apply builtins.attrNames`` per arch.
        This is the second-cheapest Nix call (just attribute names,
        no closure evaluation).
    3.  **Per-check drvPath.** One ``nix eval --raw
        <path>#checks.<arch>.<name>.drvPath`` per check.  This is the
        step that actually loads the check's closure — but each call
        only sees its own check.

    Each per-check call surfaces the real Nix error trace in stderr on
    failure, so unlike the bulk path no second ``nix eval --show-trace``
    round trip is needed to recover the trace.

    Total nix eval invocations: ``1 + A + A·C`` (no ``-e``) or
    ``A + A·C`` (``-e`` given), where ``A`` = arches, ``C`` = checks
    per arch.  Bulk path: ``1 + F`` (bulk + per-failing-check
    trace).  Per-check path is slower in wall-clock but bounded in
    memory by the largest single check.

    Args:
        path: Flake path (e.g. ``"."`` or ``"github:owner/repo"``).
        eval_arch: Architectures to evaluate (default: all). When
            provided, the arch-listing step is skipped.
        on_progress: Optional callback invoked once per check just
            before its per-check eval fires, receiving
            ``"arch/name"``. Used by the CLI to advertise progress
            under ``-v``.

    Returns:
        Tuple of (EvalResult, eval_elapsed_per_arch) — same shape
        as :func:`discover_checks` so the rest of the pipeline
        (suite structure, JUnit, summary) is unchanged.
    """
    arches: list[str]
    arch_listing_total_seconds: float = 0.0
    if eval_arch:
        arches = list(eval_arch)
    else:
        # 1. List arches via `builtins.attrNames`. Cheapest possible
        # Nix call: only the keys of the `checks` attrset, no closure.
        list_arches_expr = "c: builtins.attrNames c"
        listing_started = time.monotonic()
        result = await _nix.run(
            [
                "eval",
                "--json",
                f"{path}#checks",
                "--apply",
                list_arches_expr,
            ],
            retcode=None,
        )
        arch_listing_total_seconds = time.monotonic() - listing_started
        if result.returncode != 0:
            return (
                EvalResult(
                    arches=[ArchChecks(arch="*", error=result.stderr.strip())],
                    path=path,
                ),
                {"*": arch_listing_total_seconds},
            )
        try:
            parsed: list[str] = json.loads(result.stdout)
        except (json.JSONDecodeError, TypeError):
            return (
                EvalResult(
                    arches=[ArchChecks(arch="*", error="Invalid JSON from nix eval")],
                    path=path,
                ),
                {"*": arch_listing_total_seconds},
            )
        arches = [a for a in parsed if isinstance(a, str)]

    # 2. List checks per arch (via `builtins.attrNames`).
    list_checks_expr = "c: builtins.attrNames c"
    # Two parallel maps: arches whose listing succeeded (and the
    # check names that come with it), and arches whose listing
    # failed (and the Nix error from the failing call). A failure
    # here is an honest per-arch eval failure — it surfaces in the
    # JUnit as an `ArchChecks(arch=arch, error=...)`, not as an
    # empty-but-passing suite. The rest of the flake still gets
    # evaluated.
    arch_to_check_names: dict[str, list[str]] = {}
    arch_listing_errors: dict[str, str] = {}
    # Per-arch wall-clock of the listing call itself, attributed
    # alongside the per-check times so the JUnit <testsuite time>
    # accounts for the full eval wall-clock (not just the per-check
    # round trips). Without this, a 1-check arch would report 0.0s
    # while the listing call was clearly not free.
    arch_listing_seconds: dict[str, float] = {}
    for arch in arches:
        listing_started = time.monotonic()
        result = await _nix.run(
            [
                "eval",
                "--json",
                f"{path}#checks.{arch}",
                "--apply",
                list_checks_expr,
            ],
            retcode=None,
        )
        listing_elapsed = time.monotonic() - listing_started
        if result.returncode != 0:
            arch_listing_errors[arch] = (result.stderr or "").strip() or (
                "evaluation failed"
            )
            continue
        try:
            parsed_checks: list[str] = json.loads(result.stdout)
        except (json.JSONDecodeError, TypeError):
            arch_listing_errors[arch] = "Invalid JSON from nix eval"
            continue
        arch_to_check_names[arch] = [n for n in parsed_checks if isinstance(n, str)]
        arch_listing_seconds[arch] = listing_elapsed

    # 3. Per-check drvPath eval. Each call only sees its own check's
    # closure, so the RAM peak is bounded by the largest single
    # check.
    by_arch: dict[str, list[CheckResult]] = {}
    eval_elapsed_per_arch: dict[str, float] = {a: 0.0 for a in arch_to_check_names}
    for arch in arch_to_check_names:
        arch_seconds = 0.0
        for name in arch_to_check_names[arch]:
            if on_progress is not None:
                on_progress(f"{arch}/{name}")
            started = time.monotonic()
            result = await _nix.run(
                ["eval", "--raw", f"{path}#checks.{arch}.{name}.drvPath"],
                retcode=None,
            )
            elapsed = time.monotonic() - started
            arch_seconds += elapsed
            if result.returncode == 0:
                drv_path = (result.stdout or "").strip() or None
                cr = CheckResult(arch=arch, name=name, drv_path=drv_path, error=None)
            else:
                # The per-check call already returned the real Nix
                # error trace in stderr. No second round trip needed.
                cr = CheckResult(
                    arch=arch,
                    name=name,
                    drv_path=None,
                    error=(result.stderr or "").strip() or "evaluation failed",
                )
            by_arch.setdefault(arch, []).append(cr)
        # Include the per-arch listing call's wall-clock so the
        # suite ``time`` accounts for the full eval work, not just
        # the per-check round trips.
        eval_elapsed_per_arch[arch] = arch_seconds + arch_listing_seconds.get(arch, 0.0)

    # Combine: arches that listed cleanly get their per-check
    # BuildEvents, arches whose listing failed get an error-only
    # `ArchChecks` so the JUnit shows the per-arch failure honestly.
    # All known arches appear (in sorted order) so the output is
    # stable across runs.
    arches_result: list[ArchChecks] = []
    for arch in sorted(set(arch_to_check_names) | set(arch_listing_errors)):
        if arch in arch_listing_errors:
            arches_result.append(ArchChecks(arch=arch, error=arch_listing_errors[arch]))
        else:
            arches_result.append(ArchChecks(arch=arch, checks=by_arch.get(arch, [])))
    # Attribute the initial `attrNames path#checks` call's wall-clock
    # evenly across the arches that made it past the per-arch
    # listing. Without this, a 1-check flake would under-report the
    # suite ``time`` (the arch-listing call's cost disappears). If
    # no arch made it, fold the time into the ``*`` placeholder so
    # it isn't lost.
    if arch_listing_total_seconds > 0:
        healthy_arches = [a for a in arches_result if a.error is None]
        if healthy_arches:
            share = arch_listing_total_seconds / len(healthy_arches)
            for ac in healthy_arches:
                eval_elapsed_per_arch[ac.arch] = (
                    eval_elapsed_per_arch.get(ac.arch, 0.0) + share
                )
        else:
            # No arch survived — record under the wildcard.
            eval_elapsed_per_arch["*"] = (
                eval_elapsed_per_arch.get("*", 0.0) + arch_listing_total_seconds
            )
    return (
        EvalResult(arches=arches_result, path=path),
        eval_elapsed_per_arch,
    )

print_summary(builds, downloaded=0, color_mode='auto')

Print a human-readable, color-coded summary of build results to stderr.

Output includes the number of paths downloaded from cache, cached builds, passed builds, and failed builds. On failure the names of failing builds are listed individually.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	List of build events to summarize.	required
downloaded	int	Number of paths downloaded from binary caches.	0
color_mode	str	One of "auto", "yes", "no" — applied via :func:_apply_color_mode at the top of this function.	'auto'

Source code in src/junix/lib.py

def print_summary(
    builds: list[BuildEvent],
    downloaded: int = 0,
    color_mode: str = "auto",
) -> None:
    """Print a human-readable, color-coded summary of build results to stderr.

    Output includes the number of paths downloaded from cache, cached
    builds, passed builds, and failed builds.  On failure the names of
    failing builds are listed individually.

    Args:
        builds: List of build events to summarize.
        downloaded: Number of paths downloaded from binary caches.
        color_mode: One of ``"auto"``, ``"yes"``, ``"no"`` — applied via
            :func:`_apply_color_mode` at the top of this function.
    """
    _apply_color_mode(color_mode)

    successful = [b for b in builds if b.success and b.stopped]
    failed = [b for b in builds if not b.success]
    cached = [b for b in builds if b.cached]

    print(file=sys.stderr)

    if downloaded:
        print(
            str(colors.info | f"  \u2b07 {downloaded} path(s) downloaded"),
            file=sys.stderr,
        )

    if cached:
        print(
            str(colors.warn | f"  \u007e {len(cached)} build(s) cached"),
            file=sys.stderr,
        )

    for b in successful:
        if not b.cached:
            print(str(colors.success | f"  \u2713 {b.name}"), file=sys.stderr)

    for b in failed:
        print(str(colors.fatal | f"  \u2717 {b.name}"), file=sys.stderr)

    total = len(builds)
    passed = len(successful)
    failed_count = len(failed)

    if failed_count:
        print(
            str(colors.bold & colors.fatal | f"\n  FAILED ({failed_count}/{total}):"),
            file=sys.stderr,
        )
        for b in failed:
            print(
                str(colors.fatal | f"    - {b.name}"),
                file=sys.stderr,
            )
    else:
        print(
            str(colors.bold & colors.success | f"\n  PASSED ({passed}/{total})"),
            file=sys.stderr,
        )

run_nix_build(attrs, store=None, default_name='nix build', log_level=0, expected_count=0, expected_names=None, expected_drv_paths=None, drv_to_name=None, extra_flags=None) async

Run nix build and return collected build events.

Uses --keep-going so Nix attempts every attribute even if some fail. Cached/substituted builds produce no actBuild protocol events; when expected_count is set, the report is padded with synthetic skipped entries so every requested attribute is accounted for.

Per-build wall-clock duration is measured by NixEventHandler between each actBuild start and stop event as the protocol stream is consumed, so each returned BuildEvent.duration reflects the real build wall-clock. The <testsuite time="..."> and <testcase time="..."> attributes in the JUnit report come from those per-build measurements (summed for the suite total).

Parameters:

Name	Type	Description	Default
attrs	list[str]	Flake attribute paths to build (e.g. ["nixpkgs#hello"]).	required
store	str | None	Remote store URL (e.g. "ssh-ng://eu.nixbuild.net").	None
default_name	str	Name for the synthetic build when no actBuild activities were received.	'nix build'
log_level	int	Maximum Nix log level to forward to stderr (default 0 = errors only; higher = more verbose).	0
expected_count	int	Minimum number of builds to report. When Nix caches some builds they produce no protocol events, so this pads the report with synthetic skipped entries.	0
expected_names	list[str] | None	Names for synthetic entries. When provided, padded entries use these names instead of default_name #N.	None
expected_drv_paths	list[str] | list[str | None] | None	Real /nix/store/...drv paths for each requested attr, in the same order as expected_names. When provided, drv_to_name-style pairing is preferred over positional pairing: each actBuild event is renamed by matching its drv_path against this list, and any missing expected name's synthetic entry is created with the correct drv path (so the handler's unbuildable_drv_paths can mark dep-failed targets as failures instead of cached).	None
drv_to_name	dict[str, str] | None	Mapping from drv_path to check name. When provided, build events are renamed before padding so that synthetic entries don't duplicate real ones.	None
extra_flags	list[str] | None	Extra flags to pass to nix build (e.g. ["--no-link"]).	None

Returns:

Type	Description
list[BuildEvent]	Tuple of (builds, error_messages, exit_code, downloaded_count).
list[str]	downloaded_count is the number of store paths fetched from
int	binary caches during this build.

Source code in src/junix/lib.py

async def run_nix_build(
    attrs: list[str],
    store: str | None = None,
    default_name: str = "nix build",
    log_level: int = 0,
    expected_count: int = 0,
    expected_names: list[str] | None = None,
    expected_drv_paths: list[str] | list[str | None] | None = None,
    drv_to_name: dict[str, str] | None = None,
    extra_flags: list[str] | None = None,
) -> tuple[list[BuildEvent], list[str], int, int]:
    """Run nix build and return collected build events.

    Uses ``--keep-going`` so Nix attempts every attribute even if some
    fail.  Cached/substituted builds produce no ``actBuild`` protocol
    events; when ``expected_count`` is set, the report is padded with
    synthetic skipped entries so every requested attribute is accounted
    for.

    Per-build wall-clock duration is measured by
    ``NixEventHandler`` between each ``actBuild`` ``start`` and
    ``stop`` event as the protocol stream is consumed, so each
    returned ``BuildEvent.duration`` reflects the real build
    wall-clock. The ``<testsuite time="...">`` and
    ``<testcase time="...">`` attributes in the JUnit report come
    from those per-build measurements (summed for the suite total).

    Args:
        attrs: Flake attribute paths to build (e.g. ``["nixpkgs#hello"]``).
        store: Remote store URL (e.g. ``"ssh-ng://eu.nixbuild.net"``).
        default_name: Name for the synthetic build when no ``actBuild``
            activities were received.
        log_level: Maximum Nix log level to forward to stderr
            (default 0 = errors only; higher = more verbose).
        expected_count: Minimum number of builds to report.  When Nix
            caches some builds they produce no protocol events, so this
            pads the report with synthetic skipped entries.
        expected_names: Names for synthetic entries.  When provided,
            padded entries use these names instead of ``default_name #N``.
        expected_drv_paths: Real ``/nix/store/...drv`` paths for each
            requested attr, in the same order as ``expected_names``.
            When provided, drv_to_name-style pairing is preferred over
            positional pairing: each actBuild event is renamed by
            matching its ``drv_path`` against this list, and any
            missing expected name's synthetic entry is created with
            the correct drv path (so the handler's
            ``unbuildable_drv_paths`` can mark dep-failed targets as
            failures instead of cached).
        drv_to_name: Mapping from drv_path to check name.  When provided,
            build events are renamed before padding so that synthetic
            entries don't duplicate real ones.
        extra_flags: Extra flags to pass to ``nix build`` (e.g.
            ``["--no-link"]``).

    Returns:
        Tuple of (builds, error_messages, exit_code, downloaded_count).
        ``downloaded_count`` is the number of store paths fetched from
        binary caches during this build.
    """
    handler = NixEventHandler()

    cmd = ["build", "--log-format", "internal-json", "--keep-going"]
    if extra_flags:
        cmd.extend(extra_flags)
    if store:
        cmd.extend(["--store", store])
    cmd.extend(attrs)

    result = await _nix.run(cmd, retcode=None)
    nix_success = result.returncode == 0

    _parse_nix_stderr(result.stderr, handler, log_level=log_level)

    handler.finalize(process_successful=nix_success)

    # handler.builds is a property that creates new objects on each
    # access, so we must capture the list once before renaming.
    builds_list = handler.builds
    expected_pairs = _rename_builds(
        builds_list,
        attrs,
        drv_to_name,
        expected_names,
        expected_drv_paths,
    )

    builds = _ensure_builds(
        builds_list,
        process_successful=nix_success,
        default_name=default_name,
        expected_count=expected_count,
        expected_names=expected_names,
        expected_pairs=expected_pairs,
        unbuildable_drv_paths=handler.unbuildable_drv_paths,
        real_failure_drv_paths=handler.real_failure_drv_paths,
    )

    errors: list[str] = list(handler._errors) if handler._errors else []
    exit_code = 0 if nix_success and _overall_success(builds) else 1
    return builds, errors, exit_code, handler.downloaded_count

translate_stream(stream, suite_name='nix', process_successful=False, default_name='nix build', log_level=0, color_mode='auto')

Read @nix JSON lines from stream and produce JUnit XML.

Non-@nix lines and Nix log messages are forwarded to stderr.

Parameters:

Name	Type	Description	Default
stream	IO[str]	Text stream to read (e.g. sys.stdin).	required
suite_name	str	Value for the <testsuite name="..."> attribute.	'nix'
process_successful	bool	Passed to NixEventHandler.finalize.	False
default_name	str	Name for the synthetic build when no actBuild activities were received.	'nix build'
log_level	int	Maximum Nix log level to forward to stderr (default 0 = errors only; higher = more verbose).	0
color_mode	str	One of "auto", "yes", "no" — applied via :func:_apply_color_mode to the summary printed to stderr.	'auto'

Returns:

Type	Description
tuple[str, int]	Tuple of (xml_string, exit_code).

Source code in src/junix/lib.py

def translate_stream(
    stream: IO[str],
    suite_name: str = "nix",
    process_successful: bool = False,
    default_name: str = "nix build",
    log_level: int = 0,
    color_mode: str = "auto",
) -> tuple[str, int]:
    """Read `@nix` JSON lines from stream and produce JUnit XML.

    Non-`@nix` lines and Nix log messages are forwarded to stderr.

    Args:
        stream: Text stream to read (e.g. sys.stdin).
        suite_name: Value for the `<testsuite name="...">` attribute.
        process_successful: Passed to NixEventHandler.finalize.
        default_name: Name for the synthetic build when no `actBuild`
            activities were received.
        log_level: Maximum Nix log level to forward to stderr
            (default 0 = errors only; higher = more verbose).
        color_mode: One of ``"auto"``, ``"yes"``, ``"no"`` — applied via
            :func:`_apply_color_mode` to the summary printed to stderr.

    Returns:
        Tuple of (xml_string, exit_code).
    """
    handler = NixEventHandler()
    handler.handle_streaming(stream, log_level=log_level)

    handler.finalize(process_successful=process_successful)
    builds = _ensure_builds(
        handler.builds,
        process_successful=not handler._errors,
        default_name=default_name,
    )
    print_summary(builds, downloaded=handler.downloaded_count, color_mode=color_mode)
    xml = report_to_xml(
        builds,
        suite_name=suite_name,
        error_messages=handler._errors or None,
    )
    exit_code = 0 if _overall_success(builds) else 1
    return xml, exit_code