refactor(checkers): let declarations own provider runtimes

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