Reusable Wire Types

Shared types used as fields within requests and responses. These are not protocol operations themselves — they are WireModel or WireString subclasses that appear as field types in the operations above.

---

WireString Types

Single length-prefixed UTF-8 strings on the wire.

StorePath

A Nix store path: one class, holding the base name.

Nix holds the base name and nothing else. src/libstore/include/nix/store/path.hh:

class StorePath
{
    std::string baseName;

The store directory is not part of the value. It belongs to the store, at StoreDirConfig::storeDir, and printStorePath is what puts it in front.

This module used to hold a second, different StorePath: a WireScalar, so a str, holding the whole path -- while pynixd.store_path held the one below. The two were not ==, did not hash alike, and a dict keyed by one never answered the other. Issue #3 holds the measurement. There is one class now, and pynixd.store_path re-exports it.

It is not a `str`. Measured before the change: no call site depended on the str-ness. Every str method called on a store path in the repository was in tests/functional, and all of them were on stdout.strip() from a subprocess. A str subclass also makes the wrong thing easy -- path[:10] slices text that means nothing -- and it hides the question this class exists to answer, which is which form a path is in.

class nix_daemon_protocol.store_path.StorePath(path='', extrainfo=None)

A Nix store path, held as its base name.

The store directory is stripped on construction and put back by __str__, so a value never carries it twice and never carries it in one place and not another.

Parameters:

• path (str | StorePath)

• extrainfo (Any)

to_wire()

The whole path, which is what the daemon wire carries.

CommonProto::Serialise<StorePath>::write of Nix calls store.printStorePath, and that is this. narinfo's References: calls to_string() instead, which is name below -- a different codec, not a differently annotated field.

Return type:

str

to_json_value()

What JSON carries: the base name.

adl_serializer<nix::StorePath>::to_json, src/libstore/path.cc:95 of Nix, writes storePath.to_string(), and from_json builds one straight back from that string. So a Realisation, which travels as JSON, carries abc-foo where the binary wire carries /nix/store/abc-foo.

This is the whole of issue #4 in one pair of methods. One value, two codecs, and the form belongs to the codec rather than to the value or to the field that holds it.

Return type:

str

property path: str

The whole path. The spelling most of pynixd already uses.

base()

The base name, without the store directory.

Return type:

str

property name: str

what Nix's to_string() answers.

Type:

The base name

hash_part()

The 32-character hash in front of the name.

Return type:

str

base_name()

The readable part, after the hash.

Return type:

str

to_path()

The whole path, as a pathlib.Path.

Return type:

Path

with_store_prefix()

Return self: __str__ already puts the store directory first.

Return type:

StorePath

DerivedPath

DerivedPath — a Nix derived path expression, plain string on the wire.

class nix_daemon_protocol.derived_path.DerivedPath(value='', *, derived_path=None)

A Nix derived path — drv!out format, plain string on wire.

Parameters:

• value (str)

• derived_path (str | None)

Return type:

DerivedPath

property value: str

Compatibility spelling for the canonical string value.

DrvOutput

DrvOutput — a derivation output identifier as a canonical wire scalar.

Wire format: plain string like sha256:abc123!out.

That is the shape with no feature, and it is one of two. WorkerProto::Serialise<DrvOutput> at worker-protocol.cc:544 of the master branch raises when realisation-with-path-not-hash is off, and writes a derivation path and an output name when it is on. The hash form does not exist in that shape at all: a DrvOutput there names the derivation that made the output, and not a hash of it.

KeyedDrvOutput in keyed_drv_output.py holds the other shape. Issue #14.

class nix_daemon_protocol.drv_output.DrvOutput(value='', *, drv_hash=None, output_name=None)

DrvOutput on the wire — "drvHash!outputName" format.

Uses the same from_str/to_str pattern as Signature but with ! as the delimiter instead of :.

Parameters:

• value (str)

• drv_hash (str | None)

• output_name (str | None)

Return type:

DrvOutput

property drv_hash: str

The derivation hash before the first exclamation mark.

property output_name: str

The output name after the first exclamation mark.

to_str()

Compatibility spelling for the canonical string value.

Return type:

str

ContentAddress

ContentAddress — a content-addressed store path identifier.

Wire format: plain string like text:sha256 or fixed:r:sha256.

class nix_daemon_protocol.content_address.ContentAddress(value='', *, content_address=None)

Content address method with algorithm — plain string on the wire.

Parameters:

• value (str)

• content_address (str | None)

Return type:

ContentAddress

property value: str

Compatibility spelling for the canonical string value.

NARHash

NARHash — base16-encoded NAR SHA256 hash as a WireString.

class nix_daemon_protocol.nar_hash.NARHash(value='', *, hash=None)

Base16-encoded NAR SHA256 hash — no algorithm prefix on wire.

WorkerProto::Serialise<UnkeyedValidPathInfo>::write at worker-protocol.cc:356 of Nix writes narHash.to_string(HashFormat::Base16, false), and false is what leaves the name of the algorithm out.

A source of this value may still carry the name, and this takes it off. LocalStore of Nix writes sha256:<digest> into the narHash column of its database, so a fast path that reads that column and answers a client sends a string that nix-daemon never sends. The client accepts it, because Hash::parseAny at worker-protocol.cc:339 reads both forms, so nothing failed and the two answers differed. tests/parity of pynixd is what saw it.

Parameters:

• value (str)

• hash (str | None)

Return type:

NARHash

property hash: str

Compatibility spelling for the canonical string value.

Signature

Signature — a Nix cryptographic signature as a canonical wire scalar.

class nix_daemon_protocol.signature.Signature(value='', *, name=None, signature=None)

A Nix signature — "name:signature" on the wire.

Parameters:

• value (str)

• name (str | None)

• signature (str | None)

Return type:

Signature

property name: str

The signing key name before the first colon.

property signature: str

The encoded signature after the first colon.

to_str()

Compatibility spelling for the canonical string value.

Return type:

str

---

WireModel Types

BasicDerivation

BasicDerivation — wire mirror of a Nix derivation.

nix_daemon_protocol.basic_derivation.structured_attrs(env)

The attributes of a __structuredAttrs derivation, or None for any other.

Such a derivation keeps every attribute in __json, preferLocalBuild and requiredSystemFeatures included, and its environment names nothing else. getBoolAttr and getStringSetAttr at derivation-options.cc:36 then read the JSON alone. nixidae/nixidae#1003.

Parameters:

env (Mapping[str, str])

Return type:

dict[str, Any] | None

nix_daemon_protocol.basic_derivation.bool_attr(env, name, *, default)

A Boolean attribute, as getBoolAttr at derivation-options.cc:36 reads it.

Parameters:

• env (Mapping[str, str])

• name (str)

• default (bool)

Return type:

bool

nix_daemon_protocol.basic_derivation.required_features_of(env)

requiredSystemFeatures: words in the environment, a list of strings in __json.

Parameters:

env (Mapping[str, str])

Return type:

set[str]

class nix_daemon_protocol.basic_derivation.BasicDerivation(*, outputs=<factory>, input_srcs=<factory>, platform, builder, args=<factory>, env=<factory>, is_dynamic=False)

Wire mirror of BasicDerivation.

Parameters:

• outputs (Annotated[dict[str, DerivationOutput], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• input_srcs (Annotated[set[StorePath], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• platform (str)

• builder (str)

• args (Annotated[list[str], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• env (Annotated[dict[str, str], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• is_dynamic (Annotated[bool, VersionMeta(min_version=None, max_version=None, serialize=False, deserialize=False, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

set_required_system_features(features)

Write requiredSystemFeatures where the builder reads it: __json or the environment.

Parameters:

features (Iterable[str])

Return type:

None

output_paths()

Each output name, and the store path it names.

DerivationOutput.path is the wire string. An output with no path yet -- deferred, or floating content-addressed -- gives the empty StorePath, which is falsy, so a caller can drop it.

Return type:

dict[str, StorePath]

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

DerivationOutput

DerivationOutput — wire representation of a single derivation output.

class nix_daemon_protocol.derivation_output.OutputKind(*values)

Classification of a single derivation output.

class nix_daemon_protocol.derivation_output.DerivationOutput(*, path, method='', hash_digest='')

Wire mirror of DerivationOutput. Three string fields on the wire.

Parameters:

• path (str)

• method (str)

• hash_digest (str)

property kind: OutputKind

Classify this output based on wire protocol fields.

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

UnkeyedValidPathInfo

UnkeyedValidPathInfo — store path metadata without the path key.

class nix_daemon_protocol.path_info.UnkeyedValidPathInfo(*, deriver=None, nar_hash, references, registration_time, nar_size, ultimate, sigs, ca)

Wire mirror of UnkeyedValidPathInfo.

Parameters:

• deriver (Annotated[StorePath | None, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• nar_hash (NARHash)

• references (set[StorePath])

• registration_time (Time)

• nar_size (int)

• ultimate (bool)

• sigs (set[Signature])

• ca (ContentAddress)

to_valid_path_info(path)

Wrap this UnkeyedValidPathInfo with a StorePath into ValidPathInfo.

Parameters:

path (StorePath)

Return type:

ValidPathInfo

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ValidPathInfo

ValidPathInfo — StorePath + UnkeyedValidPathInfo as nested fields.

class nix_daemon_protocol.valid_path_info.ValidPathInfo(*, path, info)

Wire mirror of ValidPathInfo.

Wire order: path then all info fields inline. WireModel serializes nested WireModel fields inline, so this produces the correct flat wire format.

Parameters:

• path (StorePath)

• info (UnkeyedValidPathInfo)

async bytes_wire()

Serialize this ValidPathInfo to bytes (in-memory BytesWriter is sync).

Return type:

bytes

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

BuildResult

BuildResult and related types — build outcome on the Nix daemon wire.

nix_daemon_protocol.build_result.DUMMY_DRV_HASH = 'sha256:0000000000000000000000000000000000000000000000000000000000000000'

Hash::dummy in base 16, the hash Nix writes into an old-shape DrvOutput key.

class nix_daemon_protocol.build_result.BuildResultStatus(*values)

Build result status codes from nix daemon protocol.

class nix_daemon_protocol.build_result.BuildMode(*values)

Build mode flags from nix daemon protocol.

class nix_daemon_protocol.build_result.BuiltOutput(out_path=<factory>, ca='', hash='', hash_algo='', nar_hash='', nar_size=0, reference='')

A built output - either plain path or content-addressed (CA) format.

Parameters:

• out_path (StorePath)

• ca (str)

• hash (str)

• hash_algo (str)

• nar_hash (str)

• nar_size (int)

• reference (str)

class nix_daemon_protocol.build_result.BuildResult(*, status=0, error_msg='', times_built=None, is_non_deterministic=None, start_time=None, stop_time=None, cpu_user=<factory>, cpu_system=<factory>, built_outputs=None, built_outputs_by_name=None)

Nix daemon protocol BuildResult.

Fields present based on protocol version: - All versions: status, error_msg - >= 1.29: times_built, is_non_deterministic, start_time, stop_time - >= 1.37: cpu_user, cpu_system - >= 1.28: built_outputs (dict[str, str])

Parameters:

• status (int)

• error_msg (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

• times_built (Annotated[int | None, VersionMeta(min_version=285, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• is_non_deterministic (Annotated[int | None, VersionMeta(min_version=285, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• start_time (Annotated[int | None, VersionMeta(min_version=285, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• stop_time (Annotated[int | None, VersionMeta(min_version=285, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• cpu_user (Annotated[OptMicroseconds, VersionMeta(min_version=293, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• cpu_system (Annotated[OptMicroseconds, VersionMeta(min_version=293, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• built_outputs (Annotated[dict[str, Realisation] | None, VersionMeta(min_version=284, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=frozenset({'realisation-with-path-not-hash'}), text_errors=strict)])

• built_outputs_by_name (Annotated[dict[str, UnkeyedRealisation] | None, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=frozenset({'realisation-with-path-not-hash'}), unless_features=None, text_errors=strict)])

error_msg: str

What the daemon said about the failure, byte for byte.

BuildResult.errorMsg is a std::string that the daemon forwards without validating, and pynixd quotes LogError.msg into it on the queue.fail path -- both already carry bytes no UTF-8 decoder accepts. Strict decoding would move the crash of issue nixidae/nixidae#662 from the log stream to the result that follows it.

built_outputs_by_name: dict[str, UnkeyedRealisation] | None

The map that the feature shape carries, keyed by the output name alone.

The key of built_outputs is a whole DrvOutput; this one names the output of the derivation that the answer is already about, so it needs no derivation in the key.

The two spell a path differently. This one carries a whole store path, because the wire writes it as a StorePath. The JSON of Realisation carries the bare <hash>-<name>. _whole_path and _bare_path are the translation, and ca:build-cache measured what skipping it costs.

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

realised_outputs()

Every output this result realised, keyed by the output name.

One accessor for the two wire shapes, so a caller needs no branch. built_outputs keys by a whole DrvOutput, which is "<drvHash>!<output>", and built_outputs_by_name keys by the output name alone. This takes the part after the ! in the first case.

A realisation of the feature shape carries no id. The request already named the derivation, so worker-protocol.cc:268 writes the output name as the key and an UnkeyedRealisation as the value. The Realisation this builds therefore has an empty id, and a caller that needs the id must build it from the derivation it already holds. No caller in this repository does: each one reads out_path and the key. Issue #14.

Return type:

dict[str, Realisation]

wire_status()

The status byte to send, which is not always the one held.

A hash mismatch is a kind of output rejection, which is the mapping Nix chose and the one a client expects. Everything else out of range becomes a plain failure: the detail belongs in error_msg, which has no such limit, and a status a client cannot read carries no detail at all.

Return type:

int

for_the_wire(features=None)

This result, with a status a client can decode and a map it can read.

Call this before sending a result to a client. It is a method and not a to_writer override on purpose: experimental_compiled refuses to compile a model that overrides its codec (_can_compile), and BuildResult is the model its own tests compile as the example. A correctness fix that quietly turned off the fast path for the hottest model on the wire would be a poor trade, so the knowledge of what is wire-safe lives here and the caller decides when to apply it.

A proxy reads a result on one connection and writes it on another, and the two negotiate their features apart. A backend that offers realisation-with-path-not-hash fills built_outputs_by_name and leaves built_outputs at None; a client that offers nothing then reads built_outputs, and a None where a map belongs raises in the writer. Give features the set of the peer this result is going to, and this fills the field that peer will read.

The fill from the feature shape to the old one carries a dummy hash. The old key is a whole DrvOutput, and the feature shape carries no hash. Nix writes sha256:<64 zeros>!<output> there, because an old client reads only the output name and the path (worker-protocol.cc:313). A key without the ! makes Nix 2.35's reader throw "Invalid derivation output id" (worker-protocol.cc:272). Issue #14, nixidae/nixidae#1003.

Parameters:

features (frozenset[str] | None)

Return type:

BuildResult

KeyedBuildResult

KeyedBuildResult — a BuildResult paired with its DerivedPath primary key.

class nix_daemon_protocol.keyed_build_result.KeyedBuildResult(*, path, result)

A BuildResult together with its DerivedPath key.

Wire order: DerivedPath string then BuildResult fields. Both are WireModel subclasses — the generic engine writes them inline.

Parameters:

• path (DerivedPath)

• result (BuildResult)

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

OptMicroseconds

OptMicroseconds — tagged-optional uint64 for microseconds.

class nix_daemon_protocol.opt_microseconds.OptMicroseconds(*, tag=0, value=None)

Optional microseconds — [tag uint64][value uint64 if tag==1].

tag=1 = present, tag=0 = absent. value is only on the wire when tag==1.

Parameters:

• tag (int)

• value (Annotated[int | None, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=~nix_daemon_protocol.opt_microseconds.OptMicroseconds.<lambda>, needs_features=None, unless_features=None, text_errors=strict)])

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Realisation

Realisation — JSON-encoded derivation realisation on the Nix daemon wire.

class nix_daemon_protocol.realisation.Realisation(*, id=<factory>, outPath=None, signatures=<factory>, dependentRealisations=<factory>)

A Realisation on the Nix daemon wire.

Wire format: length-prefixed UTF-8 containing a JSON object. id and dependentRealisations keys/values are plain strings on the wire (e.g. "sha256-abc123!out"), matching the old framework's JSON serialization.

Parameters:

• id (Annotated[DrvOutput, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• outPath (Annotated[StorePath | None, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• signatures (Annotated[list[Signature], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• dependentRealisations (Annotated[dict[str, str], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

async classmethod from_reader(ctx)

Read all non-ClassVar fields in declaration order.

Parameters:

ctx (ReadContext)

async to_writer(ctx)

Write all non-ClassVar fields in declaration order.

Parameters:

ctx (WriteContext)

Return type:

None

to_json(**kwargs)

Serialize to JSON string.

Uses Pydantic's model_dump_json.

Return type:

str

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Time

Time and TimeSpan — wire types for timestamps and durations.

class nix_daemon_protocol.wire_time.Time(value=0, *, ts=None)

Unix timestamp in seconds on the wire. Exposes as datetime property.

Parameters:

• value (int)

• ts (int | None)

Return type:

Time

property ts: int

Compatibility spelling for the raw Unix timestamp.

class nix_daemon_protocol.wire_time.TimeSpan(value=0, *, seconds=None)

Timespan in seconds on the wire. Exposes as timedelta property.

Parameters:

• value (int)

• seconds (int | None)

Return type:

TimeSpan

property seconds: int

Compatibility spelling for the raw duration in seconds.

---

Stderr Stream Types

The Nix daemon stderr stream uses a tagged-union wire format: [uint64 code][message body]...[uint64 STDERR_LAST].

WireModel types for the Nix daemon stderr/log stream.

The stderr stream uses a tagged-union wire format:

[uint64 code][message body][uint64 code][message body]...[uint64 STDERR_LAST]

Each message type has a code field that is written on the wire (serialize=True) but skipped during body reading (deserialize=False) because WireLogs.from_reader dispatches on code before delegating to the individual message's from_reader.

Every `str` field here is opaque bytes, and never validated text. The wire string is a length-prefixed byte run (readString moves bytes, and worker-protocol-connection.cc:53 hands them to the sink unread), the builder's output is split on n alone and emitted as a raw std::string (build-log.cc:39), and the client prints what it got (terminal.cc:39 walks invalid UTF-8 one byte at a time). A builder prints whatever it prints -- a gzip stream in the middle of nixpkgs fixup output -- and the daemon forwards it without looking. Strict UTF-8 decoding fails the whole build on those bytes, so each field below takes text_errors and round-trips them. Issue nixidae/nixidae#662.

class nix_daemon_protocol.logs.TraceLine(*, pos=0, hint='')

A single trace entry inside LogError.

Parameters:

• pos (int)

• hint (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class nix_daemon_protocol.logs.ActivityField(*, type=FieldType.INT, valint=None, valstr=None)

A single typed field in LogStartActivity or LogResult.

Wire format: uint64 type then either uint64 valint (if INT) or string valstr (if STRING).

Parameters:

• type (FieldType)

• valint (Annotated[int | None, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=~nix_daemon_protocol.logs.ActivityField.<lambda>, needs_features=None, unless_features=None, text_errors=strict)])

• valstr (Annotated[str | None, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=~nix_daemon_protocol.logs.ActivityField.<lambda>, needs_features=None, unless_features=None, text_errors=surrogateescape)])

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class nix_daemon_protocol.logs.LogNext(*, code=1869376871, text='')

STDERR_NEXT — a log line from the daemon.

Parameters:

• code (Annotated[int, VersionMeta(min_version=None, max_version=None, serialize=True, deserialize=False, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• text (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class nix_daemon_protocol.logs.LogStartActivity(*, code=1398035028, act_id=0, level=0, type=0, text='', fields=<factory>, parent=0)

STDERR_START_ACTIVITY — begin a tracked activity.

Parameters:

• code (Annotated[int, VersionMeta(min_version=None, max_version=None, serialize=True, deserialize=False, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• act_id (int)

• level (int)

• type (int)

• text (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

• fields (Annotated[list[ActivityField], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• parent (int)

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class nix_daemon_protocol.logs.LogStopActivity(*, code=1398034256, act_id=0)

STDERR_STOP_ACTIVITY — end a tracked activity.

Parameters:

• code (Annotated[int, VersionMeta(min_version=None, max_version=None, serialize=True, deserialize=False, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• act_id (int)

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class nix_daemon_protocol.logs.LogResult(*, code=1381190740, act_id=0, result_type=0, fields=<factory>)

STDERR_RESULT — result data for an activity.

Parameters:

• code (Annotated[int, VersionMeta(min_version=None, max_version=None, serialize=True, deserialize=False, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• act_id (int)

• result_type (int)

• fields (Annotated[list[ActivityField], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class nix_daemon_protocol.logs.LogError(*, code=1668838512, type='', level=0, name='', msg='', have_pos=0, traces=<factory>)

STDERR_ERROR — the daemon is reporting an error.

Parameters:

• code (Annotated[int, VersionMeta(min_version=None, max_version=None, serialize=True, deserialize=False, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

• type (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

• level (int)

• name (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

• msg (Annotated[str, VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=surrogateescape)])

• have_pos (int)

• traces (Annotated[list[TraceLine], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

async nix_daemon_protocol.logs.read_stream(ctx)

Yield stderr/log messages until STDERR_LAST.

Parameters:

ctx (ReadContext)

Return type:

AsyncIterator[LogMessage]

async nix_daemon_protocol.logs.drain(ctx, raise_on_error=True)

Read and discard all stderr/log messages until STDERR_LAST.

The exception carries the message of the daemon, and nothing in front of it. A proxy sends that text on to its own client, which prints it after the word "error:". This put "Backend error: " in front, so a client of pynixd read error: Backend error: Cannot delete path '...', and a client of nix-daemon read error: Cannot delete path '...'. The class of the exception already says where the message came from.

Parameters:

• ctx (ReadContext)

• raise_on_error (bool)

Return type:

LogError | None

class nix_daemon_protocol.logs.WireLogs(*, messages=<factory>)

Container for a complete stderr stream.

from_reader reads a tagged-union stream (dispatching on code), stopping at STDERR_LAST or after LogError.

to_writer writes each message (which includes its code) followed by STDERR_LAST.

Parameters:

messages (Annotated[list[LogNext | LogStartActivity | LogStopActivity | LogResult | LogError], VersionMeta(min_version=None, max_version=None, serialize=None, deserialize=None, wire_depends_on=None, needs_features=None, unless_features=None, text_errors=strict)])

add(msg)

Append a log message to the stream.

Parameters:

msg (LogNext | LogStartActivity | LogStopActivity | LogResult | LogError)

Return type:

None

async classmethod from_reader(ctx)

Read tagged-union stderr stream.

Parameters:

ctx (ReadContext)

Return type:

WireLogs

model_config = {'arbitrary_types_allowed': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

async to_writer(ctx)

Write each message body (including its code) then STDERR_LAST.

Parameters:

ctx (WriteContext)

Return type:

None