PR 06: Add builtin sample environments

ergon_builtins/ergon_builtins/benchmarks/gdpeval/benchmark.py

@@ -5,7 +5,7 @@
 :class:`Benchmark` interface.
 """
 
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from typing import Any, ClassVar
 
 from ergon_core.api import Benchmark, BenchmarkRequirements, Task
@@ -92,22 +92,7 @@ def build_instances(self) -> Mapping[str, Sequence[Task[GDPTaskConfig]]]:
         All tasks land in a single ``"default"`` instance since there is
         no multi-instance structure in the GDP dataset.
         """
-        tasks: list[Task[GDPTaskConfig]] = []
-        for payload in self._load_task_configs():
-            description = extract_task_description(payload.task_id, repo_id=self.dataset_repo)
-            tasks.append(
-                GDPEvalTask(
-                    task_slug=payload.task_id,
-                    instance_key="default",
-                    description=description,
-                    task_payload=payload,
-                    worker=self._worker_factory(),
-                    sandbox=self._sandbox_factory(),
-                    evaluators=(self._evaluator_factory(),),
-                )
-            )
-
-        return {"default": tasks}
+        return {"default": [sample.tasks[0] for sample in self._environment().all_samples()]}
 
     def evaluator_requirements(self) -> Sequence[str]:
         return ()
@@ -130,3 +115,29 @@ def _load_task_configs(self) -> list[GDPTaskConfig]:
                 )
             )
         return configs
+
+    # Compatibility loader surface for GDPEvalEnvironment. PR11 deletes
+    # benchmark-centered authoring and these wrappers.
+    def iter_rows(self) -> Iterator[GDPTaskConfig]:
+        yield from self._load_task_configs()
+
+    def load_rows(self) -> Sequence[GDPTaskConfig]:
+        return self._load_task_configs()
+
+    def _environment(self) -> Any:
+        from ergon_builtins.environments.gdpeval import GDPEvalEnvironment
+
+        return GDPEvalEnvironment(
+            name=self.name,
+            dataset_repo=self.dataset_repo,
+            split=self.split,
+            limit=self.limit,
+            loader=self,
+            task_description=lambda row: extract_task_description(
+                row.task_id,
+                repo_id=self.dataset_repo,
+            ),
+            worker=lambda _row: self._worker_factory(),
+            sandbox=lambda _row: self._sandbox_factory(),
+            evaluators=lambda _row: (self._evaluator_factory(),),
+        )

ergon_builtins/ergon_builtins/benchmarks/minif2f/benchmark.py

@@ -7,7 +7,7 @@
 
 import json
 import logging
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from pathlib import Path
 from typing import Any, ClassVar
 
@@ -76,33 +76,7 @@ def __init__(
     # ------------------------------------------------------------------
 
     def build_instances(self) -> Mapping[str, Sequence[Task[MiniF2FTaskPayload]]]:
-        problems = self._load_problems()
-        tasks: list[Task[MiniF2FTaskPayload]] = []
-        for problem in problems:
-            payload = MiniF2FTaskPayload(
-                name=problem.name,
-                informal_statement=problem.informal_statement,
-                formal_statement=problem.formal_statement,
-                header=problem.header,
-            )
-            description = (
-                f"{problem.informal_statement}\n\n"
-                f"Your task: prove the following theorem in Lean 4.\n\n"
-                f"{problem.header}\n"
-                f"{problem.formal_statement}"
-            )
-            tasks.append(
-                MiniF2FTask(
-                    task_slug=problem.name,
-                    instance_key="default",
-                    description=description,
-                    task_payload=payload,
-                    worker=self._worker_factory(),
-                    sandbox=self._sandbox_factory(),
-                    evaluators=(self._evaluator_factory(),),
-                )
-            )
-        return {"default": tasks}
+        return {"default": [sample.tasks[0] for sample in self._environment().all_samples()]}
 
     def evaluator_requirements(self) -> Sequence[str]:
         return ()
@@ -144,3 +118,24 @@ def _load_problems(self) -> list[MiniF2FProblem]:
 
         logger.info("Loaded %d MiniF2F-v2c problems", len(problems))
         return problems
+
+    # Compatibility loader surface for MiniF2FEnvironment. PR11 deletes
+    # benchmark-centered authoring and these wrappers.
+    def iter_rows(self) -> Iterator[MiniF2FProblem]:
+        yield from self._load_problems()
+
+    def load_rows(self) -> Sequence[MiniF2FProblem]:
+        return self._load_problems()
+
+    def _environment(self) -> Any:
+        from ergon_builtins.environments.minif2f import MiniF2FEnvironment
+
+        return MiniF2FEnvironment(
+            name=self.name,
+            limit=self.limit,
+            data_dir=self.data_dir,
+            loader=self,
+            worker=lambda _row: self._worker_factory(),
+            sandbox=lambda _row: self._sandbox_factory(),
+            evaluators=lambda _row: (self._evaluator_factory(),),
+        )

ergon_builtins/ergon_builtins/benchmarks/researchrubrics/benchmark.py

@@ -4,7 +4,7 @@
 whether agents know when and what to ask stakeholders.
 """
 
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from typing import Any, ClassVar
 
 from datasets import load_dataset
@@ -15,7 +15,6 @@
 from ergon_core.core.shared.settings import settings
 
 from ergon_builtins.benchmarks.researchrubrics.sandbox import ResearchE2BSandbox
-from ergon_builtins.benchmarks.researchrubrics.rubric import ResearchRubricsRubric
 from ergon_builtins.benchmarks.researchrubrics.task_schemas import (
     ResearchRubricsTaskPayload,
     RubricCriterion,
@@ -83,28 +82,7 @@ def __init__(
     # ------------------------------------------------------------------
 
     def build_instances(self) -> Mapping[str, Sequence[Task[ResearchRubricsTaskPayload]]]:
-        payloads = self._load_rows()
-        tasks: list[Task[ResearchRubricsTaskPayload]] = []
-        for payload in payloads:
-            evaluator = self._evaluator_factory()
-            if isinstance(evaluator, ResearchRubricsRubric) and not evaluator.rubric_criteria:
-                evaluator = ResearchRubricsRubric(
-                    name=evaluator.name,
-                    metadata=evaluator.metadata,
-                    rubric_criteria=tuple(payload.rubrics),
-                )
-            tasks.append(
-                ResearchRubricsTask(
-                    task_slug=payload.sample_id,
-                    instance_key="default",
-                    description=payload.prompt,
-                    task_payload=payload,
-                    worker=self._worker_factory(),
-                    sandbox=self._sandbox_factory(),
-                    evaluators=(evaluator,),
-                )
-            )
-        return {"default": tasks}
+        return {"default": [sample.tasks[0] for sample in self._environment().all_samples()]}
 
     def evaluator_requirements(self) -> Sequence[str]:
         return ()
@@ -125,6 +103,27 @@ def _load_rows(self) -> list[ResearchRubricsTaskPayload]:
 
         return [_payload_from_row(train_ds[idx]) for idx in range(len(train_ds))]
 
+    # Compatibility loader surface for ResearchRubricsEnvironment. PR11
+    # deletes benchmark-centered authoring and these wrappers.
+    def iter_rows(self) -> Iterator[ResearchRubricsTaskPayload]:
+        yield from self._load_rows()
+
+    def load_rows(self) -> Sequence[ResearchRubricsTaskPayload]:
+        return self._load_rows()
+
+    def _environment(self) -> Any:
+        from ergon_builtins.environments.researchrubrics import ResearchRubricsEnvironment
+
+        return ResearchRubricsEnvironment(
+            name=self.name,
+            dataset_name=self.dataset_name,
+            limit=self.limit,
+            loader=self,
+            worker=lambda _row: self._worker_factory(),
+            sandbox=lambda _row: self._sandbox_factory(),
+            evaluators=lambda _row: (self._evaluator_factory(),),
+        )
+
 
 def _payload_from_row(
     row: Mapping[str, Any],  # slopcop: ignore[no-typing-any]

ergon_builtins/ergon_builtins/benchmarks/swebench_verified/benchmark.py

@@ -6,7 +6,7 @@
 """
 
 import logging
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from typing import Any, ClassVar
 
 from datasets import load_dataset
@@ -77,27 +77,33 @@ def __init__(
         self._evaluator_factory = evaluator_factory
 
     def build_instances(self) -> Mapping[str, Sequence[Task[SWEBenchTaskPayload]]]:
-        instances = _load_rows(limit=self.limit)
-        tasks: list[Task[SWEBenchTaskPayload]] = []
-        for instance in instances:
-            payload = SWEBenchTaskPayload.from_instance(instance)
-            tasks.append(
-                SweBenchTask(
-                    task_slug=instance.instance_id,
-                    instance_key="default",
-                    description=payload.build_worker_description(),
-                    task_payload=payload,
-                    worker=self._worker_factory(),
-                    sandbox=self._sandbox_factory(),
-                    evaluators=(self._evaluator_factory(),),
-                )
-            )
-        logger.info("Loaded %d SWE-Bench Verified instances", len(tasks))
-        return {"default": tasks}
+        samples = self._environment().all_samples()
+        logger.info("Loaded %d SWE-Bench Verified instances", len(samples))
+        return {"default": [sample.tasks[0] for sample in samples]}
 
     def evaluator_requirements(self) -> Sequence[str]:
         return ()
 
+    # Compatibility loader surface for SweBenchVerifiedEnvironment. PR11
+    # deletes benchmark-centered authoring and these wrappers.
+    def iter_rows(self) -> Iterator[SWEBenchInstance]:
+        yield from _load_rows(limit=self.limit)
+
+    def load_rows(self) -> Sequence[SWEBenchInstance]:
+        return _load_rows(limit=self.limit)
+
+    def _environment(self) -> Any:
+        from ergon_builtins.environments.swebench_verified import SweBenchVerifiedEnvironment
+
+        return SweBenchVerifiedEnvironment(
+            name=self.name,
+            limit=self.limit,
+            loader=self,
+            worker=lambda _row: self._worker_factory(),
+            sandbox=lambda _row: self._sandbox_factory(),
+            evaluators=lambda _row: (self._evaluator_factory(),),
+        )
+
 
 def _load_rows(*, limit: int | None = None) -> list[SWEBenchInstance]:
     """Load and validate SWE-Bench instances from HuggingFace."""

ergon_builtins/ergon_builtins/environments/__init__.py

@@ -0,0 +1,13 @@
+"""Builtin sample-producing environments."""
+
+from ergon_builtins.environments.gdpeval import GDPEvalEnvironment
+from ergon_builtins.environments.minif2f import MiniF2FEnvironment
+from ergon_builtins.environments.researchrubrics import ResearchRubricsEnvironment
+from ergon_builtins.environments.swebench_verified import SweBenchVerifiedEnvironment
+
+__all__ = [
+    "GDPEvalEnvironment",
+    "MiniF2FEnvironment",
+    "ResearchRubricsEnvironment",
+    "SweBenchVerifiedEnvironment",
+]

ergon_builtins/ergon_builtins/environments/_resolve.py

@@ -0,0 +1,33 @@
+"""Typed component resolution helpers for builtin environments."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+from typing import TypeVar, cast
+
+from ergon_core.api.rubric import Evaluator
+from ergon_core.api.sandbox import Sandbox
+from ergon_core.api.worker import Worker
+
+RowT = TypeVar("RowT")
+
+
+def resolve_worker(value: Worker | Callable[[RowT], Worker], row: RowT) -> Worker:
+    if isinstance(value, Worker):
+        return value
+    return cast(Callable[[RowT], Worker], value)(row)
+
+
+def resolve_sandbox(value: Sandbox | Callable[[RowT], Sandbox], row: RowT) -> Sandbox:
+    if isinstance(value, Sandbox):
+        return value
+    return cast(Callable[[RowT], Sandbox], value)(row)
+
+
+def resolve_evaluators(
+    value: Sequence[Evaluator] | Callable[[RowT], Sequence[Evaluator]],
+    row: RowT,
+) -> tuple[Evaluator, ...]:
+    if callable(value):
+        return tuple(cast(Callable[[RowT], Sequence[Evaluator]], value)(row))
+    return tuple(value)