MachineIR API

MachineIR belongs to one target plugin and records its instruction dialect, sections, entry points, ABI identity, command lineage, and program format.

Target-specific machine program contract.

NonEmptyText module-attribute

NonEmptyText: TypeAlias = Annotated[str, NON_EMPTY]

PositiveInt module-attribute

PositiveInt: TypeAlias = Annotated[int, POSITIVE]

_MACHINE_RESERVED module-attribute

_MACHINE_RESERVED = frozenset({'predicted_start', 'predicted_end', 'predicted_duration', 'estimated_time', 'latency_estimate'})

__all__ module-attribute

__all__ = ['MachineEntryPoint', 'MachineInstruction', 'MachineIR', 'MachineOpcode', 'MachineSection', 'MachineSectionKind']

FrozenDict

FrozenDict(source: Mapping[str, V] | None = None, *, items: Iterable[tuple[str, V]] | None = None)

Bases: Mapping[str, V], Generic[V]

A compact, hashable mapping with recursively frozen values.

Source code in src/blueprinting/schema/frozen.py

def __init__(
    self,
    source: Mapping[str, V] | None = None,
    *,
    items: Iterable[tuple[str, V]] | None = None,
) -> None:
    if source is not None and items is not None:
        raise TypeError("provide either source or items, not both")
    raw_items = source.items() if source is not None else (items or ())
    copied: dict[str, V] = {}
    for key, value in raw_items:
        if not isinstance(key, str):
            raise TypeError("FrozenDict keys must be strings")
        if key in copied:
            raise ValueError(f"duplicate FrozenDict key: {key!r}")
        copied[key] = cast(V, freeze(value))
    self._items: tuple[tuple[str, V], ...] = tuple(sorted(copied.items(), key=lambda pair: pair[0]))
    self._hash: int | None = None

__reduce__

__reduce__() -> tuple[type[FrozenDict[V]], tuple[dict[str, V]]]

Use the public constructor for process and UI cache round-trips.

Source code in src/blueprinting/schema/frozen.py

def __reduce__(self) -> tuple[type[FrozenDict[V]], tuple[dict[str, V]]]:
    """Use the public constructor for process and UI cache round-trips."""

    return FrozenDict, (dict(self._items),)

DiagnosticBag

DiagnosticBag()

Bases: DiagnosticBag

Compatibility builder that publishes VerificationReport.

Source code in src/blueprinting/schema/diagnostics.py

def __init__(self) -> None:
    self._items: list[Diagnostic] = []

VerificationReport dataclass

VerificationReport(diagnostics: tuple[Diagnostic, ...] = ())

Bases: DiagnosticSet

Backward-compatible name for the domain-free immutable diagnostics.

CommandId dataclass

CommandId(value: str)

Bases: StableId

InstructionId dataclass

InstructionId(value: str)

Bases: StableId

Lineage

Typed provenance from source entities to one lowering product.

CanonicalIRMixin

Behavior shared by immutable canonical IR roots.

to_json

to_json() -> str

Serialize through a self-describing, digest-checked envelope.

Source code in src/blueprinting/synthesizer/stages/common.py

def to_json(self) -> str:
    """Serialize through a self-describing, digest-checked envelope."""

    self.require_valid()
    snapshot = IRSnapshot(
        schema_name=self.header.schema_name,
        schema_version=self.header.schema_version,
        producer_version=self.header.producer_version,
        feature_set=self.header.feature_set,
        content_digest=self.digest,
        payload=self,
    )
    return canonical_dumps(snapshot)

from_json classmethod

from_json(payload: str) -> Checked[IR]

Decode an exact-schema snapshot without exception control flow.

Source code in src/blueprinting/synthesizer/stages/common.py

@classmethod
def from_json(cls: type[IR], payload: str) -> Checked[IR]:
    """Decode an exact-schema snapshot without exception control flow."""

    try:
        return Ok(cls._decode_json(payload))
    except IRVerificationError as error:
        return Err(DiagnosticSet(tuple(error.diagnostics)))
    except SerializationError as error:
        return Err(
            DiagnosticSet.of(
                Diagnostic("serialization.snapshot", str(error), ("snapshot",)),
            )
        )

require_from_json classmethod

require_from_json(payload: str) -> IR

Explicit exception adapter for trusted internal/replay boundaries.

Source code in src/blueprinting/synthesizer/stages/common.py

@classmethod
def require_from_json(cls: type[IR], payload: str) -> IR:
    """Explicit exception adapter for trusted internal/replay boundaries."""

    return cls.from_json(payload).or_raise(
        lambda diagnostics: SerializationError("; ".join(item.render() for item in diagnostics.errors))
    )

load_migrated classmethod

load_migrated(payload: str, *, registry: Any = None) -> Checked[IR]

Load through an explicit registered migration path.

Source code in src/blueprinting/synthesizer/stages/common.py

@classmethod
def load_migrated(cls: type[IR], payload: str, *, registry: Any = None) -> Checked[IR]:
    """Load through an explicit registered migration path."""

    if registry is None:
        from ..schema_migration import DEFAULT_SCHEMA_MIGRATIONS

        registry = DEFAULT_SCHEMA_MIGRATIONS
    try:
        result = registry.migrate_json(
            payload,
            schema_name=cls.SCHEMA_NAME,
            target_version=cls.SCHEMA_VERSION,
        )
    except SerializationError as error:
        return Err(DiagnosticSet.of(Diagnostic("serialization.migration", str(error), ("snapshot",))))
    return cls.from_json(result.payload)

verify

verify() -> Checked[IR]

Return the immutable snapshot or all expected verifier failures.

Source code in src/blueprinting/synthesizer/stages/common.py

def verify(self: IR) -> Checked[IR]:
    """Return the immutable snapshot or all expected verifier failures."""

    return checked(self, self.diagnostics())

IRHeader

Version and provenance header embedded in every canonical IR.

SchemaVersion

Semantic version of one serialized IR schema.

MachineOpcode

Dialect-qualified opcode owned by one target plugin.

MachineInstruction

Target instruction with explicit dependencies, operands, and lineage.

MachineSectionKind

Bases: Enum

Container role of a machine-program section.

MachineSection

Aligned code or data section in a target machine program.

MachineEntryPoint

Named externally addressable instruction in a machine program.

MachineIR

Bases: CanonicalIRMixin

Target-owned instruction dialect before final binary/container emission.

enum

enum(tag: str) -> Callable[[type[T]], type[T]]

Register one closed enumeration through the public authoring surface.

Source code in src/blueprinting/schema/deriving.py

def enum(tag: str) -> Callable[[type[T]], type[T]]:
    """Register one closed enumeration through the public authoring surface."""

    if not isinstance(tag, str) or _WIRE_RE.fullmatch(tag) is None:
        raise TypeError("canonical enum tag must be a dotted lowercase wire identity")
    return enum_type(tag)

record

record(tag: str, *, order: bool = False) -> Callable[[type[T]], type[T]]

Derive an immutable canonical record from an annotated class declaration.

Source code in src/blueprinting/schema/deriving.py

@dataclass_transform(frozen_default=True)
def record(
    tag: str,
    *,
    order: bool = False,
) -> Callable[[type[T]], type[T]]:
    """Derive an immutable canonical record from an annotated class declaration."""

    if not isinstance(tag, str) or _WIRE_RE.fullmatch(tag) is None:
        raise TypeError("canonical record tag must be a dotted lowercase wire identity")

    def decorate(cls: type[T]) -> type[T]:
        if "__dataclass_fields__" in cls.__dict__:
            raise TypeError("@record derives its own frozen dataclass; do not combine it with @dataclass")
        _install_structural_post_init(cls)
        cls = dataclass(frozen=True, slots=True, order=order)(cls)
        return record_type(tag)(cls)

    return decorate

is_content_digest

is_content_digest(value: str) -> bool

Source code in src/blueprinting/synthesizer/stages/common.py

def is_content_digest(value: str) -> bool:
    return isinstance(value, str) and _DIGEST_RE.fullmatch(value) is not None

make_header

make_header(schema_name: str, schema_version: SchemaVersion, *, parent_digests: Iterable[str] = (), features: Iterable[str] = (), producer_version: str = '0.0.0') -> IRHeader

Source code in src/blueprinting/synthesizer/stages/common.py

def make_header(
    schema_name: str,
    schema_version: SchemaVersion,
    *,
    parent_digests: Iterable[str] = (),
    features: Iterable[str] = (),
    producer_version: str = "0.0.0",
) -> IRHeader:
    return IRHeader(
        schema_name=schema_name,
        schema_version=schema_version,
        producer_version=producer_version,
        feature_set=REQUIRED_IR_FEATURES | frozenset(features),
        parent_digests=tuple(parent_digests),
    )

reject_reserved_attributes

reject_reserved_attributes(bag: DiagnosticBag, attributes: FrozenDict, reserved: frozenset[str], *path: str) -> None

Reject semantic fields smuggled through extension dictionaries.

Source code in src/blueprinting/synthesizer/stages/common.py

def reject_reserved_attributes(
    bag: DiagnosticBag,
    attributes: FrozenDict,
    reserved: frozenset[str],
    *path: str,
) -> None:
    """Reject semantic fields smuggled through extension dictionaries."""

    def walk(value: Any, location: tuple[str, ...]) -> None:
        if isinstance(value, FrozenDict):
            for key, item in value.items():
                if key.lower() in reserved:
                    bag.error(
                        "attribute.reserved",
                        f"{key!r} is a semantic field and is illegal in this IR layer",
                        *(location + (key,)),
                    )
                walk(item, location + (key,))
        elif isinstance(value, (tuple, frozenset)):
            values = value if isinstance(value, tuple) else tuple(sorted(value, key=repr))
            for index, item in enumerate(values):
                walk(item, location + (str(index),))

    walk(attributes, tuple(path))

verify_ordered_dag

verify_ordered_dag(bag: DiagnosticBag, entities: Sequence[Entity], id_of: Callable[[Entity], StableId], dependencies_of: Callable[[Entity], Iterable[StableId]], path: str) -> None

Verify references and require canonical topological sequence order.

Source code in src/blueprinting/synthesizer/stages/common.py

def verify_ordered_dag(
    bag: DiagnosticBag,
    entities: Sequence[Entity],
    id_of: Callable[[Entity], StableId],
    dependencies_of: Callable[[Entity], Iterable[StableId]],
    path: str,
) -> None:
    """Verify references and require canonical topological sequence order."""

    identifiers = {id_of(entity) for entity in entities}
    seen = set()
    for index, entity in enumerate(entities):
        identifier = id_of(entity)
        dependencies = tuple(dependencies_of(entity))
        if len(set(dependencies)) != len(dependencies):
            bag.error("dag.duplicate_dependency", "dependencies must be unique", path, str(index), "dependencies")
        if identifier in dependencies:
            bag.error("dag.self_dependency", f"{identifier} depends on itself", path, str(index), "dependencies")
        for dependency in dependencies:
            if dependency not in identifiers:
                bag.error(
                    "reference.unknown",
                    f"dependency {dependency} is not declared",
                    path,
                    str(index),
                    "dependencies",
                )
            elif dependency not in seen:
                bag.error(
                    "dag.not_topological",
                    f"dependency {dependency} must precede {identifier}",
                    path,
                    str(index),
                    "dependencies",
                    hint="store canonical DAG nodes in topological order",
                )
        seen.add(identifier)

verify_unique_ids

verify_unique_ids(bag: DiagnosticBag, entities: Sequence[Entity], id_of: Callable[[Entity], StableId], path: str) -> None

Source code in src/blueprinting/synthesizer/stages/common.py

def verify_unique_ids(
    bag: DiagnosticBag,
    entities: Sequence[Entity],
    id_of: Callable[[Entity], StableId],
    path: str,
) -> None:
    seen = set()
    for index, entity in enumerate(entities):
        identifier = id_of(entity)
        if identifier in seen:
            bag.error("id.duplicate", f"duplicate identifier {identifier}", path, str(index), "id")
        seen.add(identifier)