ConcretePlanIR API

ConcretePlanIR is the target-bound authoritative command envelope. Common coordination semantics stay in the envelope; target-only correctness semantics live in a registered typed extension.

Target- and deployment-bound authoritative command plan.

NonEmptyText module-attribute

NonEmptyText: TypeAlias = Annotated[str, NON_EMPTY]

NonNegativeInt module-attribute

NonNegativeInt: TypeAlias = Annotated[int, NON_NEGATIVE]

PositiveInt module-attribute

PositiveInt: TypeAlias = Annotated[int, POSITIVE]

SynchronizationTokens module-attribute

SynchronizationTokens: TypeAlias = Annotated[tuple[TokenId, ...], NON_EMPTY, UNIQUE_ITEMS]

CommandBodyVariant module-attribute

CommandBodyVariant: TypeAlias = Launch | CollectiveCommand | Transfer | Barrier | Signal | Wait | HostCall

CommandSynchronizationVariant module-attribute

CommandSynchronizationVariant: TypeAlias = Unsynchronized | WaitFor | SignalAfter | WaitAndSignal

_CONCRETE_RESERVED module-attribute

_CONCRETE_RESERVED = frozenset({'start', 'start_time', 'predicted_start', 'end', 'end_time', 'predicted_end', 'duration', 'latency', 'estimated_time', 'predicted_duration'})

__all__ module-attribute

__all__ = ['AccessMode', 'Barrier', 'BufferBinding', 'BufferUse', 'CommandKind', 'CommandBody', 'CommandBodyVariant', 'CommandSynchronization', 'CommandSynchronizationVariant', 'CollectiveCommand', 'ConcreteCommand', 'ConcretePlanIR', 'DevicePlacement', 'ImplementationRef', 'HostCall', 'IssueSlot', 'MemoryRegion', 'Launch', 'QueueIssueOrder', 'QueueKind', 'QueueScheduleExtension', 'QueueSpec', 'RouteConstraint', 'SlotDataflowExtension', 'Signal', 'SignalAfter', 'TargetScheduleExtension', 'Transfer', 'Unsynchronized', 'Wait', 'WaitAndSignal', 'WaitFor']

ValueConstraint

Bases: Enum

Small closed vocabulary of reusable structural refinements.

VariantSpec dataclass

VariantSpec(family: ADTSpec, local_tag: str, wire_tag: str, constructor: type[Any])

Manifest entry for one explicitly named ADT constructor.

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.

BufferId dataclass

BufferId(value: str)

Bases: StableId

CommandId dataclass

CommandId(value: str)

Bases: StableId

DeviceId dataclass

DeviceId(value: str)

Bases: StableId

Lineage

Typed provenance from source entities to one lowering product.

MemoryRegionId dataclass

MemoryRegionId(value: str)

Bases: StableId

QueueId dataclass

QueueId(value: str)

Bases: StableId

TokenId dataclass

TokenId(value: str)

Bases: StableId

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.

DevicePlacement

Binding from a logical rank to a physical target device.

QueueKind

Bases: Enum

Target execution-engine category represented by a command queue.

QueueSpec

A target-owned command queue attached to one physical device.

MemoryRegion

A finite physical memory region available to concrete buffers.

BufferBinding

Physical placement of one portable buffer with preserved lineage.

ImplementationRef

Versioned target implementation and ABI selected for a command.

AccessMode

Bases: Enum

Concrete command access performed on a bound buffer.

BufferUse

Typed buffer access declared by one concrete command.

CommandKind

Bases: Enum

Derived command category used by generic consumers and diagnostics.

CommandBody

Closed family of mutually exclusive target command semantics.

Launch

Bases: CommandBody

CollectiveCommand

Bases: CommandBody

Transfer

Bases: CommandBody

Barrier

Bases: CommandBody

Signal

Bases: CommandBody

Wait

Bases: CommandBody

HostCall

Bases: CommandBody

CommandSynchronization

Closed synchronization clause orthogonal to executable command semantics.

Unsynchronized

Bases: CommandSynchronization

WaitFor

Bases: CommandSynchronization

SignalAfter

Bases: CommandSynchronization

WaitAndSignal

Bases: CommandSynchronization

ConcreteCommand

Graph envelope around one typed implementation-bound command body.

TargetScheduleExtension

Marker for registered target-owned scheduling correctness semantics.

QueueIssueOrder

Correctness-significant issue order for one target queue.

QueueScheduleExtension

Bases: TargetScheduleExtension

Typed target extension for queue-ordered execution semantics.

IssueSlot

Exact cycle and slot assigned to a command by a slot target.

RouteConstraint

Correctness-significant physical route required by a transfer command.

SlotDataflowExtension

Bases: TargetScheduleExtension

Typed target extension for slot issue and routed dataflow semantics.

ConcretePlanIR

Bases: CanonicalIRMixin

Dependency-driven plan consumed by both simulation and emission.

adt

adt(*, wire: str) -> Callable[[type[T]], type[T]]

Declare the shared semantic wire namespace for a closed sum type.

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

@dataclass_transform(frozen_default=True)
def adt(*, wire: str) -> Callable[[type[T]], type[T]]:
    """Declare the shared semantic wire namespace for a closed sum type."""

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

    def decorate(cls: type[T]) -> type[T]:
        if "__dataclass_fields__" in cls.__dict__:
            raise TypeError("@adt derives its own frozen dataclass; do not combine it with @dataclass")
        cls = dataclass(frozen=True, slots=True)(cls)
        if cls in _ADT_SPECS:
            raise RuntimeError(f"ADT family {cls.__name__} is already registered")
        if any(item.wire == wire for item in _ADT_SPECS.values()):
            raise RuntimeError(f"ADT wire namespace {wire!r} is already registered")
        spec = ADTSpec(wire, cls)
        _ADT_SPECS[cls] = spec
        setattr(cls, "__adt_spec__", spec)  # noqa: B010

        root = cls

        def adt_new(constructor: type[Any], *_args: Any, **_kwargs: Any) -> Any:
            if constructor is root:
                raise TypeError(f"ADT family {root.__name__} is abstract; instantiate one of its sealed variants")
            return object.__new__(constructor)

        setattr(cls, "__new__", staticmethod(adt_new))  # noqa: B010
        return cls

    return decorate

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

seal_adt

seal_adt(family: type[Any], variants: Any) -> Any

Declare the explicit runtime closure corresponding to a static Union alias.

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

def seal_adt(family: type[Any], variants: Any) -> Any:
    """Declare the explicit runtime closure corresponding to a static Union alias."""

    if family not in _ADT_SPECS:
        raise TypeError(f"{getattr(family, '__name__', family)!r} is not a declared ADT family")
    members = _union_members(variants)
    if any(_VARIANT_SPECS.get(item) is None for item in members):
        raise TypeError("an ADT closure may contain only registered variants")
    if any(_VARIANT_SPECS[item].family.root is not family for item in members):
        raise TypeError("an ADT closure cannot mix constructors from different families")
    if len(set(members)) != len(members):
        raise ValueError("an ADT closure cannot repeat a constructor")
    registered = tuple(item.constructor for item in adt_manifest(family))
    if set(members) != set(registered):
        missing = tuple(item.__name__ for item in registered if item not in members)
        unknown = tuple(item.__name__ for item in members if item not in registered)
        detail = []
        if missing:
            detail.append(f"missing {', '.join(missing)}")
        if unknown:
            detail.append(f"unknown {', '.join(unknown)}")
        raise ValueError(f"ADT closure must contain exactly its registered variants: {'; '.join(detail)}")
    previous = _ADT_CLOSURES.get(family)
    if previous is not None and previous != registered:
        raise RuntimeError(f"ADT family {family.__name__} is already sealed with another closure")
    _ADT_CLOSURES[family] = registered
    return variants

variant

variant(local_tag: str) -> Callable[[type[T]], type[T]]

Derive and register one explicitly named constructor of an ADT family.

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

@dataclass_transform(frozen_default=True)
def variant(local_tag: str) -> Callable[[type[T]], type[T]]:
    """Derive and register one explicitly named constructor of an ADT family."""

    if not isinstance(local_tag, str) or _LOCAL_TAG_RE.fullmatch(local_tag) is None:
        raise TypeError("variant tag must be a short lowercase kebab-case identity")

    def decorate(cls: type[T]) -> type[T]:
        family = _family_of(cls)
        if family.root in _ADT_CLOSURES:
            raise RuntimeError(f"ADT family {family.root.__name__} is sealed and cannot accept late variants")
        if any(item.family == family and item.local_tag == local_tag for item in _VARIANT_SPECS.values()):
            raise RuntimeError(f"ADT family {family.root.__name__} already declares variant {local_tag!r}")
        constructor = record(family.tag(local_tag))(cls)
        spec = VariantSpec(family, local_tag, family.tag(local_tag), constructor)
        _VARIANT_SPECS[constructor] = spec
        setattr(constructor, "__variant_spec__", spec)  # noqa: B010
        return constructor

    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_known_references

verify_known_references(bag: DiagnosticBag, references: Iterable[StableId], known: Iterable[StableId], *path: str) -> None

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

def verify_known_references(
    bag: DiagnosticBag,
    references: Iterable[StableId],
    known: Iterable[StableId],
    *path: str,
) -> None:
    known_set = set(known)
    for reference in references:
        if reference not in known_set:
            bag.error("reference.unknown", f"unknown reference {reference}", *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)

_verify_target_extension

_verify_target_extension(plan: ConcretePlanIR, bag: DiagnosticBag) -> None

Source code in src/blueprinting/synthesizer/stages/concrete_plan/ir.py

def _verify_target_extension(plan: ConcretePlanIR, bag: DiagnosticBag) -> None:
    command_by_id = {item.id: item for item in plan.commands}
    if isinstance(plan.target_extension, QueueScheduleExtension):
        queue_ids = {item.id for item in plan.queues}
        ordered_commands = tuple(command for order in plan.target_extension.orders for command in order.commands)
        expected = tuple(item.id for item in plan.commands if item.queue is not None)
        if len(ordered_commands) != len(set(ordered_commands)):
            bag.error("target.queue.duplicate_command", "queued commands must appear exactly once", "target_extension")
        if set(ordered_commands) != set(expected):
            bag.error("target.queue.coverage", "queue orders must cover every queued command", "target_extension")
        for order_index, order in enumerate(plan.target_extension.orders):
            if order.queue not in queue_ids:
                bag.error(
                    "reference.unknown", f"unknown extension queue {order.queue}", "target_extension", str(order_index)
                )
                continue
            position = {command: index for index, command in enumerate(order.commands)}
            for command_id in order.commands:
                command = command_by_id.get(command_id)
                if command is None:
                    bag.error("reference.unknown", f"unknown extension command {command_id}", "target_extension")
                    continue
                if command.queue != order.queue:
                    bag.error("target.queue.mismatch", "command is listed under a different queue", "target_extension")
                for dependency in command.dependencies:
                    if dependency in position and position[dependency] >= position[command_id]:
                        bag.error("target.queue.order", "queue order violates a command dependency", "target_extension")
        return

    if isinstance(plan.target_extension, SlotDataflowExtension):
        if plan.queues or any(item.queue is not None for item in plan.commands):
            bag.error("target.slot.queue", "slot/dataflow plans cannot carry queue semantics", "target_extension")
        slot_by_command = {item.command: item for item in plan.target_extension.issue_slots}
        if set(slot_by_command) != set(command_by_id):
            bag.error("target.slot.coverage", "issue slots must cover every command exactly once", "target_extension")
        for command in plan.commands:
            current = slot_by_command.get(command.id)
            if current is None:
                continue
            for dependency in command.dependencies:
                previous = slot_by_command.get(dependency)
                if previous is not None and (previous.cycle, previous.slot) >= (current.cycle, current.slot):
                    bag.error("target.slot.order", "issue slots violate a command dependency", "target_extension")
        device_ids = {item.id for item in plan.devices}
        routes = {item.command: item for item in plan.target_extension.routes}
        transfers = {item.id for item in plan.commands if item.kind is CommandKind.TRANSFER}
        if set(routes) != transfers:
            bag.error("target.route.coverage", "every transfer command requires exactly one route", "target_extension")
        for route in plan.target_extension.routes:
            if route.source_device not in device_ids or route.destination_device not in device_ids:
                bag.error("reference.unknown", "route endpoint is not a concrete device", "target_extension")
        return

    bag.error("target.extension.unknown", "unsupported typed target schedule extension", "target_extension")