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.
    """

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.

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.

stopped = False class-attribute instance-attribute

Whether the activity stop event was received.

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

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

        Call finalize() first to properly close any in-flight builds.
        """
        return [self._to_build_event(aid) for aid in self._build_activity_ids]

    @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

    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 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) -> 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) -> 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", [])),
        )
        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) -> None:
        aid: int = event["id"]
        act = self._activities.get(aid)
        if act is not None:
            act.stopped = True
            # 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) -> None:
        aid: int = event["id"]
        act = self._activities.get(aid)
        if act is None:
            return
        rtype: int = event.get("type", 0)
        fields: list = 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) -> 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 mark the matching build as failed.

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

            error: Cannot build '/nix/store/xxx-yyy.drv'.

        We extract the derivation path and mark the corresponding build
        activity as failed, since the ``stop`` event for builds does not
        carry the result code in Nix 2.34.x.
        """
        m = re.search(r"'(/nix/store/[^']+\.drv)'", error_msg)
        if not m:
            return
        drv_path = m.group(1)
        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
            ):
                act.failed = True
                break

    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

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

builds property

Return the collected build events.

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

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.

_mark_failed_from_error(error_msg)

Parse a build error message and mark the matching build as failed.

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

error: Cannot build '/nix/store/xxx-yyy.drv'.

We extract the derivation path and 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 mark the matching build as failed.

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

        error: Cannot build '/nix/store/xxx-yyy.drv'.

    We extract the derivation path and mark the corresponding build
    activity as failed, since the ``stop`` event for builds does not
    carry the result code in Nix 2.34.x.
    """
    m = re.search(r"'(/nix/store/[^']+\.drv)'", error_msg)
    if not m:
        return
    drv_path = m.group(1)
    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
        ):
            act.failed = True
            break

_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)]

_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 = 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."""

    @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.

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