refactor(checkers): let declarations own provider runtimes (#1299)

* refactor(checkers): let declarations own provider runtimes

Compose declaration-owned clean-process runtimes with the latest batched checker-identity path. Existing checker families retain the legacy registry; new declarations may carry one unassigned provider runtime, and the composition root batches identity material across the full declaration set before authorization.

* fix(checkers): defer declaration runtime measurement

Store passive provider-runtime factories on exact replay declarations and realize each runtime once, at installation. Importing domain declarations no longer identifies or hashes checker source, while the existing installer and authorization boundary keep the same resolved runtime contract.

* refactor(checkers): keep direct runtime migration compatibility

Retain the existing direct provider_runtime constructor as a temporary compatibility seam while making provider_runtime_factory lazy and cached. Reject dual ownership and preserve pre-authorization invariants for both forms.

* fix(checkers): type the lazy runtime accessor

Declare the compatibility accessor's dynamic return as Any so mypy accepts the dataclass field interception while the public provider_runtime field retains its precise declared type.

* test: include check-all in the primary help contract

* fix declaration-owned checker runtime compatibility

* fix(checkers): satisfy static validation

* fix(checkers): handle declaration-owned runtime installation

* fix(checkers): require explicit optional provider omission

* fix(checkers): omit unavailable optional FLINT replays

Author: morluto

src/jacobian/checker_operations.py

@@ -3,12 +3,15 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import Any
 
 from jacobian.contracts.capabilities import CapabilityProviderRuntime
 from jacobian.contracts.checkers import EvidenceKind
 from jacobian.contracts.results import ContractModel
 
+ProviderRuntimeFactory = Callable[[], CapabilityProviderRuntime]
+
 # Producer operation verb segments stripped when deriving a verifier capability
 # ID. Each producer capability ID contains exactly one of these segments; the
 # derived verifier ID removes it and appends ``.verify``.
@@ -107,11 +110,21 @@ class ExactReplayCheckerDeclaration:
     verification_title: str | None = None
     verification_description: str | None = None
     verification_tags: tuple[str, ...] = ()
-    provider_runtime_factory: Callable[..., CapabilityProviderRuntime] | None = None
+    provider_runtime: CapabilityProviderRuntime | None = field(
+        default=None,
+        repr=False,
+        compare=False,
+    )
+    provider_runtime_factory: ProviderRuntimeFactory | None = field(
+        default=None,
+        repr=False,
+        compare=False,
+    )
+    optional: bool = False
     supports_input: Callable[[object], bool] | None = None
 
     def __post_init__(self) -> None:
-        for field, value in {
+        for field_name, value in {
             "capability_id": self.capability_id,
             "function": self.function,
             "format_id": self.format_id,
@@ -121,11 +134,18 @@ def __post_init__(self) -> None:
         }.items():
             if not value.strip():
                 raise ValueError(
-                    f"exact replay checker declaration {field} must not be empty"
+                    f"exact replay checker declaration {field_name} must not be empty"
                 )
-        if self.provider_runtime_factory is None:
+        runtime = object.__getattribute__(self, "provider_runtime")
+        factory = object.__getattribute__(self, "provider_runtime_factory")
+        if runtime is not None and factory is not None:
+            raise ValueError(
+                "declaration must provide either provider_runtime or "
+                "provider_runtime_factory, not both"
+            )
+        if runtime is not None and runtime.checker_ids:
             raise ValueError(
-                "exact replay checker declaration requires a provider runtime factory"
+                "declaration-owned provider runtime must not pre-authorize checker IDs"
             )
         derived_id = derive_verification_capability_id(self.capability_id)
         explicit_text = (
@@ -173,6 +193,28 @@ def __post_init__(self) -> None:
                 derive_verification_tags(self.capability_id),
             )
 
+    def __getattribute__(self, name: str) -> Any:
+        if name != "provider_runtime":
+            return object.__getattribute__(self, name)
+        runtime = object.__getattribute__(self, "provider_runtime")
+        if runtime is not None:
+            return runtime
+        factory = object.__getattribute__(self, "provider_runtime_factory")
+        if factory is None:
+            return None
+        realized = factory()
+        if not isinstance(realized, CapabilityProviderRuntime):
+            raise TypeError(
+                "declaration-owned provider runtime factory must return "
+                "CapabilityProviderRuntime"
+            )
+        if realized.checker_ids:
+            raise ValueError(
+                "declaration-owned provider runtime must not pre-authorize checker IDs"
+            )
+        object.__setattr__(self, "provider_runtime", realized)
+        return realized
+
 
 @dataclass(frozen=True, slots=True)
 class CheckerOperation:
@@ -199,9 +241,9 @@ def __post_init__(self) -> None:
             "format_version": self.format_version,
             "reason": self.reason,
         }
-        for field, value in required_text.items():
+        for field_name, value in required_text.items():
             if not value.strip():
-                raise ValueError(f"checker operation {field} must not be empty")
+                raise ValueError(f"checker operation {field_name} must not be empty")
         if not self.claim_schema_uris:
             raise ValueError("checker operation must declare a claim schema")
         if not self.semantics_uris:

src/jacobian/domains/matrix_lattice/checkers.py

@@ -66,48 +66,55 @@ def _hnf_runtime(*, checker_ids: tuple[str, ...] = ()) -> CapabilityProviderRunt
         "check_matrix_determinant",
         "matrix.determinant.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
     ExactReplayCheckerDeclaration(
         "matrix.rank.compute",
         MatrixRankRequest,
         "check_matrix_rank",
         "matrix.rank.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
     ExactReplayCheckerDeclaration(
         "matrix.multiply.compute",
         RationalMatrixProductRequest,
         "check_matrix_product",
         "matrix.product.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
     ExactReplayCheckerDeclaration(
         "matrix.normal_form.rref.compute",
         RationalMatrixRequest,
         "check_matrix_rref",
         "matrix.rref.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
     ExactReplayCheckerDeclaration(
         "matrix.nullspace.compute",
         RationalMatrixRequest,
         "check_matrix_nullspace",
         "matrix.nullspace.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
     ExactReplayCheckerDeclaration(
         "matrix.characteristic_polynomial.compute",
         SquareRationalMatrixRequest,
         "check_matrix_characteristic_polynomial",
         "matrix.characteristic-polynomial.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
     ExactReplayCheckerDeclaration(
         "matrix.normal_form.smith.compute",
         IntegerMatrixRequest,
         "check_matrix_smith_normal_form",
         "matrix.smith-normal-form.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
     ),
 )
 

src/jacobian/domains/number_theory/checkers.py

@@ -62,6 +62,7 @@ def _integer_lcm_runtime(
         "integer.prime-factorization.flint-replay",
         entrypoint_module=_EXACT_DOMAIN_ENTRYPOINT,
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         replay_method="Python-FLINT prime-factorization replay",
         reason=(
             "operator-authorized Python-FLINT checker independent of the "
@@ -88,6 +89,7 @@ def _integer_lcm_runtime(
         "integer.powerful.flint-replay",
         entrypoint_module=_EXACT_DOMAIN_ENTRYPOINT,
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         replay_method="Python-FLINT powerful-number replay",
         reason=(
             "operator-authorized Python-FLINT checker independent of the "
@@ -115,6 +117,7 @@ def _integer_lcm_runtime(
         "modular.polynomial-residue-image.flint-replay",
         entrypoint_module=_EXACT_DOMAIN_ENTRYPOINT,
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         replay_method="Python-FLINT exhaustive modular-polynomial replay",
         reason=(
             "operator-authorized Python-FLINT checker independently reconstructs "

src/jacobian/domains/polynomial/checkers.py

@@ -227,6 +227,7 @@ def _materialized_syzygy_supports(payload: object) -> bool:
         "check_polynomial_gcd",
         "polynomial.gcd.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         supports_input=_univariate_polynomial("left", "right"),
     ),
     ExactReplayCheckerDeclaration(
@@ -235,6 +236,7 @@ def _materialized_syzygy_supports(payload: object) -> bool:
         "check_polynomial_resultant",
         "polynomial.resultant.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         supports_input=_univariate_polynomial("left", "right"),
     ),
     ExactReplayCheckerDeclaration(
@@ -243,6 +245,7 @@ def _materialized_syzygy_supports(payload: object) -> bool:
         "check_polynomial_discriminant",
         "polynomial.discriminant.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         supports_input=_univariate_polynomial("polynomial"),
     ),
     ExactReplayCheckerDeclaration(
@@ -251,6 +254,7 @@ def _materialized_syzygy_supports(payload: object) -> bool:
         "check_polynomial_square_free",
         "polynomial.square-free.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         supports_input=_univariate_polynomial("polynomial"),
     ),
     ExactReplayCheckerDeclaration(
@@ -259,6 +263,7 @@ def _materialized_syzygy_supports(payload: object) -> bool:
         "check_polynomial_factorization",
         "polynomial.factorization.flint-replay",
         provider_runtime_factory=_flint_exact_replay_runtime,
+        optional=True,
         supports_input=_univariate_polynomial("polynomial"),
     ),
 )

src/jacobian/exact_domain_checkers.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable, Mapping
+from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any, Literal
 
@@ -59,7 +59,6 @@
 from jacobian.verification.service import VerificationService
 
 _LOGGER = logging.getLogger(__name__)
-_OPTIONAL_EXACT_REPLAY_PROVIDERS = frozenset({"jacobian.exact-domain-checkers"})
 
 
 @dataclass(frozen=True, slots=True)
@@ -86,60 +85,45 @@ class _InstalledDeclaration:
 @dataclass(frozen=True, slots=True)
 class _DeclaredRuntimeGroup:
     probe: CapabilityProviderRuntime
-    factory: Callable[..., CapabilityProviderRuntime]
     members: tuple[tuple[InstalledDomainBundle, ExactReplayCheckerDeclaration], ...]
-    factories: tuple[Callable[..., CapabilityProviderRuntime], ...]
-
-
-def _declaration_factory(
-    declaration: ExactReplayCheckerDeclaration,
-) -> Callable[..., CapabilityProviderRuntime]:
-    factory = declaration.provider_runtime_factory
-    if factory is None:
-        raise ValueError(
-            "exact replay checker declaration requires a provider runtime factory"
-        )
-    return factory
 
 
 def _declared_runtime_groups(
     pairs: tuple[tuple[InstalledDomainBundle, ExactReplayCheckerDeclaration], ...],
 ) -> tuple[_DeclaredRuntimeGroup, ...]:
-    probes: dict[
-        Callable[..., CapabilityProviderRuntime], CapabilityProviderRuntime
-    ] = {}
     grouped: dict[
         str,
         tuple[
             CapabilityProviderRuntime,
-            list[Callable[..., CapabilityProviderRuntime]],
             list[tuple[InstalledDomainBundle, ExactReplayCheckerDeclaration]],
         ],
     ] = {}
     for installed, declaration in pairs:
-        factory = _declaration_factory(declaration)
-        probe = probes.setdefault(factory, factory())
+        factory = object.__getattribute__(declaration, "provider_runtime_factory")
+        probe = factory() if factory is not None else declaration.provider_runtime
+        if probe is None:
+            continue
+        if not isinstance(probe, CapabilityProviderRuntime):
+            raise TypeError(
+                "provider runtime factory must return CapabilityProviderRuntime"
+            )
         current = grouped.get(probe.provider)
         if current is None:
-            grouped[probe.provider] = (probe, [factory], [(installed, declaration)])
+            grouped[probe.provider] = (probe, [(installed, declaration)])
             continue
-        existing_probe, factories, members = current
-        if existing_probe != probe:
+        existing_probe, members = current
+        if existing_probe.model_dump(mode="json") != probe.model_dump(mode="json"):
             raise ValueError(
                 "exact replay grouped distinct probes under one provider "
                 f"identity: {probe.provider}"
             )
-        if factory not in factories:
-            factories.append(factory)
         members.append((installed, declaration))
     return tuple(
         _DeclaredRuntimeGroup(
             probe=probe,
-            factory=factories[0],
             members=tuple(members),
-            factories=tuple(factories),
         )
-        for probe, factories, members in grouped.values()
+        for probe, members in grouped.values()
     )
 
 
@@ -150,15 +134,13 @@ def _authorize_replay_operation(
     authorize: bool,
     provider_runtime: CapabilityProviderRuntime,
     source_available: bool,
+    optional: bool,
     capability_id: str,
     diagnostics: list[CapabilityDiagnostic],
 ) -> str | None:
     if provider_runtime.availability is CapabilityProviderAvailability.AVAILABLE:
         return installer.install(operation, authorize=authorize).checker_id
-    can_omit = (
-        provider_runtime.provider in _OPTIONAL_EXACT_REPLAY_PROVIDERS
-        and source_available
-    )
+    can_omit = optional and source_available
     if not can_omit:
         return installer.install(operation, authorize=authorize).checker_id
     diagnostic = CapabilityDiagnostic(
@@ -191,18 +173,11 @@ def _authorized_provider_runtimes(
             for _installed, declaration in group.members
             if (checker_id := checker_ids[declaration.capability_id]) is not None
         )
-        runtime = group.factory(checker_ids=authorized)
-        for factory in group.factories:
-            if factory is group.factory:
-                continue
-            other = factory(checker_ids=authorized)
-            if other != runtime:
-                raise ValueError(
-                    "exact replay grouped distinct runtimes under one provider "
-                    f"identity: {group.probe.provider}"
-                )
+        runtime = group.probe.model_copy(update={"checker_ids": authorized})
         existing = provider_runtimes.get(runtime.provider)
-        if existing is not None and existing != runtime:
+        if existing is not None and existing.model_dump(
+            mode="json"
+        ) != runtime.model_dump(mode="json"):
             raise ValueError(
                 "exact replay grouped distinct runtimes under one provider "
                 f"identity: {runtime.provider}"
@@ -225,15 +200,26 @@ def install_exact_domain_checkers(
     """Install independent exact replay against dynamically registered schemas."""
 
     installer = CheckerInstaller(checkers)
-    groups = _declared_runtime_groups(_available_declaration_bundles(bundles))
+    available_declarations = _available_declaration_bundles(bundles)
     checker_ids: dict[str, str | None] = {}
     declaration_providers: dict[str, str] = {}
     diagnostics: list[CapabilityDiagnostic] = []
+    checker_ids.update(
+        (declaration.capability_id, None)
+        for _installed, declaration in available_declarations
+    )
+    if not authorize and not installer.bind_existing:
+        return ExactDomainCheckerInstallation(
+            checker_ids=checker_ids,
+            provider_runtimes={},
+            declaration_providers={},
+        )
     source_available = (
         exact_domain_checker_source_provider_runtime().availability
         is CapabilityProviderAvailability.AVAILABLE
     )
     with batch_checker_manifest_measurement():
+        groups = _declared_runtime_groups(available_declarations)
         for group in groups:
             for installed, declaration in group.members:
                 declaration_providers[declaration.capability_id] = group.probe.provider
@@ -264,6 +250,7 @@ def install_exact_domain_checkers(
                     authorize=authorize,
                     provider_runtime=group.probe,
                     source_available=source_available,
+                    optional=declaration.optional,
                     capability_id=declaration.capability_id,
                     diagnostics=diagnostics,
                 )