diff --git a/data/datasets/alfworld.py b/data/datasets/alfworld.py index 3b647daf..da4f8308 100644 --- a/data/datasets/alfworld.py +++ b/data/datasets/alfworld.py @@ -125,6 +125,8 @@ 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 @@ -132,6 +134,7 @@ #: 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 diff --git a/data/datasets/appworld.py b/data/datasets/appworld.py index 0b257689..3b3f40a7 100644 --- a/data/datasets/appworld.py +++ b/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" diff --git a/data/datasets/bird_minidev.py b/data/datasets/bird_minidev.py index 04e9fadb..485cfefc 100644 --- a/data/datasets/bird_minidev.py +++ b/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. diff --git a/data/datasets/costbench.py b/data/datasets/costbench.py index 28e029c7..4a3f92d3 100644 --- a/data/datasets/costbench.py +++ b/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' diff --git a/data/datasets/costbench_dynamic.py b/data/datasets/costbench_dynamic.py index 25808943..24cf1747 100644 --- a/data/datasets/costbench_dynamic.py +++ b/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' diff --git a/data/datasets/ds1000.py b/data/datasets/ds1000.py index f8a85f46..10b97269 100644 --- a/data/datasets/ds1000.py +++ b/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): diff --git a/data/datasets/jericho.py b/data/datasets/jericho.py index 9d2e8c53..c57a1b68 100644 --- a/data/datasets/jericho.py +++ b/data/datasets/jericho.py @@ -68,6 +68,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 from pathlib import Path @@ -75,6 +78,7 @@ 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): diff --git a/data/datasets/kumo.py b/data/datasets/kumo.py index e45c3423..e021173d 100644 --- a/data/datasets/kumo.py +++ b/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 diff --git a/data/datasets/livecodebench.py b/data/datasets/livecodebench.py index 9a35d25f..85d3f421 100644 --- a/data/datasets/livecodebench.py +++ b/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`). diff --git a/data/datasets/locomo.py b/data/datasets/locomo.py index 1090f393..acb48cf7 100644 --- a/data/datasets/locomo.py +++ b/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 diff --git a/data/datasets/scienceworld.py b/data/datasets/scienceworld.py index b5702aad..8e598aed 100644 --- a/data/datasets/scienceworld.py +++ b/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] diff --git a/data/datasets/scienceworld/runtime.py b/data/datasets/scienceworld/runtime.py index 65c460ba..32c1fc81 100644 --- a/data/datasets/scienceworld/runtime.py +++ b/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 diff --git a/data/datasets/spreadsheetbench.py b/data/datasets/spreadsheetbench.py index 1c34bb08..5ac6ab1c 100644 --- a/data/datasets/spreadsheetbench.py +++ b/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=(",", ":")) diff --git a/data/datasets/spreadsheetbench2.py b/data/datasets/spreadsheetbench2.py index 3aab4579..9462f522 100644 --- a/data/datasets/spreadsheetbench2.py +++ b/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 = { diff --git a/data/datasets/tau2_bench.py b/data/datasets/tau2_bench.py index 90797e6a..9ff46238 100644 --- a/data/datasets/tau2_bench.py +++ b/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 diff --git a/data/datasets/tau3_bench.py b/data/datasets/tau3_bench.py index c4e032bc..9a56cbc5 100644 --- a/data/datasets/tau3_bench.py +++ b/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 diff --git a/data/scaffolds/declarations.py b/data/scaffolds/declarations.py new file mode 100644 index 00000000..758ee0ff --- /dev/null +++ b/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]) diff --git a/data/scaffolds/native_qa/__init__.py b/data/scaffolds/native_qa/__init__.py index bf241c6e..e1e8627d 100644 --- a/data/scaffolds/native_qa/__init__.py +++ b/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: diff --git a/data/scaffolds/native_research/adapter.py b/data/scaffolds/native_research/adapter.py index f65a50af..9b5dcf4a 100644 --- a/data/scaffolds/native_research/adapter.py +++ b/data/scaffolds/native_research/adapter.py @@ -10,6 +10,7 @@ from pathlib import Path from harnessgrad.config.areas import ConfigMissing +from data.scaffolds.declarations import once from harnessgrad.dataset_runtime import DatasetRuntimeUnavailable from harnessgrad.paths import DatasetPath, dataset_root from .profiles import COMMIT, PROFILES, SOURCES @@ -36,6 +37,8 @@ def owned(root, relative): class Adapter: + VERIFIER_KIND = 'command' + def __init__(self, name, root=None): self.name = name self.profile = PROFILES[name] @@ -48,6 +51,9 @@ def root(self): return dataset_root(self.name) def manifest(self): + return once(('native-manifest', self, str(self.root)), self._read_manifest) + + def _read_manifest(self): if self.root.resolve() == ROOT or self.root.resolve().is_relative_to(ROOT): raise ValueError('native dataset payloads must live outside the platform tree') path = self.root / 'manifest.json' @@ -113,7 +119,8 @@ def checked_world(self, manifest): raise ValueError('native world/database dependency digest mismatch') def records(self): - return {r['task_id']: r for r in self.manifest()['records']} + return once(('native-records', self, str(self.root)), + lambda: {r['task_id']: r for r in self.manifest()['records']}) def load(self): # The public inventory is independent of model/image provisioning. @@ -252,7 +259,7 @@ def verifier(self, tid): if profile['judge']: spec['judge'] = self.judge(m) if self.name == 'travelplanner': - self.checked_world(m) + # readiness() already checked this world's bytes for this task. database = self.root / 'database' if not database.is_dir(): raise ConfigMissing('travelplanner: the complete native database is required') @@ -284,6 +291,7 @@ class TaskMapping(Mapping): def __init__(self, adapter, fn): self.adapter, self.fn = adapter, fn def __iter__(self): return iter(self.adapter.records()) def __len__(self): return len(self.adapter.records()) + def __contains__(self, tid): return tid in self.adapter.records() def __getitem__(self, tid): if tid not in self.adapter.records(): raise KeyError(tid) return self.fn(tid) diff --git a/data/scaffolds/provider_worker.py b/data/scaffolds/provider_worker.py index b80ca512..bccc99e7 100644 --- a/data/scaffolds/provider_worker.py +++ b/data/scaffolds/provider_worker.py @@ -99,7 +99,10 @@ def _inventory(name): setups, verifiers = {}, {} if sample is not None: _, setups, verifiers = registry.load_tasks(name, task_ids=[sample['task_id']]) - declared = getattr(getattr(module, '_SPEC', None), 'VERIFIER_KIND', None) + declared = getattr(module, 'VERIFIER_KIND', None) + for owner in ('_SPEC', '_spec', 'ADAPTER'): + if declared is None: + declared = getattr(getattr(module, owner, None), 'VERIFIER_KIND', None) kinds = ([declared] if declared else sorted({v['kind'] for v in getattr(module, 'VERIFY', {}).values()})) return {'metadata': metadata, 'tasks': len(tasks), 'sample': sample, 'env_kinds': sorted({env.get('kind', 'files') for env in (envs or {}).values()}), @@ -160,7 +163,8 @@ def dispatch(request: dict) -> dict: if binding is None or binding.get("dataset") != execution_name: raise ValueError("provider requires the selected execution artifact") bindings = {} if binding is None else {binding["dataset"]: binding} - with registry._local_provider(), dataset_runtime.execution_snapshot(bindings): + from data.scaffolds.declarations import scope + with registry._local_provider(), dataset_runtime.execution_snapshot(bindings), scope(): if operation == 'inventory': return _inventory(name) if operation == 'control': diff --git a/data/scaffolds/public_native.py b/data/scaffolds/public_native.py index 6d8ac074..13499cd1 100644 --- a/data/scaffolds/public_native.py +++ b/data/scaffolds/public_native.py @@ -4,6 +4,7 @@ import hashlib import json from pathlib import Path, PurePosixPath +from data.scaffolds.scorer_input import source_digest ROOT = Path(__file__).resolve().parents[2] PROFILE = 'public-native-local-v1' @@ -30,12 +31,10 @@ def bind(dataset: str, spec: dict) -> dict: source = Path(entry['from']).resolve(strict=True) if not source.is_relative_to(ROOT) or not source.is_file(): raise ValueError('public scorer code must come from the installed platform') - content = source.read_bytes() - value = {'source': source.relative_to(ROOT).as_posix()} + value = {'source': source.relative_to(ROOT).as_posix(), 'sha256': source_digest(source)} else: content = entry['content'].encode('utf-8') - value = {'content': entry['content']} - value['sha256'] = hashlib.sha256(content).hexdigest() + value = {'content': entry['content'], 'sha256': hashlib.sha256(content).hexdigest()} if name in sources and sources[name] != value: raise ValueError('conflicting public scorer code entries') sources[name] = value diff --git a/data/scaffolds/registry.py b/data/scaffolds/registry.py index a813bea7..936921dd 100644 --- a/data/scaffolds/registry.py +++ b/data/scaffolds/registry.py @@ -810,6 +810,15 @@ def _selected_tasks(name: str, tasks: list[dict], task_ids) -> list[dict]: return [task for task in tasks if task["task_id"] in chosen] +@contextmanager +def _declaration_batch(): + from data.scaffolds.declarations import scope + from harnessgrad.dataset_runtime import execution_declarations + with scope(), execution_declarations(): + yield + + +@_declaration_batch() def load_envs(name: str, *, task_ids=None) -> dict[str, dict]: """Resolve selected ENV values, falling back to DEFAULT_ENV or files. @@ -1044,6 +1053,7 @@ def _check_verifier(spec, tid: str, name: str) -> None: f"'exit0_stdout'") +@_declaration_batch() def load_tasks(name: str, *, task_ids=None) -> tuple[list[dict], dict[str, dict], dict[str, dict]]: """Return selected (tasks, setups, verifiers) as a list and ordinary dicts. diff --git a/data/scaffolds/scorer_input.py b/data/scaffolds/scorer_input.py index fad48e8e..518f95e5 100644 --- a/data/scaffolds/scorer_input.py +++ b/data/scaffolds/scorer_input.py @@ -15,7 +15,6 @@ import json import sys import tempfile -from functools import lru_cache from pathlib import Path @@ -23,9 +22,7 @@ def inventory_digest(files: dict) -> str: return hashlib.sha256(json.dumps(files, sort_keys=True, separators=(",", ":")).encode()).hexdigest() -@lru_cache(maxsize=128) -def _file_digest(path: str, stat_identity: tuple) -> str: - # Reuse source identities while building many specs for the same static DB/JAR. +def _file_digest(path: str) -> str: # The consuming check always hashes its own private copy without this cache. digest = hashlib.sha256() with Path(path).open("rb") as stream: @@ -34,6 +31,21 @@ def _file_digest(path: str, stat_identity: tuple) -> str: return digest.hexdigest() +def source_digest(path: Path) -> str: + """Reuse source hashes only within one declaration snapshot. + + Metadata is part of the key, but not a process-wide trust decision: filesystems + may have coarse timestamps. Every new scope and each consuming checker reads + the bytes again, even if a caller restored the file's modification time. + """ + from data.scaffolds.declarations import once + path = Path(path).resolve(strict=True) + stat = path.stat() + key = ('scorer-source', str(path), stat.st_dev, stat.st_ino, stat.st_size, + stat.st_mtime_ns, stat.st_ctime_ns) + return once(key, lambda: _file_digest(str(path))) + + def bind(spec: dict, task_id: str, paths: list[str]) -> dict: entries = {e["path"]: e for e in spec["inputs"]} files = {} @@ -42,10 +54,7 @@ def bind(spec: dict, task_id: str, paths: list[str]) -> dict: if "content" in entry: files[path] = hashlib.sha256(entry["content"].encode("utf-8")).hexdigest() else: - source = Path(entry["from"]) - stat = source.stat() - files[path] = _file_digest(str(source), (stat.st_dev, stat.st_ino, stat.st_size, - stat.st_mtime_ns, stat.st_ctime_ns)) + files[path] = source_digest(Path(entry["from"])) digest = inventory_digest(files) spec["scorer_input"] = {"task_id": task_id, "path": "native-inputs", "sha256": digest, "files": files} diff --git a/data/scaffolds/spreadsheet_runtime/metadata.py b/data/scaffolds/spreadsheet_runtime/metadata.py new file mode 100644 index 00000000..a69a1513 --- /dev/null +++ b/data/scaffolds/spreadsheet_runtime/metadata.py @@ -0,0 +1,12 @@ +"""Reuse native inventories only during declaration planning, never checker launch.""" +from copy import deepcopy +import os + +from data.scaffolds.declarations import once +from . import native + + +def selected_native(): + manifest = os.environ.get('HG_SPREADSHEET_NATIVE_MANIFEST') + return deepcopy(once(('spreadsheet-native', manifest), + lambda: native.snapshot_native_python(native.require_runtime()))) diff --git a/docs/dataset-runtime-interfaces.md b/docs/dataset-runtime-interfaces.md index d676ac09..73bbf48f 100644 --- a/docs/dataset-runtime-interfaces.md +++ b/docs/dataset-runtime-interfaces.md @@ -392,6 +392,14 @@ provider 和 execution 共用存储机制，但声明、索引和制品按 profi 结构检查；启动前的 `admit_binding` 重读绑定指定的 manifest，核验 venv 内容、 基础解释器、声明、精确依赖和路径，不读取最新索引换环境，不安装或静默回退。 +声明解析与实际启动是两个独立阶段。`registry.load_tasks/load_envs` 在有界的声明 scope 中， +同一个执行环境只做一次完整库存校验和解释器探测；provider worker 的 `execution_snapshot` +在进入时校验一次，逐任务查询仅复制该绑定。scope 结束后不保留“已校验”权限。 +ALFWorld、Jericho、ScienceWorld 和 DS-1000 的任务声明按键生成，枚举键不生成 setup/verifier。 +公共评分代码、JAR、ROM 等共享源摘要，以及 native research 清单/索引和表格 native 库存， +只在声明快照中复用。新的解析、实际启动与评分输入消费仍重新读取、校验相关字节。 +不会仅凭文件时间戳把旧校验结果永久沿用到下一次运行。 + 服务在 `SETUP.tool_relay.python_runtime` 放入 service 绑定，`python` 必须与绑定一致， 并通过 `read_paths` 显式列出可信服务需要的代码和输入。平台以 `-I -B` 启动， 只读挂载运行时与白名单路径，使用每任务私有的 HOME、临时目录、缓存、配置目录和 cwd， @@ -794,7 +802,10 @@ provider 是可信适配器的依赖执行进程，不是额外的安全沙箱 当前 QA 后端不接收模型凭据或模型配置。运行时库存覆盖 venv 内容和基础解释器文件， 不声称封闭宿主系统库或完整标准库。provider 回执含实际运行时身份。driver 在一次 有界 loading session 中首次选择环境，随后请求核验同一个 manifest，不随 ready 索引 -变化重新选择。实际加载制品通过第 9.1 节接入 H0、结果缓存和恢复运行比较。 +变化重新选择。该 loading session 中相同存储配置的 catalog 复用已验证结果的副本， +避免仅为读取 split/oracle 等元数据再次启动 provider；非冻结客户端仍保持逐次读取。 +实际 resolve/control 请求仍在启动 worker 前重新校验。实际加载制品通过第 9.1 节接入 +H0、结果缓存和恢复运行比较。 ## 9. 可选评分身份 diff --git a/harnessgrad/dataset_provider.py b/harnessgrad/dataset_provider.py index 35edb0d3..34024055 100644 --- a/harnessgrad/dataset_provider.py +++ b/harnessgrad/dataset_provider.py @@ -6,6 +6,8 @@ """ from __future__ import annotations +from copy import deepcopy + import json import math import os @@ -96,6 +98,7 @@ def __init__(self, name: str, *, timeout_s: float | None = None, freeze_runtime= self.freeze_runtime = freeze_runtime self.runtime = None self.execution_runtime = None + self._catalog_cache = {} def _request(self, operation: str, *, task_ids=None, config_snapshot=None) -> dict: configuration = _configuration(self.name, config_snapshot) @@ -176,10 +179,18 @@ def _request(self, operation: str, *, task_ids=None, config_snapshot=None) -> di def catalog(self, *, config_snapshot: dict | None = None) -> dict: """Return public tasks separately from the trusted fallback answer map.""" + key = None + if getattr(self, 'freeze_runtime', False): + configuration = _configuration(self.name, config_snapshot) + key = tuple(sorted(configuration.items())) + if key in self._catalog_cache: + return deepcopy(self._catalog_cache[key]) result = self._request("catalog", config_snapshot=config_snapshot) if (not isinstance(result.get("tasks"), list) or not isinstance(result.get("scorable"), dict) or not isinstance(result.get("metadata"), dict)): raise DatasetProviderError(self.name, "catalog", "malformed catalog", error_type="ProtocolError") + if key is not None: + self._catalog_cache[key] = deepcopy(result) return result def resolve(self, task_ids=None, config_snapshot: dict | None = None) -> dict: diff --git a/harnessgrad/dataset_runtime.py b/harnessgrad/dataset_runtime.py index 69f6a555..746d230b 100644 --- a/harnessgrad/dataset_runtime.py +++ b/harnessgrad/dataset_runtime.py @@ -10,6 +10,7 @@ from contextlib import contextmanager from contextvars import ContextVar +from copy import deepcopy import hashlib import json @@ -41,6 +42,7 @@ RECIPE_VERSION = 1 _NAME = re.compile(r"[a-z][a-z0-9_]*\Z") _EXECUTION_SNAPSHOT = ContextVar("dataset_execution_snapshot", default=None) +_DECLARATION_RUNTIMES = ContextVar("dataset_declaration_runtimes", default=None) _PIN = re.compile(r"([A-Za-z0-9][A-Za-z0-9._-]*)==([A-Za-z0-9][A-Za-z0-9.!+_-]*)\Z") @@ -405,8 +407,29 @@ def resolve_execution(name: str, *, root=None) -> dict: if snapshot is not None: if name not in snapshot: raise _unavailable(name, "execution artifact was not selected by the provider client", execution=True) - return admit_binding(snapshot[name], "service") - return _resolve_profile(name, execution_descriptor, root, execution=True) + # The provider scope admitted this exact binding at entry. Declaration + # lookup is metadata, not a launch; launchers still call admit_binding. + return deepcopy(snapshot[name]) + cache = _DECLARATION_RUNTIMES.get() + if cache is None: + return _resolve_profile(name, execution_descriptor, root, execution=True) + key = (name, str(_store(root))) + if key not in cache: + cache[key] = _resolve_profile(name, execution_descriptor, root, execution=True) + return deepcopy(cache[key]) + + +@contextmanager +def execution_declarations(): + """Verify each execution artifact once while constructing task declarations.""" + if _DECLARATION_RUNTIMES.get() is not None: + yield + return + token = _DECLARATION_RUNTIMES.set({}) + try: + yield + finally: + _DECLARATION_RUNTIMES.reset(token) @contextmanager @@ -417,8 +440,8 @@ def execution_snapshot(bindings): validated = validate_binding(binding, "service") if validated["dataset"] != name: raise ValueError("execution snapshot dataset mismatch") - admit_binding(validated, "service") - selected[name] = validated + admitted = admit_binding(validated, "service") + selected[name] = deepcopy(admitted) token = _EXECUTION_SNAPSHOT.set(selected) try: yield diff --git a/tests/quality/test_dataset_declaration_performance.py b/tests/quality/test_dataset_declaration_performance.py new file mode 100644 index 00000000..4b1e0dcb --- /dev/null +++ b/tests/quality/test_dataset_declaration_performance.py @@ -0,0 +1,201 @@ +"""Count heavyweight operations with authored pools; no benchmark data or models.""" +import ast +import hashlib +import json +import os +from pathlib import Path +from types import ModuleType, SimpleNamespace +from unittest.mock import Mock + +import pytest + +from data.scaffolds import declarations, registry, scorer_input +from harnessgrad import dataset_runtime as runtime +from test_dataset_execution_runtime import execution +from test_dataset_runtime import environment + +ROOT = Path(__file__).resolve().parents[2] + + +def test_alfworld_import_does_not_resolve_runtime_for_any_of_134_tasks(monkeypatch): + source = ROOT/'data/datasets/alfworld.py' + tree = ast.parse(source.read_bytes()) + for node in tree.body: + if isinstance(node, ast.FunctionDef) and node.name in ('_collect','_study_ids'): + node.body = [ast.Return(ast.Name('_AUTHORED', ast.Load()) if node.name=='_collect' else ast.List([],ast.Load()))] + ast.fix_missing_locations(tree) + module = ModuleType('authored_alfworld_performance'); module.__file__=str(source) + module._AUTHORED = [{'task_id':f'authored/{i}','stripped_game':'{}'} for i in range(134)] + resolve=Mock(return_value={'python':'/authored/python'}) + monkeypatch.setattr(runtime,'resolve_execution',resolve) + monkeypatch.setattr(runtime,'bind',lambda selected,role:{**selected,'role':role}) + monkeypatch.delenv('HG_ALFWORLD_POOL',raising=False) + exec(compile(tree,str(source),'exec'),module.__dict__) + assert module._LOAD_ERROR is None + assert len(module.SETUP)==len(module.VERIFY)==134 + assert list(module.SETUP)==[row['task_id'] for row in module._AUTHORED] + assert 'authored/1' in module.SETUP and module.SETUP.get('absent') is None + assert resolve.call_count==0 + for key in ('authored/1','authored/5'): + assert module.SETUP[key]['tool_relay']['python']=='/authored/python' + assert resolve.call_count==2 # Only the two requested task values were constructed. + + +def test_lazy_map_lookup_is_fresh_and_keys_never_build(): + build=Mock(side_effect=lambda record:{'files':[]}) + selected=declarations.TaskDeclarations([{'task_id':'one'},{'task_id':'two'}],build) + assert len(selected)==2 and list(selected)==['one','two'] and 'one' in selected + assert selected.get('absent') is None and not build.called + selected['one']['files'].append('caller mutation') + assert selected['one']['files']==[] + assert build.call_count==2 + + +def test_execution_snapshot_checks_once_but_launch_admission_stays_fresh(execution,monkeypatch): + _,_,store,_=execution + artifact=runtime.prepare_execution('control',root=store);binding=runtime.bind(artifact,'service') + verify=Mock(wraps=runtime._verify_manifest);monkeypatch.setattr(runtime,'_verify_manifest',verify) + with runtime.execution_snapshot({'control':binding}): + for _ in range(134): + selected=runtime.resolve_execution('control') + assert selected['descriptor_sha256']==artifact['descriptor_sha256'] + assert 'role' not in selected + selected['runtime_paths'].append('/caller-only') + assert verify.call_count==1 + assert '/caller-only' not in runtime.resolve_execution('control')['runtime_paths'] + marker=next(Path(artifact['path']).rglob('runtime_marker.py')) + marker.chmod(0o644);marker.write_text('changed after metadata admission') + with pytest.raises(runtime.DatasetRuntimeUnavailable,match='inventory changed'): + runtime.admit_binding(binding,'service') + with pytest.raises(runtime.DatasetRuntimeUnavailable): + with runtime.execution_snapshot({'control':binding}): pass + + +def test_runtime_cache_is_scoped_and_not_a_process_wide_permission(execution,monkeypatch): + _,_,store,_=execution;runtime.prepare_execution('control',root=store) + verify=Mock(wraps=runtime._verify_manifest);monkeypatch.setattr(runtime,'_verify_manifest',verify) + with runtime.execution_declarations(): + for _ in range(134): runtime.resolve_execution('control',root=store) + assert verify.call_count==1 + with runtime.execution_declarations():runtime.resolve_execution('control',root=store) + assert verify.call_count==2 + runtime.resolve_execution('control',root=store) + assert verify.call_count==3 + + +def test_native_manifest_and_record_index_reuse_is_metadata_scoped(tmp_path,monkeypatch): + from data.scaffolds.native_research.adapter import Adapter,COMMIT + folder=tmp_path/'native';folder.mkdir() + path=folder/'manifest.json' + path.write_text(json.dumps({'schema':'harnessgrad.native-research.v1','dataset':'browsecomp', + 'upstream_commit':COMMIT,'source':{},'records':[{'task_id':f'browsecomp/{i}'} for i in range(134)],'files':{}})) + adapter=Adapter('browsecomp',root=folder) + read=Mock(wraps=adapter._read_manifest);monkeypatch.setattr(adapter,'_read_manifest',read) + with declarations.scope(): + for _ in range(134): assert len(adapter.records())==134 + assert read.call_count==1 + path.write_text('{}') + with declarations.scope(): + with pytest.raises(ValueError,match='manifest'):adapter.records() + assert read.call_count==2 + + +def test_native_sheet_inventory_is_reused_only_for_metadata(monkeypatch): + from data.scaffolds.spreadsheet_runtime import metadata,native + inspect=Mock(return_value={'files':{},'deployment':'authored'}) + monkeypatch.setattr(native,'require_runtime',inspect) + with declarations.scope(): + first=metadata.selected_native();first['files']['caller-only']='value' + for _ in range(134):assert metadata.selected_native()['files']=={} + assert inspect.call_count==1 + metadata.selected_native();assert inspect.call_count==2 + + +def test_source_digest_cache_invalidates_when_mtime_is_restored(tmp_path,monkeypatch): + source=tmp_path/'source';source.write_bytes(b'first') + old=source.stat();first=scorer_input.source_digest(source) + source.write_bytes(b'other');os.utime(source,ns=(old.st_atime_ns,old.st_mtime_ns)) + assert scorer_input.source_digest(source)==hashlib.sha256(b'other').hexdigest()!=first + + +def test_provider_inventory_uses_kind_contract_without_traversing_values(monkeypatch): + from data.scaffolds import provider_worker as worker + task={'task_id':'authored/one','goal':'public'} + module=SimpleNamespace(KIND='dataset',VERIFIER_KIND='command') + class NoTraversal: + def values(self):raise AssertionError('inventory traversed the full verifier pool') + module.VERIFY=NoTraversal() + monkeypatch.setattr(worker.importlib,'import_module',lambda name:module) + monkeypatch.setattr(registry,'load_split',lambda name:([task],{},None)) + monkeypatch.setattr(registry,'load_envs',lambda name:{task['task_id']:{'kind':'files'}}) + monkeypatch.setattr(registry,'load_tasks',lambda *args,**kwargs:([task],{},{task['task_id']:{'kind':'command'}})) + result=worker._inventory('authored') + assert result['verifier_kinds']==['command'] and result['tasks']==1 + + +def test_provider_catalog_is_cached_only_in_a_frozen_loading_session(tmp_path,monkeypatch): + from harnessgrad import dataset_provider as client + monkeypatch.setattr(runtime,'descriptor',lambda name:{'provider':'adapter','descriptor_sha256':'authored'}) + monkeypatch.setenv('HG_DATA_AUTHORED',str(tmp_path/'first')) + provider=client.DatasetProvider('authored',freeze_runtime=True) + request=Mock(return_value={'tasks':[{'task_id':'one'}],'scorable':{},'metadata':{}}) + monkeypatch.setattr(provider,'_request',request) + provider.catalog()['tasks'].clear() + assert provider.catalog()['tasks']==[{'task_id':'one'}] and request.call_count==1 + monkeypatch.setenv('HG_DATA_AUTHORED',str(tmp_path/'second')) + provider.catalog();assert request.call_count==2 + fresh=client.DatasetProvider('authored');monkeypatch.setattr(fresh,'_request',request) + fresh.catalog();fresh.catalog();assert request.call_count==4 + + +@pytest.mark.parametrize('dataset',['jericho','scienceworld']) +def test_interactive_mapping_keys_do_not_prepare_unselected_tasks(dataset,monkeypatch): + import importlib + module=importlib.import_module('data.datasets.'+dataset) + records=[{'task_id':f'authored/{i}'} for i in range(134)] + runtime_lookup=Mock(side_effect=AssertionError('mapping enumeration selected an engine')) + monkeypatch.setattr(runtime,'resolve_execution',runtime_lookup) + build=Mock(return_value={'files':[]}) + if dataset=='jericho': + monkeypatch.setitem(module.__dict__,'_RECORDS',records) + mapping=module._map(build) + else: + monkeypatch.setattr(module,'_ensure',lambda:records) + monkeypatch.setattr(module,'_setup',build) + mapping=module._Lazy().get('SETUP') + assert len(mapping)==134 and list(mapping)==[row['task_id'] for row in records] + assert 'authored/3' in mapping and not build.called and not runtime_lookup.called + mapping['authored/3'] + build.assert_called_once_with(records[3]) + + +def test_repeated_source_reads_are_bounded_by_declaration_scope(tmp_path,monkeypatch): + source=tmp_path/'code.py';source.write_text('authored source') + read=Mock(wraps=scorer_input._file_digest);monkeypatch.setattr(scorer_input,'_file_digest',read) + with declarations.scope(): + for _ in range(134):scorer_input.source_digest(source) + assert read.call_count==1 + with declarations.scope():scorer_input.source_digest(source) + assert read.call_count==2 + + +def test_travelplanner_checks_world_once_per_task_and_detects_mid_batch_change(tmp_path,monkeypatch): + from data.scaffolds.native_research.adapter import Adapter,COMMIT + from data.scaffolds.native_research import component_runtime + folder=tmp_path/'native';(folder/'database').mkdir(parents=True) + world=folder/'database/authored.txt';world.write_text('authored world') + case=folder/'case.json';case.write_text(json.dumps({'task_id':'travelplanner/one','question':'authored'})) + manifest={'schema':'harnessgrad.native-research.v1','dataset':'travelplanner','upstream_commit':COMMIT, + 'source':{},'judge_model':'authored-no-calls','records':[{'task_id':'travelplanner/one','case':'case.json'}], + 'files':{'case.json':hashlib.sha256(case.read_bytes()).hexdigest(), + 'database/authored.txt':hashlib.sha256(world.read_bytes()).hexdigest()}} + (folder/'manifest.json').write_text(json.dumps(manifest)) + monkeypatch.setattr(component_runtime,'binding',lambda *a:{'python':'/authored/python','runtime_paths':[]}) + adapter=Adapter('travelplanner',root=folder) + check=Mock(wraps=adapter.checked_world);monkeypatch.setattr(adapter,'checked_world',check) + with declarations.scope(): + spec=adapter.verifier('travelplanner/one') + assert spec['kind']=='command' and check.call_count==1 + world.write_text('changed after metadata selection') + with pytest.raises(ValueError,match='world/database'): + adapter.verifier('travelplanner/one')