protocol

Parse the @nix JSON protocol emitted by --log-format internal-json.

The protocol is a line-based JSON stream where each line starts with @nix followed by a JSON object.

Activity types:

actRealise      = 102
actCopyPaths    = 103
actBuilds       = 104
actBuild        = 105

Result types:

resFileLinked       = 100
resBuildLogLine     = 101
resUntrustedPath    = 102
resCorruptedPath    = 103
resSetPhase         = 104
resProgress         = 105
resSetExpected      = 106
resPostBuildLogLine = 107

Build result codes:

Built                    = 0
Substituted              = 1
AlreadyValid             = 2
PermanentFailure         = 3
InputRejected            = 4
OutputRejected           = 5
TransientFailure         = 6
CachedFailure            = 7
TimedOut                 = 8
MiscFailure              = 9
DependencyFailed         = 10
LogLimitExceeded         = 11
NotDeterministic         = 12
ResolvesToAlreadyValid   = 13
NoSubstituters           = 14

BuildEvent dataclass

Collected information about one derivation build.

Source code in src/junix/protocol.py

@dataclass
class BuildEvent:
    """Collected information about one derivation build."""

    drv_path: str = ""
    """Store path of the derivation being built."""

    name: str = ""
    """Human-readable name derived from drv_path."""

    log_lines: list[str] = field(default_factory=list)
    """Build log lines captured during the build."""

    phase: str = ""
    """Last reported build phase (e.g. unpackPhase, buildPhase)."""

    started: bool = False
    """Whether the activity start event was received."""

    stopped: bool = False
    """Whether the activity stop event was received."""

    success: bool = True
    """Whether the build completed successfully.

    Defaults to True and set to False only if a failure is detected
    (e.g. process exits non-zero and build never stopped).
    """

    result_code: int | None = None
    """Build result code (0=Built, 1=Substituted, 2=AlreadyValid, etc.).

    None when not received (unknown / not reported by protocol).
    """

    cached: bool = False
    """Whether the build was served from cache (substituted / already-valid).

    True when Nix did not actually compile anything.
    """

    started_at: float | None = None
    """``time.monotonic()`` reading at the ``actBuild`` start event.

    Set when ``NixEventHandler`` sees the ``start`` event for this
    build's activity. ``None`` for synthetic / cached builds that
    never produced a start event.
    """

    stopped_at: float | None = None
    """``time.monotonic()`` reading at the ``actBuild`` stop event.

    Set when ``NixEventHandler`` sees the ``stop`` event for this
    build's activity. ``None`` when the build never produced a stop
    event (process crashed, etc.).
    """

    @property
    def duration(self) -> float | None:
        """Wall-clock duration of the build in seconds, or ``None``.

        Computed as ``stopped_at - started_at`` when both are set.
        The ``actBuild`` start event fires when Nix dispatches the
        build, and the ``stop`` event fires when the build finishes
        — the gap is the real build wall-clock as observed in the
        event stream junix consumes. ``None`` for synthetic / cached
        builds.
        """
        if self.started_at is None or self.stopped_at is None:
            return None
        return self.stopped_at - self.started_at

    failed_dep_drv_path: str | None = None
    """When ``success`` is False because of a *transitive* dep failure
    (not the build itself), the drv path of the failing dep that
    caused the skip.  ``None`` for direct build failures, cached
    results, and successful builds.  Populated by ``_ensure_builds``
    for synthetic entries created for dep-failed targets so the
    JUnit ``<failure>`` can name the root cause and CI viewers
    can group dep-failures separately from direct ones.
    """

cached = False class-attribute instance-attribute

Whether the build was served from cache (substituted / already-valid).

True when Nix did not actually compile anything.

drv_path = '' class-attribute instance-attribute

Store path of the derivation being built.

duration property

Wall-clock duration of the build in seconds, or None.

Computed as stopped_at - started_at when both are set. The actBuild start event fires when Nix dispatches the build, and the stop event fires when the build finishes — the gap is the real build wall-clock as observed in the event stream junix consumes. None for synthetic / cached builds.

failed_dep_drv_path = None class-attribute instance-attribute

When success is False because of a transitive dep failure (not the build itself), the drv path of the failing dep that caused the skip. None for direct build failures, cached results, and successful builds. Populated by _ensure_builds for synthetic entries created for dep-failed targets so the JUnit <failure> can name the root cause and CI viewers can group dep-failures separately from direct ones.

log_lines = field(default_factory=list) class-attribute instance-attribute

Build log lines captured during the build.

name = '' class-attribute instance-attribute

Human-readable name derived from drv_path.

phase = '' class-attribute instance-attribute

Last reported build phase (e.g. unpackPhase, buildPhase).

result_code = None class-attribute instance-attribute

Build result code (0=Built, 1=Substituted, 2=AlreadyValid, etc.).

None when not received (unknown / not reported by protocol).

started = False class-attribute instance-attribute

Whether the activity start event was received.

started_at = None class-attribute instance-attribute

time.monotonic() reading at the actBuild start event.

Set when NixEventHandler sees the start event for this build's activity. None for synthetic / cached builds that never produced a start event.

stopped = False class-attribute instance-attribute

Whether the activity stop event was received.

stopped_at = None class-attribute instance-attribute

time.monotonic() reading at the actBuild stop event.

Set when NixEventHandler sees the stop event for this build's activity. None when the build never produced a stop event (process crashed, etc.).

success = True class-attribute instance-attribute

Whether the build completed successfully.

Defaults to True and set to False only if a failure is detected (e.g. process exits non-zero and build never stopped).

NixEventHandler

Consume @nix protocol lines and produce a list of BuildEvent objects.

Typical usage:

handler = NixEventHandler()
for line in source:
    handler.handle_line(line)
handler.finalize(process_successful=True)
for build in handler.builds:
    ...

Source code in src/junix/protocol.py

class NixEventHandler:
    """Consume `@nix` protocol lines and produce a list of BuildEvent objects.

    Typical usage:

    ```python
    handler = NixEventHandler()
    for line in source:
        handler.handle_line(line)
    handler.finalize(process_successful=True)
    for build in handler.builds:
        ...
    ```
    """

    def __init__(self) -> None:
        self._activities: dict[int, _Activity] = {}
        self._build_activity_ids: list[int] = []
        self._errors: list[str] = []
        self._messages: list[tuple[int, str]] = []  # (level, message)
        self._downloaded_count: int = 0
        # Drv paths Nix explicitly reported as "Cannot build" (i.e. not
        # even attempted). Populated from ``error: Cannot build '...'.``
        # msg events. Used by ``run_nix_build`` to mark dep-failed
        # targets as failures instead of cached/skipped.
        self._unbuildable_drv_paths: set[str] = set()
        # Drv paths of targets that were skipped *because* one of their
        # dependencies failed (``Reason: 1 dependency failed``).  Used
        # by ``run_nix_build`` to flag these specifically as
        # dep-failed (different from a direct build failure).  Kept as
        # an ordered list to preserve Nix's emission order — useful
        # when reasoning about which failing dep caused which skip.
        self._dep_failed_target_drv_paths: list[str] = []
        # Drv paths of builds that failed for reasons *other* than a
        # dependency (e.g. ``builder failed with exit code N``,
        # ``output rejected``, ``TimedOut``).  These are the *cause*
        # of the dep-failed targets: every ``_dep_failed_target_drv_paths``
        # entry in a run is presumed to depend on one of these (or
        # on another drv that depends on one of these).  When there
        # is exactly one entry, ``run_nix_build`` attributes all the
        # dep-failed targets to it.  With multiple entries, the
        # earliest one (top of the list) is used as a best-effort
        # default.
        self._real_failure_drv_paths: list[str] = []

    @property
    def builds(self) -> list[BuildEvent]:
        """Return the collected build events.

        Call finalize() first to properly close any in-flight builds.
        """
        # Dedupe by drv_path. Nix normally emits one ``actBuild`` per
        # build, but some Nix versions (or remote-builder proxying
        # scenarios) may emit two start/stop pairs for the same drv.
        # Keeping the first occurrence preserves the first start
        # timestamp, the first result_code if any, and the first
        # failure classification applied by ``_mark_failed_from_error``;
        # the second pair carries no additional information.
        seen: set[str] = set()
        result: list[BuildEvent] = []
        for aid in self._build_activity_ids:
            be = self._to_build_event(aid)
            if be.drv_path and be.drv_path in seen:
                continue
            if be.drv_path:
                seen.add(be.drv_path)
            result.append(be)
        return result

    @property
    def downloaded_count(self) -> int:
        """Return the number of paths that were downloaded from cache.

        Counts ``actCopyPath`` (type 100) start events, which Nix emits
        for each individual path it downloads from a binary cache.
        """
        return self._downloaded_count

    @property
    def unbuildable_drv_paths(self) -> set[str]:
        """Return the set of drv paths Nix said it could not build.

        Populated from ``error: Cannot build '...'.`` msg events. A drv
        path in this set was either reported as a real build failure
        (``builder failed``) or a dep-failed target (``1 dependency
        failed``). In both cases, every requested target that resolves
        to one of these drv paths must NOT be reported as cached or
        passed — they could not be built in this run.
        """
        return set(self._unbuildable_drv_paths)

    @property
    def dep_failed_target_drv_paths(self) -> list[str]:
        """Return drv paths of targets skipped due to a failed dep.

        These are the targets whose drv path appeared in a
        ``Cannot build '...'. Reason: 1 dependency failed`` msg.
        They did not even start building, and the JUnit ``<failure>``
        for them should name the failing dep so users can click
        through to the root cause.
        """
        return list(self._dep_failed_target_drv_paths)

    @property
    def real_failure_drv_paths(self) -> list[str]:
        """Return drv paths of builds that failed *for their own reasons*.

        These are the drv paths that appeared in ``Cannot build '...'.``
        with a reason other than ``1 dependency failed`` (e.g.
        ``builder failed with exit code N``, ``output rejected``).
        They are the candidates for the *cause* of the
        ``dep_failed_target_drv_paths``: when a run has exactly one
        such drv, all dep-failed targets are attributed to it.
        """
        return list(self._real_failure_drv_paths)

    def handle_line(self, line: str) -> bool:
        """Process one line of text.

        Args:
            line: A single line of text (without trailing newline).

        Returns:
            True if the line was a recognised `@nix` line, False otherwise
            (caller should forward it to stderr).
        """
        if not line.startswith(NIX_LINE_PREFIX):
            return False

        try:
            event = json.loads(line[len(NIX_LINE_PREFIX) :])
        except json.JSONDecodeError:
            # Malformed JSON -- ignore but continue.
            return True

        self._handle_event(event)
        return True

    def handle_stream(self, stream: IO[str]) -> list[str]:
        """Read an entire text stream (e.g. stdin) line by line.

        Non-`@nix` lines are collected and returned so the caller can
        forward them to stderr.

        Args:
            stream: A text-mode iterable (e.g. sys.stdin).

        Returns:
            List of lines that did not start with `@nix`.
        """
        return [line for line in stream if not self.handle_line(line)]

    def handle_streaming(
        self,
        stream: IO[str],
        log_level: int = 0,
        emit: Callable[[str], None] | None = None,
    ) -> None:
        """Read a stream line by line and emit human-readable output immediately.

        Non-``@nix`` lines are emitted via ``emit`` as they are read.  ``@nix``
        log messages and build log/phase events are emitted in human-readable
        form when their level is at or below ``log_level``.

        The streaming behaviour is used by ``junix translate`` so that a long
        build can be watched live.  State is still updated so the JUnit report
        can be produced after the stream ends.

        Args:
            stream: Text stream to read (e.g. ``sys.stdin``).
            log_level: Maximum Nix log level to forward to stderr.  ``msg``
                events are forwarded when their own level is at or below this
                threshold; ``result`` events (build log lines and phases) are
                forwarded when ``log_level >= 3``.
            emit: Callable that receives each line to forward.  Defaults to
                printing to ``sys.stderr`` with newlines preserved.
        """
        if emit is None:

            def _emit(line: str) -> None:
                print(line, end="", file=sys.stderr)

            emit = _emit

        for raw in stream:
            if not raw.startswith(NIX_LINE_PREFIX):
                emit(raw)
                continue

            line = raw.rstrip("\n").rstrip("\r")
            try:
                event = json.loads(line[len(NIX_LINE_PREFIX) :])
            except json.JSONDecodeError:
                # Malformed JSON: ignore, consistent with handle_line.
                continue

            self._handle_event(event)
            self._emit_stream_event(event, log_level=log_level, emit=emit)

    def _emit_stream_event(
        self,
        event: dict[str, Any],
        log_level: int,
        emit: Callable[[str], None],
    ) -> None:
        """Emit a single human-readable line for a parsed ``@nix`` event.

        Args:
            event: Parsed JSON object from an ``@nix`` line.
            log_level: Current verbosity threshold.
            emit: Callback that receives lines to forward.
        """
        action: str = event.get("action", "")
        if action == "msg":
            msg: str = event.get("msg", "")
            level: int = event.get("level", 0)
            clean = self._strip_ansi(msg)
            if clean.startswith("error:") or level <= log_level:
                # Preserve ANSI codes in the raw message so Nix's own
                # colouring is preserved on a TTY.
                emit(msg if msg.endswith("\n") else f"{msg}\n")
        elif action == "result":
            if log_level < 3:
                return
            formatted = self._format_result_event(event)
            if formatted is not None:
                emit(f"{formatted}\n")

    def _format_result_event(self, event: dict[str, Any]) -> str | None:
        """Format a ``result`` event as Nix-style human-readable output.

        Returns ``None`` when the event should not be forwarded (unknown
        activity, non-build activity, or unsupported result type).

        Args:
            event: Parsed ``result`` event.

        Returns:
            Formatted line like ``"<name>> <log line>"``, or ``None``.
        """
        aid: int | None = event.get("id")
        act = self._activities.get(aid) if aid is not None else None
        if act is None or not act.is_build:
            return None

        rtype: int | None = event.get("type")
        if rtype is None:
            return None
        fields: list[Any] = event.get("fields", [])
        if not fields:
            return None

        drv = ""
        if act.fields and isinstance(act.fields[0], str):
            drv = act.fields[0]
        name = store_path_to_name(drv) if drv else act.text or f"<activity-{aid}>"

        # Build log lines and phase changes share the same "<name>> <text>"
        # presentation. Kept as separate branches in case future formatting
        # diverges (e.g. phase labels could be parenthesised).
        if rtype in (RES_BUILD_LOG_LINE, RES_POST_BUILD_LOG_LINE, RES_SET_PHASE):
            return f"{name}> {fields[0]}"
        return None

    def finalize(self, process_successful: bool = True) -> None:
        """Mark any builds that never received a stop as failed.

        Call this once the input stream is exhausted. If the overall Nix
        process failed (process_successful=False), any unstopped build is
        marked as a failure.

        Args:
            process_successful: Whether the overall Nix process succeeded.
        """
        for aid in self._build_activity_ids:
            act = self._activities.get(aid)
            if act is None:
                continue
            if not act.stopped:
                act.stopped = True
                if not process_successful:
                    act.failed = True

    def _handle_event(self, event: dict[str, Any]) -> None:
        action: str = event.get("action", "")

        if action == "start":
            self._on_start(event)
        elif action == "stop":
            self._on_stop(event)
        elif action == "result":
            self._on_result(event)
        elif action == "msg":
            self._on_msg(event)
        # "set" exists in older Nix versions; ignore here.

    def _on_start(self, event: dict[str, Any]) -> None:
        aid: int = event["id"]
        act = _Activity(
            activity_id=aid,
            activity_type=event.get("type", 0),
            text=event.get("text", ""),
            parent=event.get("parent", 0),
            fields=list(event.get("fields", [])),
        )
        if act.is_build:
            act.started_at = time.monotonic()
        self._activities[aid] = act
        if act.is_build:
            self._build_activity_ids.append(aid)
        if act.activity_type == ACT_COPY_PATH:
            self._downloaded_count += 1

    def _on_stop(self, event: dict[str, Any]) -> None:
        aid: int = event["id"]
        act = self._activities.get(aid)
        if act is not None:
            act.stopped = True
            if act.is_build and act.started_at is not None:
                act.stopped_at = time.monotonic()
            # Stop events for build activities carry the result code in fields[0].
            if act.is_build and event.get("fields"):
                try:
                    act.result_code = int(event["fields"][0])
                except (ValueError, IndexError):
                    pass

    def _on_result(self, event: dict[str, Any]) -> None:
        aid: int = event["id"]
        act = self._activities.get(aid)
        if act is None:
            return
        rtype: int = event.get("type", 0)
        fields: list[Any] = list(event.get("fields", []))

        if rtype == RES_BUILD_LOG_LINE or rtype == RES_POST_BUILD_LOG_LINE:
            if fields:
                act.log_lines.append(str(fields[0]))
        elif rtype == RES_SET_PHASE:
            if fields:
                act.phase = str(fields[0])

    @staticmethod
    def _strip_ansi(text: str) -> str:
        """Remove ANSI escape sequences from text.

        Args:
            text: Text that may contain ANSI escape codes.

        Returns:
            Clean text with ANSI escapes removed.
        """
        return re.sub(r"\x1b\[[0-9;]*m", "", text)

    def _on_msg(self, event: dict[str, Any]) -> None:
        # General log messages from the Nix logger.
        msg = event.get("msg", "")
        level = event.get("level", 0)
        clean = self._strip_ansi(msg)
        if clean.startswith("error:"):
            self._errors.append(clean)
            self._mark_failed_from_error(clean)
        else:
            self._messages.append((level, clean))

    def _mark_failed_from_error(self, error_msg: str) -> None:
        """Parse a build error message and classify the failure.

        Nix error messages for build failures contain the derivation path::

            error: Cannot build '/nix/store/xxx-yyy.drv'.
                   Reason: <reason>.

        We extract the derivation path and:
          - add it to ``_unbuildable_drv_paths`` so ``run_nix_build``
            can detect dependent targets that never emitted an
            ``actBuild`` event (Nix skipped them entirely because
            their dep failed);
          - append it to either ``_dep_failed_target_drv_paths``
            (``Reason: 1 dependency failed`` — this drv is a *target*
            whose dep failed) or ``_real_failure_drv_paths``
            (any other reason — this drv is a *cause* of the
            dep-failed targets);
          - mark the corresponding build activity as failed, since
            the ``stop`` event for builds does not carry the
            result code in Nix 2.34.x.
        """
        # Match both single- and double-quoted drv paths.  Nix
        # currently uses single quotes in the ``Cannot build`` msg
        # but the regex tolerates either to avoid breaking the
        # dep-failed detection on a trivial format change.  The
        # bracketed character class is the only thing that varies
        # between formats.
        m = re.search(r"""['"](/nix/store/[^'"]+\.drv)['"]""", error_msg)
        if not m:
            return
        drv_path = m.group(1)
        self._unbuildable_drv_paths.add(drv_path)
        if "1 dependency failed" in error_msg:
            self._dep_failed_target_drv_paths.append(drv_path)
        else:
            self._real_failure_drv_paths.append(drv_path)
        for aid in self._build_activity_ids:
            act = self._activities.get(aid)
            if act is None:
                continue
            if (
                act.fields
                and len(act.fields) > 0
                and isinstance(act.fields[0], str)
                and act.fields[0] == drv_path
            ):
                # Mark every matching activity as failed. Nix
                # normally emits one ``actBuild`` per build, but
                # some Nix versions (or remote-builder proxying
                # scenarios) can emit two start/stop pairs for
                # the same drv.  Marking only the first would
                # leave the second marked as success and produce
                # a false-positive <testcase> in the JUnit
                # report.  The dedup in ``handler.builds`` keeps
                # the first occurrence anyway, so the extra
                # mark here is purely defensive.
                act.failed = True

    def get_messages(self, threshold: int = 0) -> list[str]:
        """Return log messages whose level is at or below the threshold.

        [Nix log levels](https://github.com/NixOS/nix/blob/bebd2f851a304e9fb2e143ce0cbeff577c6a37ac/src/libutil/include/nix/util/error.hh#L39) (lower = more important):

        ```
        0 = lvlError
        1 = lvlWarn
        2 = lvlNotice
        3 = lvlInfo
        4 = lvlTalkative
        5 = lvlChatty
        6 = lvlDebug
        7 = lvlVomit
        ```

        Args:
            threshold: Maximum level to include (default 0 = errors only).

        Returns:
            List of message strings at or below the threshold.
        """
        return [msg for lvl, msg in self._messages if lvl <= threshold]

    def _to_build_event(self, activity_id: int) -> BuildEvent:
        act = self._activities.get(activity_id)
        if act is None:
            return BuildEvent(drv_path="<unknown>", name="<unknown>", success=False)

        drv: str = ""
        if act.fields and isinstance(act.fields[0], str):
            drv = act.fields[0]
        if not drv:
            drv = act.text

        name = store_path_to_name(drv) if drv else f"<activity-{activity_id}>"

        # Use result code when available, otherwise fall back to old logic.
        if act.result_code is not None:
            success = act.result_code in SUCCESS_CODES
            cached = act.result_code in CACHED_CODES
        else:
            # Without a result code, treat stopped-without-error as success (unless forced-failed).
            success = act.stopped and not act.failed
            cached = False

        # If nix didn't emit any log lines for this build (e.g. it was
        # substituted from a cache and the substitution produced no
        # `result` event), fall back to the drv path so the JUnit
        # <system-out> is never empty.  This keeps the report useful
        # even when nix is silent.
        log_lines: list[str] = list(act.log_lines) or [drv]

        return BuildEvent(
            drv_path=drv,
            name=name,
            log_lines=log_lines,
            phase=act.phase,
            started=act.stopped or True,
            stopped=act.stopped,
            success=success,
            result_code=act.result_code,
            cached=cached,
            started_at=act.started_at,
            stopped_at=act.stopped_at,
        )

builds property

Return the collected build events.

Call finalize() first to properly close any in-flight builds.

dep_failed_target_drv_paths property

Return drv paths of targets skipped due to a failed dep.

These are the targets whose drv path appeared in a Cannot build '...'. Reason: 1 dependency failed msg. They did not even start building, and the JUnit <failure> for them should name the failing dep so users can click through to the root cause.

downloaded_count property

Return the number of paths that were downloaded from cache.

Counts actCopyPath (type 100) start events, which Nix emits for each individual path it downloads from a binary cache.

real_failure_drv_paths property

Return drv paths of builds that failed for their own reasons.

These are the drv paths that appeared in Cannot build '...'. with a reason other than 1 dependency failed (e.g. builder failed with exit code N, output rejected). They are the candidates for the cause of the dep_failed_target_drv_paths: when a run has exactly one such drv, all dep-failed targets are attributed to it.

unbuildable_drv_paths property

Return the set of drv paths Nix said it could not build.

Populated from error: Cannot build '...'. msg events. A drv path in this set was either reported as a real build failure (builder failed) or a dep-failed target (1 dependency failed). In both cases, every requested target that resolves to one of these drv paths must NOT be reported as cached or passed — they could not be built in this run.

_emit_stream_event(event, log_level, emit)

Emit a single human-readable line for a parsed @nix event.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	Parsed JSON object from an @nix line.	required
log_level	int	Current verbosity threshold.	required
emit	Callable[[str], None]	Callback that receives lines to forward.	required

Source code in src/junix/protocol.py

def _emit_stream_event(
    self,
    event: dict[str, Any],
    log_level: int,
    emit: Callable[[str], None],
) -> None:
    """Emit a single human-readable line for a parsed ``@nix`` event.

    Args:
        event: Parsed JSON object from an ``@nix`` line.
        log_level: Current verbosity threshold.
        emit: Callback that receives lines to forward.
    """
    action: str = event.get("action", "")
    if action == "msg":
        msg: str = event.get("msg", "")
        level: int = event.get("level", 0)
        clean = self._strip_ansi(msg)
        if clean.startswith("error:") or level <= log_level:
            # Preserve ANSI codes in the raw message so Nix's own
            # colouring is preserved on a TTY.
            emit(msg if msg.endswith("\n") else f"{msg}\n")
    elif action == "result":
        if log_level < 3:
            return
        formatted = self._format_result_event(event)
        if formatted is not None:
            emit(f"{formatted}\n")

_format_result_event(event)

Format a result event as Nix-style human-readable output.

Returns None when the event should not be forwarded (unknown activity, non-build activity, or unsupported result type).

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	Parsed result event.	required

Returns:

Type	Description
str | None	Formatted line like "<name>> <log line>", or None.

Source code in src/junix/protocol.py

def _format_result_event(self, event: dict[str, Any]) -> str | None:
    """Format a ``result`` event as Nix-style human-readable output.

    Returns ``None`` when the event should not be forwarded (unknown
    activity, non-build activity, or unsupported result type).

    Args:
        event: Parsed ``result`` event.

    Returns:
        Formatted line like ``"<name>> <log line>"``, or ``None``.
    """
    aid: int | None = event.get("id")
    act = self._activities.get(aid) if aid is not None else None
    if act is None or not act.is_build:
        return None

    rtype: int | None = event.get("type")
    if rtype is None:
        return None
    fields: list[Any] = event.get("fields", [])
    if not fields:
        return None

    drv = ""
    if act.fields and isinstance(act.fields[0], str):
        drv = act.fields[0]
    name = store_path_to_name(drv) if drv else act.text or f"<activity-{aid}>"

    # Build log lines and phase changes share the same "<name>> <text>"
    # presentation. Kept as separate branches in case future formatting
    # diverges (e.g. phase labels could be parenthesised).
    if rtype in (RES_BUILD_LOG_LINE, RES_POST_BUILD_LOG_LINE, RES_SET_PHASE):
        return f"{name}> {fields[0]}"
    return None

_mark_failed_from_error(error_msg)

Parse a build error message and classify the failure.

Nix error messages for build failures contain the derivation path::

error: Cannot build '/nix/store/xxx-yyy.drv'.
       Reason: <reason>.

We extract the derivation path and

• add it to _unbuildable_drv_paths so run_nix_build can detect dependent targets that never emitted an actBuild event (Nix skipped them entirely because their dep failed);
• append it to either _dep_failed_target_drv_paths (Reason: 1 dependency failed — this drv is a target whose dep failed) or _real_failure_drv_paths (any other reason — this drv is a cause of the dep-failed targets);
• mark the corresponding build activity as failed, since the stop event for builds does not carry the result code in Nix 2.34.x.

Source code in src/junix/protocol.py

def _mark_failed_from_error(self, error_msg: str) -> None:
    """Parse a build error message and classify the failure.

    Nix error messages for build failures contain the derivation path::

        error: Cannot build '/nix/store/xxx-yyy.drv'.
               Reason: <reason>.

    We extract the derivation path and:
      - add it to ``_unbuildable_drv_paths`` so ``run_nix_build``
        can detect dependent targets that never emitted an
        ``actBuild`` event (Nix skipped them entirely because
        their dep failed);
      - append it to either ``_dep_failed_target_drv_paths``
        (``Reason: 1 dependency failed`` — this drv is a *target*
        whose dep failed) or ``_real_failure_drv_paths``
        (any other reason — this drv is a *cause* of the
        dep-failed targets);
      - mark the corresponding build activity as failed, since
        the ``stop`` event for builds does not carry the
        result code in Nix 2.34.x.
    """
    # Match both single- and double-quoted drv paths.  Nix
    # currently uses single quotes in the ``Cannot build`` msg
    # but the regex tolerates either to avoid breaking the
    # dep-failed detection on a trivial format change.  The
    # bracketed character class is the only thing that varies
    # between formats.
    m = re.search(r"""['"](/nix/store/[^'"]+\.drv)['"]""", error_msg)
    if not m:
        return
    drv_path = m.group(1)
    self._unbuildable_drv_paths.add(drv_path)
    if "1 dependency failed" in error_msg:
        self._dep_failed_target_drv_paths.append(drv_path)
    else:
        self._real_failure_drv_paths.append(drv_path)
    for aid in self._build_activity_ids:
        act = self._activities.get(aid)
        if act is None:
            continue
        if (
            act.fields
            and len(act.fields) > 0
            and isinstance(act.fields[0], str)
            and act.fields[0] == drv_path
        ):
            # Mark every matching activity as failed. Nix
            # normally emits one ``actBuild`` per build, but
            # some Nix versions (or remote-builder proxying
            # scenarios) can emit two start/stop pairs for
            # the same drv.  Marking only the first would
            # leave the second marked as success and produce
            # a false-positive <testcase> in the JUnit
            # report.  The dedup in ``handler.builds`` keeps
            # the first occurrence anyway, so the extra
            # mark here is purely defensive.
            act.failed = True

_strip_ansi(text) staticmethod

Remove ANSI escape sequences from text.

Parameters:

Name	Type	Description	Default
text	str	Text that may contain ANSI escape codes.	required

Returns:

Type	Description
str	Clean text with ANSI escapes removed.

Source code in src/junix/protocol.py

@staticmethod
def _strip_ansi(text: str) -> str:
    """Remove ANSI escape sequences from text.

    Args:
        text: Text that may contain ANSI escape codes.

    Returns:
        Clean text with ANSI escapes removed.
    """
    return re.sub(r"\x1b\[[0-9;]*m", "", text)

finalize(process_successful=True)

Mark any builds that never received a stop as failed.

Call this once the input stream is exhausted. If the overall Nix process failed (process_successful=False), any unstopped build is marked as a failure.

Parameters:

Name	Type	Description	Default
process_successful	bool	Whether the overall Nix process succeeded.	True

Source code in src/junix/protocol.py

def finalize(self, process_successful: bool = True) -> None:
    """Mark any builds that never received a stop as failed.

    Call this once the input stream is exhausted. If the overall Nix
    process failed (process_successful=False), any unstopped build is
    marked as a failure.

    Args:
        process_successful: Whether the overall Nix process succeeded.
    """
    for aid in self._build_activity_ids:
        act = self._activities.get(aid)
        if act is None:
            continue
        if not act.stopped:
            act.stopped = True
            if not process_successful:
                act.failed = True

get_messages(threshold=0)

Return log messages whose level is at or below the threshold.

Nix log levels (lower = more important):

0 = lvlError
1 = lvlWarn
2 = lvlNotice
3 = lvlInfo
4 = lvlTalkative
5 = lvlChatty
6 = lvlDebug
7 = lvlVomit

Parameters:

Name	Type	Description	Default
threshold	int	Maximum level to include (default 0 = errors only).	0

Returns:

Type	Description
list[str]	List of message strings at or below the threshold.

Source code in src/junix/protocol.py

def get_messages(self, threshold: int = 0) -> list[str]:
    """Return log messages whose level is at or below the threshold.

    [Nix log levels](https://github.com/NixOS/nix/blob/bebd2f851a304e9fb2e143ce0cbeff577c6a37ac/src/libutil/include/nix/util/error.hh#L39) (lower = more important):

    ```
    0 = lvlError
    1 = lvlWarn
    2 = lvlNotice
    3 = lvlInfo
    4 = lvlTalkative
    5 = lvlChatty
    6 = lvlDebug
    7 = lvlVomit
    ```

    Args:
        threshold: Maximum level to include (default 0 = errors only).

    Returns:
        List of message strings at or below the threshold.
    """
    return [msg for lvl, msg in self._messages if lvl <= threshold]

handle_line(line)

Process one line of text.

Parameters:

Name	Type	Description	Default
line	str	A single line of text (without trailing newline).	required

Returns:

Type	Description
bool	True if the line was a recognised @nix line, False otherwise
bool	(caller should forward it to stderr).

Source code in src/junix/protocol.py

def handle_line(self, line: str) -> bool:
    """Process one line of text.

    Args:
        line: A single line of text (without trailing newline).

    Returns:
        True if the line was a recognised `@nix` line, False otherwise
        (caller should forward it to stderr).
    """
    if not line.startswith(NIX_LINE_PREFIX):
        return False

    try:
        event = json.loads(line[len(NIX_LINE_PREFIX) :])
    except json.JSONDecodeError:
        # Malformed JSON -- ignore but continue.
        return True

    self._handle_event(event)
    return True

handle_stream(stream)

Read an entire text stream (e.g. stdin) line by line.

Non-@nix lines are collected and returned so the caller can forward them to stderr.

Parameters:

Name	Type	Description	Default
stream	IO[str]	A text-mode iterable (e.g. sys.stdin).	required

Returns:

Type	Description
list[str]	List of lines that did not start with @nix.

Source code in src/junix/protocol.py

def handle_stream(self, stream: IO[str]) -> list[str]:
    """Read an entire text stream (e.g. stdin) line by line.

    Non-`@nix` lines are collected and returned so the caller can
    forward them to stderr.

    Args:
        stream: A text-mode iterable (e.g. sys.stdin).

    Returns:
        List of lines that did not start with `@nix`.
    """
    return [line for line in stream if not self.handle_line(line)]

handle_streaming(stream, log_level=0, emit=None)

Read a stream line by line and emit human-readable output immediately.

Non-@nix lines are emitted via emit as they are read. @nix log messages and build log/phase events are emitted in human-readable form when their level is at or below log_level.

The streaming behaviour is used by junix translate so that a long build can be watched live. State is still updated so the JUnit report can be produced after the stream ends.

Parameters:

Name	Type	Description	Default
stream	IO[str]	Text stream to read (e.g. sys.stdin).	required
log_level	int	Maximum Nix log level to forward to stderr. msg events are forwarded when their own level is at or below this threshold; result events (build log lines and phases) are forwarded when log_level >= 3.	0
emit	Callable[[str], None] | None	Callable that receives each line to forward. Defaults to printing to sys.stderr with newlines preserved.	None

Source code in src/junix/protocol.py

def handle_streaming(
    self,
    stream: IO[str],
    log_level: int = 0,
    emit: Callable[[str], None] | None = None,
) -> None:
    """Read a stream line by line and emit human-readable output immediately.

    Non-``@nix`` lines are emitted via ``emit`` as they are read.  ``@nix``
    log messages and build log/phase events are emitted in human-readable
    form when their level is at or below ``log_level``.

    The streaming behaviour is used by ``junix translate`` so that a long
    build can be watched live.  State is still updated so the JUnit report
    can be produced after the stream ends.

    Args:
        stream: Text stream to read (e.g. ``sys.stdin``).
        log_level: Maximum Nix log level to forward to stderr.  ``msg``
            events are forwarded when their own level is at or below this
            threshold; ``result`` events (build log lines and phases) are
            forwarded when ``log_level >= 3``.
        emit: Callable that receives each line to forward.  Defaults to
            printing to ``sys.stderr`` with newlines preserved.
    """
    if emit is None:

        def _emit(line: str) -> None:
            print(line, end="", file=sys.stderr)

        emit = _emit

    for raw in stream:
        if not raw.startswith(NIX_LINE_PREFIX):
            emit(raw)
            continue

        line = raw.rstrip("\n").rstrip("\r")
        try:
            event = json.loads(line[len(NIX_LINE_PREFIX) :])
        except json.JSONDecodeError:
            # Malformed JSON: ignore, consistent with handle_line.
            continue

        self._handle_event(event)
        self._emit_stream_event(event, log_level=log_level, emit=emit)

_Activity dataclass

An in-flight protocol activity being tracked.

Source code in src/junix/protocol.py

@dataclass
class _Activity:
    """An in-flight protocol activity being tracked."""

    activity_id: int = 0
    activity_type: int = 0
    text: str = ""
    parent: int = 0
    fields: list[Any] = field(default_factory=list)
    log_lines: list[str] = field(default_factory=list)
    phase: str = ""
    stopped: bool = False
    failed: bool = False  # forcibly marked as failed by finalize()
    result_code: int | None = None
    """Build result code captured from the stop event fields."""
    started_at: float | None = None
    """``time.monotonic()`` reading at the start event (build activities only)."""
    stopped_at: float | None = None
    """``time.monotonic()`` reading at the stop event (build activities only)."""

    @property
    def is_build(self) -> bool:
        return self.activity_type in (ACT_BUILD,)

    @property
    def is_builds(self) -> bool:
        return self.activity_type in (ACT_BUILDS,)

result_code = None class-attribute instance-attribute

Build result code captured from the stop event fields.

started_at = None class-attribute instance-attribute

time.monotonic() reading at the start event (build activities only).

stopped_at = None class-attribute instance-attribute

time.monotonic() reading at the stop event (build activities only).

store_path_to_name(store_path)

Extract the human-readable derivation name from a Nix store path.

/nix/store/<hash>-<name>[-<version>] -> <name>[-<version>] /nix/store/<hash>-<name>.drv -> <name>.drv

Uses the same logic as Nix's storePathToName().

Parameters:

Name	Type	Description	Default
store_path	str	A Nix store path (e.g. /nix/store/abc123-hello-2.12.1.drv).	required

Returns:

Type	Description
str	The human-readable name portion of the path.

Source code in src/junix/protocol.py

def store_path_to_name(store_path: str) -> str:
    """Extract the human-readable derivation name from a Nix store path.

    `/nix/store/<hash>-<name>[-<version>]` -> `<name>[-<version>]`
    `/nix/store/<hash>-<name>.drv`          -> `<name>.drv`

    Uses the same logic as [Nix's `storePathToName()`](https://github.com/NixOS/nix/blob/bebd2f851a304e9fb2e143ce0cbeff577c6a37ac/src/libmain/progress-bar.cc#L34-L39).

    Args:
        store_path: A Nix store path (e.g. `/nix/store/abc123-hello-2.12.1.drv`).

    Returns:
        The human-readable name portion of the path.
    """
    base = os.path.basename(store_path.rstrip("/"))
    idx = base.find("-")
    return base[idx + 1 :] if idx != -1 else base