topmark.config.machine.envelopes

topmark / config / machine / envelopes

Envelope builders for config-related machine-readable output.

Envelopes are small, JSON-friendly Python mappings that follow TopMark's machine output conventions:

• JSON: a single envelope object containing meta plus one or more named payloads.
• NDJSON: one JSON object per line, each carrying the kind and meta envelope.

This module is pure (no I/O, no ConsoleLike) and delegates payload construction to topmark.config.machine.payloads.

When NDJSON needs one diagnostic per line, this module flattens staged config-validation logs at the output boundary and streams the resulting compatibility diagnostics.

build_config_json_envelope

build_config_json_envelope(
    *,
    config,
    resolved_toml,
    meta,
    cfg_provenance_payload=None,
)

Build the JSON envelope for a FrozenConfig snapshot.

Shape

{"meta": , "config": } or, when provenance is requested: { "meta": , "config_provenance": , "config": , }

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Immutable runtime configuration to serialize.	required
resolved_toml	ResolvedTopmarkTomlSources	Resolved TOML sources used to build the optional layered provenance export.	required
meta	MetaPayload	Machine-output metadata (tool/version).	required
cfg_provenance_payload	TomlProvenancePayload | None	Optional machine-readable layered config provenance payload to include before the final flattened config payload.	None

Returns:

Type	Description
dict[str, object]	A JSON-envelope mapping (not yet serialized).

Source code in src/topmark/config/machine/envelopes.py

def build_config_json_envelope(
    *,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    meta: MetaPayload,
    cfg_provenance_payload: TomlProvenancePayload | None = None,
) -> dict[str, object]:
    """Build the JSON envelope for a `FrozenConfig` snapshot.

    Shape:
        {"meta": <MetaPayload>, "config": <ConfigPayload>}
        or, when provenance is requested:
        {
            "meta": <MetaPayload>,
            "config_provenance": <TomlProvenancePayload>,
            "config": <ConfigPayload>,
        }

    Args:
        config: Immutable runtime configuration to serialize.
        resolved_toml: Resolved TOML sources used to build the optional layered
            provenance export.
        meta: Machine-output metadata (tool/version).
        cfg_provenance_payload: Optional machine-readable layered config provenance
            payload to include before the final flattened config payload.

    Returns:
        A JSON-envelope mapping (not yet serialized).
    """
    payload: ConfigPayload = build_config_payload(
        config,
        resolved_toml=resolved_toml,
    )
    if cfg_provenance_payload is None:
        return build_json_envelope(meta=meta, config=payload)

    return build_json_envelope(
        meta=meta,
        config_provenance=cfg_provenance_payload,
        config=payload,
    )

iter_config_ndjson_records

iter_config_ndjson_records(
    *,
    config,
    resolved_toml,
    meta,
    cfg_provenance_payload=None,
)

Iterate NDJSON records for a FrozenConfig snapshot.

Shapes

• without provenance:
• with provenance: 1) {"kind": "config_provenance", "meta": ..., "config_provenance": ...} 2) {"kind": "config", "meta": ..., "config": ...}

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Immutable runtime configuration to serialize.	required
resolved_toml	ResolvedTopmarkTomlSources	Resolved TOML sources used to build the optional layered provenance export.	required
meta	MetaPayload	Machine-output metadata (tool/version).	required
cfg_provenance_payload	TomlProvenancePayload | None	Optional machine-readable layered config provenance payload to emit before the final flattened config record.	None

Yields:

Type	Description
dict[str, object]	An iterable of NDJSON record mappings (not yet serialized).

Source code in src/topmark/config/machine/envelopes.py

def iter_config_ndjson_records(
    *,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    meta: MetaPayload,
    cfg_provenance_payload: TomlProvenancePayload | None = None,
) -> Iterator[dict[str, object]]:
    """Iterate NDJSON records for a `FrozenConfig` snapshot.

    Shapes:
        - without provenance:
            {"kind": "config", "meta": <MetaPayload>, "config": <ConfigPayload>}
        - with provenance:
            1) {"kind": "config_provenance", "meta": ..., "config_provenance": ...}
            2) {"kind": "config", "meta": ..., "config": ...}

    Args:
        config: Immutable runtime configuration to serialize.
        resolved_toml: Resolved TOML sources used to build the optional layered
            provenance export.
        meta: Machine-output metadata (tool/version).
        cfg_provenance_payload: Optional machine-readable layered config provenance
            payload to emit before the final flattened config record.

    Yields:
        An iterable of NDJSON record mappings (not yet serialized).
    """
    if cfg_provenance_payload is not None:
        yield build_ndjson_record(
            kind=ConfigKind.CONFIG_PROVENANCE,
            meta=meta,
            container_key=ConfigKey.CONFIG_PROVENANCE,
            payload=cfg_provenance_payload,
        )

    payload: ConfigPayload = build_config_payload(
        config,
        resolved_toml=resolved_toml,
    )

    yield build_ndjson_record(
        kind=ConfigKind.CONFIG,
        meta=meta,
        payload=payload,
    )

build_config_diagnostics_json_envelope

build_config_diagnostics_json_envelope(*, config, meta)

Build the JSON envelope for config diagnostics.

Shape

{"meta": , "config_diagnostics": }

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Immutable runtime configuration providing staged validation logs that are flattened for machine-readable output.	required
meta	MetaPayload	Machine-output metadata (tool/version).	required

Returns:

Type	Description
dict[str, object]	A JSON-envelope mapping (not yet serialized).

Source code in src/topmark/config/machine/envelopes.py

def build_config_diagnostics_json_envelope(
    *,
    config: FrozenConfig,
    meta: MetaPayload,
) -> dict[str, object]:
    """Build the JSON envelope for config diagnostics.

    Shape:
        {"meta": <MetaPayload>, "config_diagnostics": <ConfigDiagnosticsPayload>}

    Args:
        config: Immutable runtime configuration providing staged validation
            logs that are flattened for machine-readable output.
        meta: Machine-output metadata (tool/version).

    Returns:
        A JSON-envelope mapping (not yet serialized).
    """
    payload: ConfigDiagnosticsPayload = build_config_diagnostics_payload(config)
    return build_json_envelope(meta=meta, config_diagnostics=payload)

iter_config_diagnostics_ndjson_records

iter_config_diagnostics_ndjson_records(*, config, meta)

Iterate NDJSON records for config diagnostics.

Shapes

• counts-only record: {"kind": "config_diagnostics", "meta": ..., "config_diagnostics": {"diagnostic_counts": ...}}
• one diagnostic record per entry (domain="config")

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Immutable runtime configuration providing staged validation logs that are flattened for NDJSON output.	required
meta	MetaPayload	Machine-output metadata (tool/version).	required

Yields:

Type	Description
dict[str, object]	NDJSON record mappings (not yet serialized), in this order:
dict[str, object]	1) one config_diagnostics record containing counts only
dict[str, object]	2+) one diagnostic record per diagnostic entry (domain=MachineDomain.CONFIG).

Source code in src/topmark/config/machine/envelopes.py

def iter_config_diagnostics_ndjson_records(
    *,
    config: FrozenConfig,
    meta: MetaPayload,
) -> Iterator[dict[str, object]]:
    """Iterate NDJSON records for config diagnostics.

    Shapes:
      - counts-only record:
            {"kind": "config_diagnostics", "meta": ...,
            "config_diagnostics": {"diagnostic_counts": ...}}
      - one `diagnostic` record per entry (domain="config")

    Args:
        config: Immutable runtime configuration providing staged validation
            logs that are flattened for NDJSON output.
        meta: Machine-output metadata (tool/version).

    Yields:
        NDJSON record mappings (not yet serialized), in this order:
        1) one `config_diagnostics` record containing *counts only*
        2+) one `diagnostic` record per diagnostic entry (domain=`MachineDomain.CONFIG`).
    """
    payload: ConfigDiagnosticsPayload = build_config_diagnostics_payload(config)

    # NDJSON counts-only + streamed diagnostics
    counts: MachineDiagnosticCounts = payload.diagnostic_counts

    yield build_ndjson_record(
        kind=ConfigKind.CONFIG_DIAGNOSTICS,
        meta=meta,
        payload={
            DiagnosticKey.DIAGNOSTIC_COUNTS.value: counts.to_dict(),
        },
    )
    # One diagnostic per line
    flattened_diagnostics: FrozenDiagnosticLog = config.validation_logs.flattened()
    yield from iter_diagnostic_ndjson_records(
        meta=meta,
        domain=MachineDomain.CONFIG,
        diagnostics=flattened_diagnostics,
    )

build_config_check_json_envelope

build_config_check_json_envelope(
    *, config, resolved_toml, meta, strict, ok
)

Build the JSON envelope for topmark config check.

Shape

{"meta": ..., "config": ..., "config_diagnostics": ..., "config_check": ...}

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Immutable runtime configuration.	required
resolved_toml	ResolvedTopmarkTomlSources	Resolved TOML sources used to build the optional layered provenance export.	required
meta	MetaPayload	Machine-output metadata (tool/version).	required
strict	bool	Whether warnings are treated as failures.	required
ok	bool	Whether the config passed validation.	required

Returns:

Type	Description
dict[str, object]	A JSON-envelope mapping (not yet serialized).

Source code in src/topmark/config/machine/envelopes.py

def build_config_check_json_envelope(
    *,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    meta: MetaPayload,
    strict: bool,
    ok: bool,
) -> dict[str, object]:
    """Build the JSON envelope for `topmark config check`.

    Shape:
        {"meta": ..., "config": ..., "config_diagnostics": ..., "config_check": ...}

    Args:
        config: Immutable runtime configuration.
        resolved_toml: Resolved TOML sources used to build the optional layered
            provenance export.
        meta: Machine-output metadata (tool/version).
        strict: Whether warnings are treated as failures.
        ok: Whether the config passed validation.

    Returns:
        A JSON-envelope mapping (not yet serialized).
    """
    cfg_diag_payload: ConfigDiagnosticsPayload = build_config_diagnostics_payload(config)
    cfg_payload: ConfigPayload = build_config_payload(
        config,
        resolved_toml=resolved_toml,
    )

    config_check_summary: ConfigCheckSummary = build_config_check_summary_payload(
        config=config,
        cfg_diag_payload=cfg_diag_payload,
        strict=strict,
        ok=ok,
    )

    envelope: dict[str, object] = build_json_envelope(
        meta=meta,
        config=cfg_payload,
        config_diagnostics=cfg_diag_payload,
        config_check=config_check_summary,
    )
    return envelope

iter_config_prefix_ndjson_records

iter_config_prefix_ndjson_records(
    *,
    config,
    resolved_toml,
    meta,
    cfg_payload=None,
    cfg_diag_payload=None,
)

Yield the standard NDJSON prefix records for config-aware machine streams.

Prefix

1) config record with the effective config snapshot 2) config_diagnostics record containing counts only

Callers that want per-diagnostic records should additionally emit iter_diagnostic_ndjson_records(...).

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Effective configuration instance.	required
resolved_toml	ResolvedTopmarkTomlSources	Resolved TOML sources used to build the optional layered provenance export.	required
meta	MetaPayload	Shared metadata payload.	required
cfg_payload	ConfigPayload | None	Optional precomputed ConfigPayload.	None
cfg_diag_payload	ConfigDiagnosticsPayload | None	Optional precomputed ConfigDiagnosticsPayload.	None

Yields:

Type	Description
dict[str, object]	NDJSON records for config and counts-only config_diagnostics.

Source code in src/topmark/config/machine/envelopes.py

def iter_config_prefix_ndjson_records(
    *,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    meta: MetaPayload,
    cfg_payload: ConfigPayload | None = None,
    cfg_diag_payload: ConfigDiagnosticsPayload | None = None,
) -> Iterator[dict[str, object]]:
    """Yield the standard NDJSON prefix records for config-aware machine streams.

    Prefix:
      1) `config` record with the effective config snapshot
      2) `config_diagnostics` record containing *counts only*

    Callers that want per-diagnostic records should additionally emit
    `iter_diagnostic_ndjson_records(...)`.

    Args:
        config: Effective configuration instance.
        resolved_toml: Resolved TOML sources used to build the optional layered
            provenance export.
        meta: Shared metadata payload.
        cfg_payload: Optional precomputed `ConfigPayload`.
        cfg_diag_payload: Optional precomputed `ConfigDiagnosticsPayload`.

    Yields:
        NDJSON records for `config` and counts-only `config_diagnostics`.
    """
    payload: ConfigPayload = (
        cfg_payload
        if cfg_payload is not None
        else build_config_payload(
            config,
            resolved_toml=resolved_toml,
        )
    )
    diag_payload: ConfigDiagnosticsPayload = (
        cfg_diag_payload
        if cfg_diag_payload is not None
        else build_config_diagnostics_payload(config)
    )
    counts_only: MachineDiagnosticCounts = diag_payload.diagnostic_counts

    yield build_ndjson_record(
        kind=ConfigKind.CONFIG,
        meta=meta,
        payload=payload,
    )

    yield build_ndjson_record(
        kind=ConfigKind.CONFIG_DIAGNOSTICS,
        meta=meta,
        payload={
            DiagnosticKey.DIAGNOSTIC_COUNTS.value: counts_only.to_dict(),
        },
    )

iter_config_check_ndjson_records

iter_config_check_ndjson_records(
    *, config, resolved_toml, meta, strict, ok
)

Iterate NDJSON records for topmark config check.

Record sequence

1) config 2) config_diagnostics (counts-only) 3) config_check 4+) diagnostic (domain="config") one per flattened compatibility diagnostic

Parameters:

Name	Type	Description	Default
config	FrozenConfig	Immutable runtime configuration.	required
resolved_toml	ResolvedTopmarkTomlSources	Resolved TOML sources used to build the optional layered provenance export.	required
meta	MetaPayload	Machine-output metadata (tool/version).	required
strict	bool	Whether warnings are treated as failures.	required
ok	bool	Whether the config passed validation.	required

Yields:

Type	Description
dict[str, object]	NDJSON record mappings (not yet serialized), in this order:
dict[str, object]	1) config
dict[str, object]	2) config_diagnostics (counts only)
dict[str, object]	3) config_check
dict[str, object]	4+) diagnostic records (domain=MachineDomain.CONFIG), one per diagnostic entry.

Source code in src/topmark/config/machine/envelopes.py

def iter_config_check_ndjson_records(
    *,
    config: FrozenConfig,
    resolved_toml: ResolvedTopmarkTomlSources,
    meta: MetaPayload,
    strict: bool,
    ok: bool,
) -> Iterator[dict[str, object]]:
    """Iterate NDJSON records for `topmark config check`.

    Record sequence:
        1) config
        2) config_diagnostics (counts-only)
        3) config_check
        4+) diagnostic (domain="config") one per flattened compatibility
            diagnostic

    Args:
        config: Immutable runtime configuration.
        resolved_toml: Resolved TOML sources used to build the optional layered
            provenance export.
        meta: Machine-output metadata (tool/version).
        strict: Whether warnings are treated as failures.
        ok: Whether the config passed validation.

    Yields:
        NDJSON record mappings (not yet serialized), in this order:
        1) `config`
        2) `config_diagnostics` (counts only)
        3) `config_check`
        4+) `diagnostic` records (domain=`MachineDomain.CONFIG`), one per diagnostic entry.
    """
    cfg_diag_payload: ConfigDiagnosticsPayload = build_config_diagnostics_payload(config)
    config_check_summary: ConfigCheckSummary = build_config_check_summary_payload(
        config=config,
        cfg_diag_payload=cfg_diag_payload,
        strict=strict,
        ok=ok,
    )

    cfg_payload: ConfigPayload = build_config_payload(
        config,
        resolved_toml=resolved_toml,
    )
    yield from iter_config_prefix_ndjson_records(
        config=config,
        resolved_toml=resolved_toml,
        cfg_payload=cfg_payload,
        meta=meta,
        cfg_diag_payload=cfg_diag_payload,
    )

    yield build_ndjson_record(
        kind=ConfigKind.CONFIG_CHECK,
        meta=meta,
        container_key=ConfigKey.CONFIG_CHECK,
        payload=config_check_summary,
    )

    # One diagnostic per line
    flattened_diagnostics: FrozenDiagnosticLog = config.validation_logs.flattened()
    yield from iter_diagnostic_ndjson_records(
        meta=meta,
        domain=MachineDomain.CONFIG,
        diagnostics=flattened_diagnostics,
    )