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.

Author: morluto

src/jacobian/checker_operations.py

@@ -90,6 +90,11 @@ class ExactReplayCheckerDeclaration:
     tags are used as written. If the domain omits the metadata, all four fields
     are strictly constructed from the producer capability ID so that no
     verifier metadata is absent at installation.
+
+    A declaration may own its complete clean-process provider runtime directly.
+    Existing built-in families may continue to use the legacy central runtime
+    registry while they are migrated. Direct runtimes must not carry checker
+    IDs before operator authorization.
     """
 
     capability_id: str
@@ -102,6 +107,7 @@ class ExactReplayCheckerDeclaration:
         "operator-authorized Python-FLINT exact replay independent of the "
         "SymPy producer"
     )
+    provider_runtime: CapabilityProviderRuntime | None = None
     verification_capability_id: str | None = None
     verification_title: str | None = None
     verification_description: str | None = None
@@ -120,6 +126,10 @@ def __post_init__(self) -> None:
                 raise ValueError(
                     f"exact replay checker declaration {field} must not be empty"
                 )
+        if self.provider_runtime is not None and self.provider_runtime.checker_ids:
+            raise ValueError(
+                "declaration-owned provider runtime must not pre-authorize checker IDs"
+            )
         derived_id = derive_verification_capability_id(self.capability_id)
         explicit_text = (
             self.verification_title,

src/jacobian/exact_domain_checkers.py

@@ -14,7 +14,6 @@
 from jacobian.capability_adapters import CapabilityAdapter
 from jacobian.capability_errors import CapabilityError, CapabilityInvocationError
 from jacobian.checker_artifacts import put_witness_envelope
-from jacobian.checker_identity import batch_checker_manifest_measurement
 from jacobian.checker_installation import CheckerInstaller
 from jacobian.checker_operations import CheckerOperation, ExactReplayCheckerDeclaration
 from jacobian.contracts.capabilities import (
@@ -160,6 +159,8 @@ class _InstalledDeclaration:
 
 
 def _provider_runtime_key(declaration: ExactReplayCheckerDeclaration) -> str:
+    if declaration.provider_runtime is not None:
+        return f"declaration:{declaration.capability_id}"
     if declaration.entrypoint_module == "jacobian_checkers.linear":
         return {
             "check_rational_solution": "linear-solution",
@@ -234,76 +235,86 @@ def install_exact_domain_checkers(
             features=("standard-library-rational-replay", "clean-process-checker"),
         ),
     }
+    available_declarations = _available_declaration_bundles(bundles)
     provider_runtimes = {
         runtime_key: factory() for runtime_key, factory in runtime_factories.items()
     }
+    for _installed, declaration in available_declarations:
+        if declaration.provider_runtime is None:
+            continue
+        runtime_key = _provider_runtime_key(declaration)
+        previous = provider_runtimes.setdefault(
+            runtime_key,
+            declaration.provider_runtime,
+        )
+        if previous != declaration.provider_runtime:
+            raise ValueError(
+                "exact replay declarations disagree on their owned provider runtime"
+            )
     checker_ids: dict[str, str | None] = {}
     declarations_by_id: dict[str, ExactReplayCheckerDeclaration] = {}
     diagnostics: list[CapabilityDiagnostic] = []
     exact_checker_source_available = (
         exact_domain_checker_source_provider_runtime().availability
         is CapabilityProviderAvailability.AVAILABLE
     )
-    with batch_checker_manifest_measurement():
-        for installed, declaration in _available_declaration_bundles(bundles):
-            declarations_by_id[declaration.capability_id] = declaration
-            runtime_key = _provider_runtime_key(declaration)
-            provider_runtime = provider_runtimes[runtime_key]
-            operation = CheckerOperation(
-                name=f"{declaration.capability_id} independent {declaration.replay_method}",
-                entrypoint=(f"{declaration.entrypoint_module}:{declaration.function}"),
-                evidence_kind=EvidenceKind.WITNESS,
-                format_id=declaration.format_id,
-                format_version="1",
-                claim_schema_uris=(
-                    installed.input_schema_uris[declaration.request_model],
+    for installed, declaration in available_declarations:
+        declarations_by_id[declaration.capability_id] = declaration
+        runtime_key = _provider_runtime_key(declaration)
+        provider_runtime = provider_runtimes[runtime_key]
+        operation = CheckerOperation(
+            name=f"{declaration.capability_id} independent {declaration.replay_method}",
+            entrypoint=(f"{declaration.entrypoint_module}:{declaration.function}"),
+            evidence_kind=EvidenceKind.WITNESS,
+            format_id=declaration.format_id,
+            format_version="1",
+            claim_schema_uris=(installed.input_schema_uris[declaration.request_model],),
+            semantics_uris=(installed.semantics_uri,),
+            candidate_schema_uris=(
+                installed.result_schema_uris[declaration.capability_id],
+            ),
+            reason=declaration.reason,
+            provider_runtime=provider_runtime,
+        )
+        if (
+            provider_runtime.availability
+            is not CapabilityProviderAvailability.AVAILABLE
+        ):
+            can_omit = (
+                runtime_key in _OPTIONAL_EXACT_REPLAY_PROVIDER_KEYS
+                and exact_checker_source_available
+            )
+            if not can_omit:
+                checker_ids[declaration.capability_id] = installer.install(
+                    operation,
+                    authorize=authorize,
+                ).checker_id
+                continue
+            diagnostic = CapabilityDiagnostic(
+                code="EXACT_REPLAY_PROVIDER_UNAVAILABLE",
+                stage="provider_availability",
+                message=(
+                    f"Independent replay for {declaration.capability_id!r} is "
+                    "not installed: "
+                    f"{provider_runtime.diagnostic or 'the provider is unavailable.'}"
                 ),
-                semantics_uris=(installed.semantics_uri,),
-                candidate_schema_uris=(
-                    installed.result_schema_uris[declaration.capability_id],
+                hint=(
+                    "Install or repair the optional python-flint backend, then retry."
                 ),
-                reason=declaration.reason,
-                provider_runtime=provider_runtime,
+                details={
+                    "capability_id": declaration.capability_id,
+                    "provider": provider_runtime.provider,
+                    "checker_authorization_affected": True,
+                },
             )
-            if (
-                provider_runtime.availability
-                is not CapabilityProviderAvailability.AVAILABLE
-            ):
-                can_omit = (
-                    runtime_key in _OPTIONAL_EXACT_REPLAY_PROVIDER_KEYS
-                    and exact_checker_source_available
-                )
-                if not can_omit:
-                    checker_ids[declaration.capability_id] = installer.install(
-                        operation,
-                        authorize=authorize,
-                    ).checker_id
-                    continue
-                diagnostic = CapabilityDiagnostic(
-                    code="EXACT_REPLAY_PROVIDER_UNAVAILABLE",
-                    stage="provider_availability",
-                    message=(
-                        f"Independent replay for {declaration.capability_id!r} is "
-                        "not installed: "
-                        f"{provider_runtime.diagnostic or 'the provider is unavailable.'}"
-                    ),
-                    hint=(
-                        "Install or repair the optional python-flint backend, then retry."
-                    ),
-                    details={
-                        "capability_id": declaration.capability_id,
-                        "provider": provider_runtime.provider,
-                        "checker_authorization_affected": True,
-                    },
-                )
-                diagnostics.append(diagnostic)
-                _LOGGER.warning("%s", diagnostic.message)
-                checker_ids[declaration.capability_id] = None
-                continue
-            checker_ids[declaration.capability_id] = installer.install(
-                operation,
-                authorize=authorize,
-            ).checker_id
+            diagnostics.append(diagnostic)
+            _LOGGER.warning("%s", diagnostic.message)
+            checker_ids[declaration.capability_id] = None
+            continue
+        checker_ids[declaration.capability_id] = installer.install(
+            operation,
+            authorize=authorize,
+        ).checker_id
     authorized_ids = {
         runtime_key: tuple(
             checker_id
@@ -313,13 +324,21 @@ def install_exact_domain_checkers(
         )
         for runtime_key in provider_runtimes
     }
+    resolved_provider_runtimes: dict[str, CapabilityProviderRuntime] = {}
+    for runtime_key, provider_runtime in provider_runtimes.items():
+        checker_ids_for_runtime = authorized_ids[runtime_key]
+        factory = runtime_factories.get(runtime_key)
+        resolved_provider_runtimes[runtime_key] = (
+            factory(checker_ids=checker_ids_for_runtime)
+            if factory is not None
+            else provider_runtime.model_copy(
+                update={"checker_ids": checker_ids_for_runtime}
+            )
+        )
     return ExactDomainCheckerInstallation(
         checker_ids=checker_ids,
         diagnostics=tuple(diagnostics),
-        provider_runtimes={
-            runtime_key: factory(checker_ids=authorized_ids[runtime_key])
-            for runtime_key, factory in runtime_factories.items()
-        },
+        provider_runtimes=resolved_provider_runtimes,
     )
 
 

src/jacobian/portfolio/core_installation.py

@@ -4,6 +4,7 @@
 
 from dataclasses import dataclass
 
+from jacobian.checker_identity import batch_checker_manifest_measurement
 from jacobian.contracts.capabilities import CapabilityProviderAvailability
 from jacobian.domain_bundles import DomainBundle
 from jacobian.domains.polynomial_nullstellensatz.core import (
@@ -68,7 +69,7 @@ def _install_nullstellensatz(self) -> None:
         """Install the named Nullstellensatz family at the composition root.
 
         This family has an artifact-producing core and an optional Singular
-        producer that depends on that core.  It is deliberately not represented
+        producer that depends on that core. It is deliberately not represented
         as a generic portfolio callback: ordinary portfolio plans contain only
         :class:`DomainBundle` declarations.
         """
@@ -192,15 +193,19 @@ def install_domain_verification(
         }
         if not exact_bundles:
             return None
-        adapters, installation = install_exact_domain_verification(
-            ctx.store,
-            ctx.schemas,
-            ctx.artifacts,
-            ctx.verification,
-            ctx.checkers,
-            bundles=exact_bundles,
-            authorize=ctx.authorizes_bundled_checkers,
-        )
+        # Batch identity material across the complete declaration set while the
+        # exact-domain installer resolves both legacy and declaration-owned
+        # provider runtimes. Nested measurement remains safe for direct callers.
+        with batch_checker_manifest_measurement():
+            adapters, installation = install_exact_domain_verification(
+                ctx.store,
+                ctx.schemas,
+                ctx.artifacts,
+                ctx.verification,
+                ctx.checkers,
+                bundles=exact_bundles,
+                authorize=ctx.authorizes_bundled_checkers,
+            )
         for adapter in adapters:
             self.context.register_capability(adapter)
         return installation

tests/unit/test_checker_declaration_runtime.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import pytest
+
+from jacobian.checker_operations import ExactReplayCheckerDeclaration
+from jacobian.contracts.capabilities import CapabilityInstallTier
+from jacobian.contracts.exact import CanonicalRational
+from jacobian.provider_runtime import source_provider_runtime
+
+
+def _runtime():
+    return source_provider_runtime(
+        "jacobian.test-declaration-checker",
+        version="1",
+        entrypoint="jacobian_checkers.linear:check_rational_solution",
+        install_tier=CapabilityInstallTier.T1,
+        license_id="MIT",
+        features=("clean-process-checker",),
+    )
+
+
+def test_exact_replay_declaration_owns_unassigned_provider_runtime() -> None:
+    runtime = _runtime()
+    declaration = ExactReplayCheckerDeclaration(
+        "test.compute.value",
+        CanonicalRational,
+        "check_rational_solution",
+        "test.value.replay",
+        entrypoint_module="jacobian_checkers.linear",
+        provider_runtime=runtime,
+    )
+
+    assert declaration.provider_runtime == runtime
+    assert declaration.provider_runtime.checker_ids == ()
+
+
+def test_exact_replay_declaration_rejects_preauthorized_runtime() -> None:
+    runtime = _runtime().model_copy(update={"checker_ids": ("checker:test",)})
+
+    with pytest.raises(ValueError, match="must not pre-authorize checker IDs"):
+        ExactReplayCheckerDeclaration(
+            "test.compute.value",
+            CanonicalRational,
+            "check_rational_solution",
+            "test.value.replay",
+            entrypoint_module="jacobian_checkers.linear",
+            provider_runtime=runtime,
+        )