Derivation infrastructure API

The infrastructure separates three concerns:

• schema authoring creates immutable, codec-visible records and closed ADTs;
• pass authoring extracts static contracts from annotations without wrapping execution;
• the transaction runner verifies inputs, outputs, lineage rules, deterministic replay, analyses, and checkpoints before commit.

Schema authoring

Expert authoring surface for canonical records and closed algebraic data.

Application users do not need these helpers. They are intentionally grouped here for core schema and trusted dialect authors; registry and manifest implementation details remain in :mod:blueprinting.schema.deriving.

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

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

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

adt_manifest

adt_manifest(family: type[Any]) -> tuple[VariantSpec, ...]

Return a deterministic documentation/schema manifest for one ADT family.

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

def adt_manifest(family: type[Any]) -> tuple[VariantSpec, ...]:
    """Return a deterministic documentation/schema manifest for one ADT family."""

    spec = _ADT_SPECS.get(family)
    if spec is None:
        raise TypeError(f"{family.__name__} is not a declared ADT family")
    return tuple(
        sorted((item for item in _VARIANT_SPECS.values() if item.family == spec), key=lambda item: item.local_tag)
    )

Pass authoring

Stable authoring surface for verified derivation passes.

Application code should execute passes through :class:PassManager. This module is for trusted pass and target-extension authors and deliberately omits transaction-runner and registry implementation details.

derivation

derivation(name: str, *, revision: str, bindings: Iterable[BindingAxis] = (), requires: Iterable[AnalysisKey] = (), preserves: Iterable[AnalysisKey] = (), produces: Iterable[AnalysisKey] = (), rules: Iterable[PassRule] = (), mutation: MutationModel = MutationModel.IMMUTABLE, verification: VerificationPolicy = VerificationPolicy.BOTH, deterministic: bool = True, uses_session_seed: bool = False, normalizer: PassNormalizer | None = None) -> Callable[[PassType], PassType]

Attach a complete exact-schema contract to a typed pass class.

Source code in src/blueprinting/synthesizer/passes/deriving.py

def derivation(
    name: str,
    *,
    revision: str,
    bindings: Iterable[BindingAxis] = (),
    requires: Iterable[AnalysisKey] = (),
    preserves: Iterable[AnalysisKey] = (),
    produces: Iterable[AnalysisKey] = (),
    rules: Iterable[PassRule] = (),
    mutation: MutationModel = MutationModel.IMMUTABLE,
    verification: VerificationPolicy = VerificationPolicy.BOTH,
    deterministic: bool = True,
    uses_session_seed: bool = False,
    normalizer: PassNormalizer | None = None,
) -> Callable[[PassType], PassType]:
    """Attach a complete exact-schema contract to a typed pass class."""

    def decorate(pass_type: PassType) -> PassType:
        source, target = _pass_types(pass_type)
        pass_type.contract = PassContract.create(
            name,
            source,
            target,
            revision=revision,
            required_bindings=frozenset(bindings),
            required_analyses=frozenset(requires),
            preserved_analyses=frozenset(preserves),
            produced_analyses=frozenset(produces),
            mutation_model=mutation,
            verification=verification,
            deterministic=deterministic,
            uses_session_seed=uses_session_seed,
            rules=tuple(rules),
            normalizer=normalizer,
        )
        previous = _PASS_CONTRACTS.get(pass_type)
        if previous is not None and previous != pass_type.contract:
            raise RuntimeError(f"pass class {pass_type.__name__} is already registered with another contract")
        _PASS_CONTRACTS[pass_type] = pass_type.contract
        return pass_type

    return decorate

relation

relation(transform: str, rewrite: str, *, source: type[Any] | str, target: type[Any] | str, verifier: Callable[[Any, Any, RelationCheckContext], None], preserves: Iterable[RuleClaim] = (), introduces: Iterable[str] = (), forbids: Iterable[str] = ()) -> PassRule

Declare a lineage shape plus an independent executable relation invariant.

Source code in src/blueprinting/synthesizer/passes/deriving.py

def relation(
    transform: str,
    rewrite: str,
    *,
    source: type[Any] | str,
    target: type[Any] | str,
    verifier: Callable[[Any, Any, RelationCheckContext], None],
    preserves: Iterable[RuleClaim] = (),
    introduces: Iterable[str] = (),
    forbids: Iterable[str] = (),
) -> PassRule:
    """Declare a lineage shape plus an independent executable relation invariant."""

    source_name = source if isinstance(source, str) else source.__name__
    target_name = target if isinstance(target, str) else target.__name__
    return PassRule(
        transform,
        source_name,
        target_name,
        rewrite,
        preserves=tuple(preserves),
        introduces=tuple(introduces),
        forbids=tuple(forbids),
        verifier=cast(RuleVerifier, verifier),
    )

claim

claim(name: str, verifier: Callable[[Any, Any, RelationCheckContext], None]) -> RuleClaim

Declare one preservation property and the predicate that proves it.

Source code in src/blueprinting/synthesizer/passes/deriving.py

def claim(name: str, verifier: Callable[[Any, Any, RelationCheckContext], None]) -> RuleClaim:
    """Declare one preservation property and the predicate that proves it."""

    return RuleClaim(name, cast(RuleVerifier, verifier))

Transaction and verification types

Declarative, immutable derivation-pass infrastructure.

The pass manager treats lowering as a sequence of typed snapshot transitions. It validates schemas, bindings, analyses, mutation behavior, verification policy, and provenance around every pass. Analysis publication is committed only after the produced IR has passed its contract, avoiding half-written caches after a failed lowering.

PassRule dataclass

PassRule(transform: str, source_entity: str, target_entity: str, rewrite: str, preserves: tuple[RuleClaim, ...] = (), introduces: tuple[str, ...] = (), forbids: tuple[str, ...] = (), verifier: RuleVerifier | None = None)

Declarative semantic contract for one named lineage transform.

TransitionRelation dataclass

TransitionRelation(rule_id: str, transform: str, source_entity: str, target_entity: str, source_ids: tuple[str, ...], target_id: str, lineage_kind: LineageKind, evidence: tuple[ClaimEvidence, ...] = ())

TransitionReport dataclass

TransitionReport(source_digest: str, target_digest: str, relations: tuple[TransitionRelation, ...] = (), status: TransitionVerificationStatus = TransitionVerificationStatus.STRUCTURAL_ONLY, canonical_conformance: ClaimEvidence | None = None)

TransitionVerifier

Verify typed entity lineage and executable derivation laws.

PassContract dataclass

PassContract(name: str, revision: str, input_type: type[CanonicalIRMixin], input_schema: SchemaRange, output_type: type[CanonicalIRMixin], output_schema: SchemaVersion, required_bindings: frozenset[BindingAxis] = frozenset(), required_analyses: frozenset[AnalysisKey] = frozenset(), preserved_analyses: frozenset[AnalysisKey] = frozenset(), produced_analyses: frozenset[AnalysisKey] = frozenset(), mutation_model: MutationModel = MutationModel.IMMUTABLE, verification: VerificationPolicy = VerificationPolicy.BOTH, deterministic: bool = True, uses_session_seed: bool = False, rules: tuple[PassRule, ...] = (), normalizer: PassNormalizer | None = None)

Complete static contract for one canonical IR transition.

normalizer_identity property

normalizer_identity: str | None

Stable diagnostic identity of the canonical derivation law.

digest property

digest: str

Content identity of the contract and its declared callable identities.

create classmethod

create(name: str, input_type: type[CanonicalIRMixin], output_type: type[CanonicalIRMixin], *, revision: str = '1', **options: Any) -> PassContract

Build the common exact-schema contract without hiding its resolved values.

Source code in src/blueprinting/synthesizer/passes/base.py

@classmethod
def create(
    cls,
    name: str,
    input_type: type[CanonicalIRMixin],
    output_type: type[CanonicalIRMixin],
    *,
    revision: str = "1",
    **options: Any,
) -> PassContract:
    """Build the common exact-schema contract without hiding its resolved values."""

    return cls(
        name=name,
        revision=revision,
        input_type=input_type,
        input_schema=SchemaRange.exact(input_type.SCHEMA_VERSION),
        output_type=output_type,
        output_schema=output_type.SCHEMA_VERSION,
        **options,
    )

PassContext dataclass

PassContext(session: SynthesisSession, analyses: AnalysisStore)

PassResult dataclass

PassResult(ir: OutputIR, analyses: tuple[AnalysisProduct, ...] = ())

Bases: Generic[OutputIR]

DerivationPass

Bases: ABC, Generic[InputIR, OutputIR]

Base class for one declaratively contracted lowering or analysis pass.

PassPipeline dataclass

PassPipeline(passes: tuple[DerivationPass[Any, Any], ...] = ())

Immutable, type-checked pass composition.

PassRecord dataclass

PassRecord(pass_name: str, contract_revision: str, contract_digest: str, input_digest: str, output_digest: str, session_fingerprint: str, duration_ns: int, mutation_model: MutationModel, produced_analyses: tuple[AnalysisKey, ...], transition_report: TransitionReport)

PassCheckpoint dataclass

PassCheckpoint(record: PassRecord, ir: CanonicalIRMixin, analysis_products: tuple[AnalysisProduct, ...])

One inspectable lowering boundary for profilers and validation hooks.

PassManager

PassManager(analyses: AnalysisStore | None = None, *, observers: Iterable[PassObserver] = (), determinism: DeterminismPolicy = DeterminismPolicy.OFF)

Executes pipelines while enforcing all pass contracts at the boundary.

Source code in src/blueprinting/synthesizer/passes/base.py

def __init__(
    self,
    analyses: AnalysisStore | None = None,
    *,
    observers: Iterable[PassObserver] = (),
    determinism: DeterminismPolicy = DeterminismPolicy.OFF,
) -> None:
    self.analyses = analyses if analyses is not None else AnalysisStore()
    self.observers = tuple(observers)
    if not isinstance(determinism, DeterminismPolicy):
        raise TypeError("determinism must be a DeterminismPolicy")
    self.determinism = determinism

run

run(pipeline: PassPipeline, ir: InputIR, *, session: SynthesisSession) -> Checked[PipelineResult[Any]]

Execute a pipeline and return expected contract failures as diagnostics.

Source code in src/blueprinting/synthesizer/passes/base.py

def run(
    self,
    pipeline: PassPipeline,
    ir: InputIR,
    *,
    session: SynthesisSession,
) -> Checked[PipelineResult[Any]]:
    """Execute a pipeline and return expected contract failures as diagnostics."""

    try:
        return Ok(self._run(pipeline, ir, session=session))
    except PassExecutionError:
        raise
    except SynthesisError as error:
        return Err(
            DiagnosticSet.of(
                Diagnostic(
                    _diagnostic_code(error),
                    str(error),
                    ("pipeline",),
                )
            )
        )

require_run

require_run(pipeline: PassPipeline, ir: InputIR, *, session: SynthesisSession) -> PipelineResult[Any]

Explicit exception adapter for application and legacy boundaries.

Source code in src/blueprinting/synthesizer/passes/base.py

def require_run(
    self,
    pipeline: PassPipeline,
    ir: InputIR,
    *,
    session: SynthesisSession,
) -> PipelineResult[Any]:
    """Explicit exception adapter for application and legacy boundaries."""

    return self._run(pipeline, ir, session=session)