Batch checker identity work and narrow lifecycle fixtures (#1357)

* test: narrow runtime fixture ownership

* perf(checkers): batch installation identity measurement

Author: morluto

src/jacobian/checker_identity.py

@@ -11,7 +11,9 @@
 import importlib.util
 import re
 import sys
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Iterable, Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass
 from functools import lru_cache
 from importlib import metadata
@@ -52,6 +54,38 @@ class _ResolvedModule:
     is_package: bool
 
 
+@dataclass(slots=True)
+class _ManifestMeasurementBatch:
+    source_modules: dict[
+        tuple[tuple[str, ...], frozenset[str]], tuple[CheckerSourceModule, ...]
+    ]
+    distributions: dict[str, CheckerPythonDistribution]
+    package_owners: Mapping[str, list[str]] | None = None
+    python_runtime: CheckerPythonRuntime | None = None
+
+
+_ACTIVE_MEASUREMENT_BATCH: ContextVar[_ManifestMeasurementBatch | None] = ContextVar(
+    "jacobian_checker_manifest_measurement_batch", default=None
+)
+
+
+@contextmanager
+def batch_checker_manifest_measurement() -> Iterator[None]:
+    """Share immutable identity measurements across one installation operation."""
+
+    active = _ACTIVE_MEASUREMENT_BATCH.get()
+    if active is not None:
+        yield
+        return
+    token = _ACTIVE_MEASUREMENT_BATCH.set(
+        _ManifestMeasurementBatch(source_modules={}, distributions={})
+    )
+    try:
+        yield
+    finally:
+        _ACTIVE_MEASUREMENT_BATCH.reset(token)
+
+
 class _DeclaredSourceLoader(importlib.abc.Loader):
     """Load one manifest-bound module from remeasured source only."""
 
@@ -154,19 +188,19 @@ def build_checker_manifest(
     first_party_packages = frozenset(
         {_JACOBIAN_PACKAGE, entrypoint_module.split(".", 1)[0]}
     )
-    checker_source_modules = _collect_source_modules(
+    checker_source_modules = _batched_source_modules(
         (entrypoint_module,),
         first_party_packages=first_party_packages,
     )
-    worker_source_modules = _collect_source_modules(
+    worker_source_modules = _batched_source_modules(
         (_CHECKER_WORKER_MODULE, *extra_modules),
         first_party_packages=first_party_packages,
     )
     return CheckerManifest(
         entrypoint=entrypoint,
         checker_source_modules=checker_source_modules,
         worker_source_modules=worker_source_modules,
-        python_runtime=_python_runtime(),
+        python_runtime=_batched_python_runtime(),
         python_distributions=_collect_python_distributions(
             (*checker_source_modules, *worker_source_modules),
             first_party_packages=first_party_packages,
@@ -206,6 +240,31 @@ def require_manifest_unchanged(manifest: CheckerManifest) -> str:
     return checker_implementation_digest(measured)
 
 
+def require_manifest_material_unchanged(manifest: CheckerManifest) -> str:
+    """Reject changes to every execution artifact already bound by a manifest.
+
+    The worker performs full dependency discovery before loading the checker.  Its
+    post-execution check only needs to remeasure that closed set: the import guard
+    prevents undeclared code from entering the process, while direct remeasurement
+    avoids rebuilding and reparsing the complete import graph a second time.
+    """
+
+    for expected in _manifest_source_modules(manifest):
+        resolved = _resolve_module(expected.module)
+        measured = _source_digest(resolved.path.read_bytes())
+        if measured != expected.source_digest:
+            raise CheckerManifestError(f"checker source changed: {expected.module}")
+    if _python_runtime() != manifest.python_runtime:
+        raise CheckerManifestError("checker Python runtime changed")
+    measured_distributions = tuple(
+        _measure_python_distribution(item.distribution)
+        for item in manifest.python_distributions
+    )
+    if measured_distributions != manifest.python_distributions:
+        raise CheckerManifestError("checker Python distribution changed")
+    return checker_implementation_digest(manifest)
+
+
 def install_manifest_import_guard(manifest: CheckerManifest) -> None:
     """Reload the checker through the manifest and reject undeclared code imports."""
 
@@ -273,6 +332,24 @@ def _collect_source_modules(
     )
 
 
+def _batched_source_modules(
+    roots: tuple[str, ...],
+    *,
+    first_party_packages: frozenset[str],
+) -> tuple[CheckerSourceModule, ...]:
+    batch = _ACTIVE_MEASUREMENT_BATCH.get()
+    if batch is None:
+        return _collect_source_modules(roots, first_party_packages=first_party_packages)
+    key = (roots, first_party_packages)
+    measured = batch.source_modules.get(key)
+    if measured is None:
+        measured = _collect_source_modules(
+            roots, first_party_packages=first_party_packages
+        )
+        batch.source_modules[key] = measured
+    return measured
+
+
 def _manifest_source_modules(
     manifest: CheckerManifest,
 ) -> tuple[CheckerSourceModule, ...]:
@@ -302,7 +379,13 @@ def _collect_python_distributions(
     for bound_source in source_modules:
         source = _resolve_module(bound_source.module)
         import_roots.update(_third_party_import_roots(source, first_party_packages))
-    package_owners = metadata.packages_distributions()
+    batch = _ACTIVE_MEASUREMENT_BATCH.get()
+    if batch is not None and batch.package_owners is not None:
+        package_owners = batch.package_owners
+    else:
+        package_owners = metadata.packages_distributions()
+        if batch is not None:
+            batch.package_owners = package_owners
     distributions = set(_WORKER_DISTRIBUTIONS)
     for root in import_roots:
         owners = package_owners.get(root)
@@ -317,7 +400,7 @@ def _collect_python_distributions(
         distributions.add(owners[0])
     measured: dict[str, CheckerPythonDistribution] = {}
     for distribution in sorted(distributions, key=_distribution_key):
-        identity = _measure_python_distribution(distribution)
+        identity = _batched_python_distribution(distribution)
         key = _distribution_key(identity.distribution)
         existing = measured.setdefault(key, identity)
         if existing != identity:
@@ -475,6 +558,29 @@ def _measure_python_distribution(distribution: str) -> CheckerPythonDistribution
     )
 
 
+def _batched_python_distribution(
+    distribution: str,
+) -> CheckerPythonDistribution:
+    batch = _ACTIVE_MEASUREMENT_BATCH.get()
+    if batch is None:
+        return _measure_python_distribution(distribution)
+    key = _distribution_key(distribution)
+    measured = batch.distributions.get(key)
+    if measured is None:
+        measured = _measure_python_distribution(distribution)
+        batch.distributions[key] = measured
+    return measured
+
+
+def _batched_python_runtime() -> CheckerPythonRuntime:
+    batch = _ACTIVE_MEASUREMENT_BATCH.get()
+    if batch is None:
+        return _python_runtime()
+    if batch.python_runtime is None:
+        batch.python_runtime = _python_runtime()
+    return batch.python_runtime
+
+
 def _distribution_file_closure(
     installed: metadata.Distribution,
 ) -> tuple[int, str]:
@@ -693,9 +799,11 @@ def _source_digest(source: bytes) -> str:
 __all__ = [
     "CheckerManifestError",
     "UndeclaredCheckerImportError",
+    "batch_checker_manifest_measurement",
     "build_checker_manifest",
     "checker_implementation_digest",
     "default_checker_sandbox_policy",
     "install_manifest_import_guard",
+    "require_manifest_material_unchanged",
     "require_manifest_unchanged",
 ]

src/jacobian/checker_worker.py

@@ -22,6 +22,7 @@
     UndeclaredCheckerImportError,
     checker_implementation_digest,
     install_manifest_import_guard,
+    require_manifest_material_unchanged,
     require_manifest_unchanged,
 )
 from jacobian.contracts.capabilities import (
@@ -159,7 +160,7 @@ def _execute(manifest_json: str, request_bytes: bytes) -> CheckerWorkerSuccess:
     with contextlib.redirect_stdout(sys.stderr):
         checker = _resolve(manifest.entrypoint)
         response = checker(request)
-    measured_after = require_manifest_unchanged(manifest)
+    measured_after = require_manifest_material_unchanged(manifest)
     if measured_after != measured_before:
         raise _CheckerWorkerFailureError("SOURCE_CHANGED")
     _, runtime_digest_after = _measure_runtime(runtime)

src/jacobian/exact_domain_checkers.py

@@ -14,6 +14,7 @@
 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 (
@@ -243,63 +244,66 @@ def install_exact_domain_checkers(
         exact_domain_checker_source_provider_runtime().availability
         is CapabilityProviderAvailability.AVAILABLE
     )
-    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],),
-            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.'}"
+    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],
                 ),
-                hint=(
-                    "Install or repair the optional python-flint backend, then retry."
+                semantics_uris=(installed.semantics_uri,),
+                candidate_schema_uris=(
+                    installed.result_schema_uris[declaration.capability_id],
                 ),
-                details={
-                    "capability_id": declaration.capability_id,
-                    "provider": provider_runtime.provider,
-                    "checker_authorization_affected": True,
-                },
+                reason=declaration.reason,
+                provider_runtime=provider_runtime,
             )
-            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
+            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
     authorized_ids = {
         runtime_key: tuple(
             checker_id

tests/boundary/providers/conftest.py

@@ -0,0 +1,15 @@
+"""Lean-provider fixtures that require the complete application runtime."""
+
+from __future__ import annotations
+
+from tests.support.complete_runtime_fixtures import (
+    attached_complete_runtime,
+    authorized_complete_runtime,
+    authorized_portfolio_template,
+)
+
+__all__ = (
+    "attached_complete_runtime",
+    "authorized_complete_runtime",
+    "authorized_portfolio_template",
+)