Avoid eager dataset setup and repeated startup runtime validation

data/datasets/alfworld.py

@@ -125,13 +125,16 @@
 
 from __future__ import annotations
 
+from data.scaffolds.declarations import TaskDeclarations
+
 from data.scaffolds.scorer_input import bind as bind_scorer_input
 
 from harnessgrad.paths import dataset_root
 
 #: See `data/scaffolds/registry.py:KIND_DATASET`. A fixture is platform test material and is
 #: deliberately not offered as a choice in the Launch form.
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 
 #: The oracle control submits the benchmark's own released reference; see
 #: `data/dataset.schema.json` (`module.oracle_about`) for what the other value means.
@@ -676,8 +679,8 @@ def load() -> tuple[list[dict], dict[str, str]]:
 DEFAULT_ENV: dict = {"kind": "files"}
 
 ENV: dict[str, dict] = {r["task_id"]: {"kind": "files"} for r in _RECORDS}
-SETUP: dict[str, dict] = {r["task_id"]: _setup(r) for r in _RECORDS}
-VERIFY: dict[str, dict] = {r["task_id"]: _verifier(r) for r in _RECORDS}
+SETUP = TaskDeclarations(_RECORDS, _setup)
+VERIFY = TaskDeclarations(_RECORDS, _verifier)
 
 if _STUDY:
     # **The declared sides, without a hash.** The train side is exactly the study file and

data/datasets/appworld.py

@@ -67,6 +67,7 @@
 from harnessgrad.paths import dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 
 DATASET_ROOT = dataset_root('appworld')
 SRC_DIR = DATASET_ROOT / "_src"

data/datasets/bird_minidev.py

@@ -64,6 +64,7 @@
 
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 
 #: The oracle control submits the benchmark's own released reference; see
 #: `data/dataset.schema.json` (`module.oracle_about`) for what the other value means.

data/datasets/costbench.py

@@ -14,6 +14,7 @@
 from data.scaffolds.asset_module import load as load_asset
 
 KIND = 'dataset'
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = 'costbench'
 ORACLE_SOURCE = 'released'
 SPLIT_SOURCE = 'declared'

data/datasets/costbench_dynamic.py

@@ -9,6 +9,7 @@
 from data.scaffolds.asset_module import load as load_asset
 
 KIND = 'dataset'
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = 'costbench'
 ORACLE_SOURCE = 'released'
 SPLIT_SOURCE = 'declared'

data/datasets/ds1000.py

@@ -64,6 +64,7 @@
 from harnessgrad.paths import dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = "ds1000"
 
 #: The oracle control submits the benchmark's own released reference; see
@@ -199,6 +200,13 @@ def __iter__(self):
     def __len__(self):
         return len(included_indices())
 
+    def __contains__(self, key):
+        try:
+            self._index(key)
+            return True
+        except KeyError:
+            return False
+
     @staticmethod
     def _index(key):
         if not isinstance(key, str) or not key.startswith("ds1000/"):
@@ -275,7 +283,12 @@ def verify_for(index: int) -> dict:
     return spec
 
 
-SETUP = {task_id(i): setup_for(i) for i in included_indices()}
+class _Setups(_TaskDeclarations):
+    def __getitem__(self, key):
+        return setup_for(self._index(key))
+
+
+SETUP = _Setups()
 
 
 class _Verifiers(_TaskDeclarations):

data/datasets/jericho.py

@@ -68,13 +68,17 @@
 """
 from __future__ import annotations
 
+from data.scaffolds.declarations import TaskDeclarations
+from data.scaffolds.scorer_input import source_digest
+
 from data.scaffolds.scorer_input import bind as bind_scorer_input
 
 from pathlib import Path
 
 from harnessgrad.paths import dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = "jericho"
 
 #: The oracle control submits the benchmark's own released reference; see
@@ -217,13 +221,12 @@ def _goal(record: dict) -> str:
 
 def _setup(record: dict) -> dict:
     """Publish only a stdlib client and bind the trusted single-game service."""
-    import hashlib
     import json
     from harnessgrad.dataset_runtime import bind, resolve_execution
 
     binding = bind(resolve_execution(MANAGED_EXECUTION_RUNTIME), "service")
     rom = Path(record["path"]).resolve(strict=True)
-    digest = hashlib.sha256(rom.read_bytes()).hexdigest()
+    digest = source_digest(rom)
     return {
         "files": [
             {"path": LOG_FILE, "content": ""},
@@ -379,8 +382,8 @@ def load() -> tuple[list[dict], dict[str, str]]:
     return tasks, {r["task_id"]: "" for r in records}
 
 
-def _map(builder) -> dict:
-    return {r["task_id"]: builder(r) for r in __getattr__("_RECORDS")}
+def _map(builder):
+    return TaskDeclarations(__getattr__("_RECORDS"), builder)
 
 
 def __getattr__(name: str):

data/datasets/kumo.py

@@ -16,6 +16,7 @@
 from harnessgrad.paths import DataRootNotConfigured, dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 ORACLE_SOURCE = "released"
 OFFICIAL_EVAL = None
 #: `"none"`: the benchmark states no division, so this adapter declares none. It used to

data/datasets/livecodebench.py

@@ -77,6 +77,7 @@ def __getattr__(self, name):
 
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = "livecodebench"
 #: The reference solutions are **ours** and the released tests validate them; see the module
 #: docstring and `data/dataset.schema.json` (`module.oracle_about`).

data/datasets/locomo.py

@@ -19,6 +19,7 @@
 from harnessgrad.paths import dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = "locomo"
 ORACLE_SOURCE = "released"
 OFFICIAL_EVAL = None

data/datasets/scienceworld.py

@@ -110,6 +110,9 @@
 """
 from __future__ import annotations
 
+from data.scaffolds.declarations import TaskDeclarations
+from data.scaffolds.scorer_input import source_digest
+
 from data.scaffolds.scorer_input import bind as bind_scorer_input
 
 import os
@@ -119,6 +122,7 @@
 
 #: See `data/scaffolds/registry.py:KIND_DATASET`.
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = "scienceworld"
 
 #: The oracle control submits the benchmark's own released reference; see
@@ -405,15 +409,14 @@ def _goal(record: dict) -> str:
 
 def _setup(record: dict, *, binding=None, world_jar=None) -> dict:
     """Publish a stdlib client and bind a fixed-task managed service."""
-    import hashlib
     from harnessgrad.dataset_runtime import bind, resolve_execution
     binding = binding or bind(resolve_execution(MANAGED_EXECUTION_RUNTIME), 'service')
     jar = world_jar or _runtime_payload(binding)
     task = {'task_name': record['task_name'], 'variation': record['variation'], 'simplifications': ''}
     java = JRE_STAGED / 'bin/java'
     context = {'task': task, 'jar': str(jar), 'java_home': str(JRE_STAGED.resolve()),
-               'java_sha256': hashlib.sha256(java.read_bytes()).hexdigest(),
-               'source_sha256': hashlib.sha256((ASSETS/'relay.py').read_bytes()).hexdigest()}
+               'java_sha256': source_digest(java),
+               'source_sha256': source_digest(ASSETS/'relay.py')}
     return {'files': [
         {'path': TASK_IN_WORKDIR, 'content': json.dumps(task)},
         {'path': LOG_IN_WORKDIR, 'content': ''},
@@ -603,12 +606,10 @@ def _split() -> dict | None:
 
 
 class _Lazy:
-    """`ENV`, `SETUP` and `VERIFY`, built on first use.
+    """Index tasks on first use; build SETUP/VERIFY only for requested keys.
 
-    Three dicts with one entry per task, and there are 7,207 tasks. Building them at import
-    would make importing this adapter cost a JVM start plus 21,621 dict constructions -- the
-    lesson the QA batch recorded in `docs/dataset-onboarding-skill.md` §15.7. The platform
-    asks for them through module `__getattr__` below.
+    Cataloguing the 7,207-task pool does not prepare services or hash the JAR.
+    Shared resources are reused only within the registry's declaration scope.
     """
 
     def __init__(self) -> None:
@@ -620,13 +621,9 @@ def get(self, name: str):
             if name == 'ENV':
                 self._cache[name] = {r['task_id']: {'kind':'files'} for r in records}
             elif name == 'SETUP':
-                from harnessgrad.dataset_runtime import bind, resolve_execution
-                binding = bind(resolve_execution(MANAGED_EXECUTION_RUNTIME), 'service')
-                jar = _runtime_payload(binding)
-                self._cache[name] = {r['task_id']: _setup(r, binding=binding, world_jar=jar) for r in records}
+                self._cache[name] = TaskDeclarations(records, _setup)
             else:
-                jar = _runtime_payload()
-                self._cache[name] = {r['task_id']: _verifier(r, world_jar=jar) for r in records}
+                self._cache[name] = TaskDeclarations(records, _verifier)
         return self._cache[name]
 
 

data/datasets/scienceworld/runtime.py

@@ -1,9 +1,9 @@
 """Read-only managed payload lookup; never import the third-party package."""
-import hashlib
+from data.scaffolds.scorer_input import source_digest
 from pathlib import Path
 JAR_SHA256 = "f4e1fdce0cdabe1bab7128a7e78ea820210d9db3c28d7272bcbbc3f3a1c5f0d7"
 def payload(binding):
     jar = Path(binding['path']) / 'lib/python3.12/site-packages/scienceworld/scienceworld.jar'
-    if jar.is_symlink() or not jar.is_file() or hashlib.sha256(jar.read_bytes()).hexdigest() != JAR_SHA256:
+    if jar.is_symlink() or not jar.is_file() or source_digest(jar) != JAR_SHA256:
         raise RuntimeError('managed ScienceWorld JAR missing or differs from sealed scorer; explicitly prepare the scienceworld execution runtime')
     return jar

data/datasets/spreadsheetbench.py

@@ -59,6 +59,7 @@
 from harnessgrad.paths import dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 MANAGED_EXECUTION_RUNTIME = "spreadsheetbench"
 
 #: The oracle control submits the benchmark's own released reference; see
@@ -233,8 +234,8 @@ def _verifier(record: dict) -> dict:
     """
     from harnessgrad.dataset_runtime import bind, resolve_execution
     binding = bind(resolve_execution("spreadsheetbench"), "verifier")
-    from data.scaffolds.spreadsheet_runtime.native import require_runtime, snapshot_native_python
-    native = snapshot_native_python(require_runtime())
+    from data.scaffolds.spreadsheet_runtime.metadata import selected_native
+    native = selected_native()
     # Pin selected contents, never a mutable environment/path at worker launch.
     native.pop("soffice", None)
     native_content = json.dumps(native, sort_keys=True, separators=(",", ":"))

data/datasets/spreadsheetbench2.py

@@ -20,6 +20,7 @@
 from data.scaffolds.scorer_input import bind as bind_scorer_input
 
 KIND = 'dataset'
+VERIFIER_KIND = 'command'
 ORACLE_SOURCE = 'released'
 OFFICIAL_EVAL = None
 SPLIT = None
@@ -83,9 +84,9 @@ def _setup(record):
 
 def _verify(record):
     from harnessgrad.dataset_runtime import bind, resolve_execution
-    from data.scaffolds.spreadsheet_runtime.native import require_runtime, snapshot_native_python
+    from data.scaffolds.spreadsheet_runtime.metadata import selected_native
     binding = bind(resolve_execution('spreadsheetbench2'), 'verifier')
-    native = snapshot_native_python(require_runtime())
+    native = selected_native()
     native.pop('soffice', None)  # Never execute an external, unbound launcher.
     native_content = json.dumps(native, sort_keys=True, separators=(',', ':'))
     sources = {

data/datasets/tau2_bench.py

@@ -74,6 +74,7 @@
 from harnessgrad.paths import configured_path, dataset_root
 
 KIND = "dataset"
+VERIFIER_KIND = 'command'
 ORACLE_SOURCE = "released"
 #: No held-out division exists in this release; the pool is the scored set.
 OFFICIAL_EVAL = None
@@ -192,7 +193,7 @@ def _case(tid: str, row: dict) -> Path:
 
 
 def _execution_runtime():
-    """Resolve and revalidate the selected dependency artifact on every declaration.
+    """Resolve the selected dependency artifact within the current declaration batch.
 
     This is an external store reference, not a relocatable copy or a sealed OS.
     The manager inventories the venv and base executable; host libraries and the

data/datasets/tau3_bench.py

@@ -16,6 +16,7 @@
 from harnessgrad.paths import configured_path, dataset_root
 
 KIND = 'dataset'
+VERIFIER_KIND = 'command'
 ORACLE_SOURCE = 'released'
 # Upstream's full-base reporting protocol, distinct from the held-out split.
 OFFICIAL_EVAL = None
@@ -93,7 +94,7 @@ def model_spec(role, tools=None):
 
 
 def _execution_runtime():
-    """Resolve and revalidate the selected dependency artifact on every declaration.
+    """Resolve the selected dependency artifact within the current declaration batch.
 
     This is an external store reference, not a relocatable copy or a sealed OS.
     The manager inventories the venv and base executable; host libraries and the

data/scaffolds/declarations.py

@@ -0,0 +1,50 @@
+"""Lazy task declarations and short-lived metadata reuse; never execution admission."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
+
+_CACHE = ContextVar('dataset_declaration_metadata', default=None)
+
+
+@contextmanager
+def scope():
+    """One synchronous metadata operation; nested readers share its snapshot."""
+    if _CACHE.get() is not None:
+        yield
+        return
+    token = _CACHE.set({})
+    try:
+        yield
+    finally:
+        _CACHE.reset(token)
+
+
+def once(key, build):
+    """Reuse trusted metadata only while its declaration scope is active."""
+    cache = _CACHE.get()
+    if cache is None:
+        return build()
+    if key not in cache:
+        cache[key] = build()
+    return cache[key]
+
+
+class TaskDeclarations(Mapping):
+    """Enumerate keys without building values; construct only requested tasks."""
+    def __init__(self, records, build):
+        self._records = {record['task_id']: record for record in records}
+        self._build = build
+
+    def __iter__(self):
+        return iter(self._records)
+
+    def __len__(self):
+        return len(self._records)
+
+    def __contains__(self, key):
+        return key in self._records
+
+    def __getitem__(self, key):
+        return self._build(self._records[key])

data/scaffolds/native_qa/__init__.py

@@ -144,11 +144,16 @@ def __iter__(self):
     def __len__(self):
         return len(self.rows)
 
+    def __contains__(self, key):
+        return key in self.rows
+
     def __getitem__(self, key):
         return self.dataset.verifier(self.rows[key])
 
 
 class Dataset:
+    VERIFIER_KIND = "command"
+
     def __init__(self,name,root=None):
         self.name=name; self.source=CATALOG[name]
         if root is None: