推导基础设施 API¶

基础设施分离三个关注点：

schema authoring 创建 immutable、codec-visible 的 record 与封闭 ADT；
pass authoring 从 annotation 提取静态 contract，不包装执行语义；
transaction runner 在 commit 前验证输入、输出、lineage rule、deterministic replay、analysis 与 checkpoint。

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))

事务与验证类型¶

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)

推导基础设施 API¶

Schema authoring¶

record ¶

adt ¶

variant ¶

adt_manifest ¶

Pass authoring¶

derivation ¶

relation ¶

claim ¶

事务与验证类型¶

PassRule dataclass ¶

TransitionRelation dataclass ¶

TransitionReport dataclass ¶

TransitionVerifier ¶

PassContract dataclass ¶

normalizer_identity property ¶

digest property ¶

create classmethod ¶

PassContext dataclass ¶

PassResult dataclass ¶

DerivationPass ¶

PassPipeline dataclass ¶

PassRecord dataclass ¶

PassCheckpoint dataclass ¶

PassManager ¶

run ¶

require_run ¶

PassRule `dataclass` ¶

TransitionRelation `dataclass` ¶

TransitionReport `dataclass` ¶

PassContract `dataclass` ¶

normalizer_identity `property` ¶

digest `property` ¶

create `classmethod` ¶

PassContext `dataclass` ¶

PassResult `dataclass` ¶

PassPipeline `dataclass` ¶

PassRecord `dataclass` ¶

PassCheckpoint `dataclass` ¶