PR 07: Add sample experiment API examples

ergon_cli/ergon_cli/domains/examples/catalogue.py

@@ -9,7 +9,7 @@
     display_name="MiniF2F local llama.cpp",
     short_description=("Run three MiniF2F Lean proof tasks with a managed local llama.cpp server."),
     purpose=("Run three MiniF2F Lean proof tasks with local llama.cpp and an E2B Lean sandbox."),
-    script_path="examples/getting_started/01_minif2f_local_llamacpp/run.py",
+    script_path="examples/getting_started/01_minif2f_local_llamacpp/submit.py",
     prerequisites=(
         "E2B_API_KEY is configured in Ergon's .env file or process environment.",
         "A local GGUF model path or Hugging Face GGUF ref is available for llama.cpp.",
@@ -77,8 +77,63 @@
     ),
 )
 
+EXPERIMENT_API_MINIF2F = ExampleDefinition(
+    slug="experiment-api-minif2f",
+    display_name="Experiment API MiniF2F",
+    short_description="Submit a single MiniF2F environment through the experiment API.",
+    purpose="Demonstrate the smallest complete Environment + Experiment submit flow.",
+    script_path="examples/experiment_api/01_minif2f_single_environment/submit.py",
+    prerequisites=("Database settings are configured.", "Provider credentials are available."),
+    options=(),
+)
+
+EXPERIMENT_API_MIXED_ENVIRONMENTS = ExampleDefinition(
+    slug="experiment-api-mixed-environments",
+    display_name="Experiment API mixed environments",
+    short_description="Submit one experiment spanning several builtin environments.",
+    purpose="Demonstrate heterogeneous environment composition for training-style launches.",
+    script_path="examples/experiment_api/02_mixed_environment_training/submit.py",
+    prerequisites=("Database settings are configured.", "Provider credentials are available."),
+    options=(),
+)
+
+EXPERIMENT_API_STREAMING = ExampleDefinition(
+    slug="experiment-api-streaming",
+    display_name="Experiment API streaming",
+    short_description="Submit from a streaming SWE-bench candidate buffer.",
+    purpose="Demonstrate streamed source buffering with a larger candidate pool.",
+    script_path="examples/experiment_api/03_streaming_hf_dataset/submit.py",
+    prerequisites=("Database settings are configured.", "SWE-bench data access is available."),
+    options=(),
+)
+
+EXPERIMENT_API_CURRICULUM_SAMPLER = ExampleDefinition(
+    slug="experiment-api-curriculum-sampler",
+    display_name="Experiment API curriculum sampler",
+    short_description="Submit with a custom sampler that orders a retained candidate pool.",
+    purpose="Demonstrate sampler customization without changing environment code.",
+    script_path="examples/experiment_api/04_curriculum_sampler/submit.py",
+    prerequisites=("Database settings are configured.", "Provider credentials are available."),
+    options=(),
+)
+
+EXPERIMENT_API_SWEBENCH = ExampleDefinition(
+    slug="experiment-api-swebench",
+    display_name="Experiment API SWE-bench",
+    short_description="Submit SWE-bench samples with row-dependent runtime configs.",
+    purpose="Demonstrate row-dependent worker, evaluator, and sandbox selection.",
+    script_path="examples/experiment_api/05_row_dependent_runtime_configs/submit.py",
+    prerequisites=("Database settings are configured.", "SWE-bench data access is available."),
+    options=(),
+)
+
 EXAMPLES: dict[str, ExampleDefinition] = {
     MINIF2F_LOCAL_LLAMACPP.slug: MINIF2F_LOCAL_LLAMACPP,
+    EXPERIMENT_API_MINIF2F.slug: EXPERIMENT_API_MINIF2F,
+    EXPERIMENT_API_MIXED_ENVIRONMENTS.slug: EXPERIMENT_API_MIXED_ENVIRONMENTS,
+    EXPERIMENT_API_STREAMING.slug: EXPERIMENT_API_STREAMING,
+    EXPERIMENT_API_CURRICULUM_SAMPLER.slug: EXPERIMENT_API_CURRICULUM_SAMPLER,
+    EXPERIMENT_API_SWEBENCH.slug: EXPERIMENT_API_SWEBENCH,
 }
 
 

ergon_cli/ergon_cli/domains/examples/runner.py

@@ -65,9 +65,15 @@ def _script_args(command: ExampleCommand) -> list[str]:
 
 def _repo_root(script_path: str) -> Path:
     configured = os.environ.get("ERGON_REPO_ROOT")
-    candidates: list[Path] = []
     if configured:
-        candidates.append(Path(configured).resolve())
+        candidate = Path(configured).resolve()
+        if (candidate / script_path).is_file():
+            return candidate
+        raise ExampleRunError(
+            f"ERGON_REPO_ROOT is set, but the configured checkout does not contain {script_path}."
+        )
+
+    candidates: list[Path] = []
     cwd = Path.cwd().resolve()
     candidates.append(cwd)
     candidates.extend(cwd.parents)

ergon_cli/tests/unit/cli/test_examples_cli.py

@@ -11,6 +11,7 @@
 import ergon_cli.domains.examples.preflight as examples_preflight
 import ergon_cli.domains.examples.runner as examples_runner
 from ergon_cli.domains.examples.commands import handle_examples
+from ergon_cli.domains.examples.catalogue import EXAMPLES
 from ergon_cli.main import build_parser
 from ergon_cli.domains.examples.preflight import ExampleSetupError, PreflightResult
 
@@ -68,6 +69,36 @@ def test_examples_subcommands_are_registered_in_main_parser() -> None:
     assert run_args.max_iterations == 4
 
 
+def test_experiment_api_examples_are_catalogued() -> None:
+    expected = {
+        "experiment-api-minif2f",
+        "experiment-api-swebench",
+        "experiment-api-mixed-environments",
+        "experiment-api-curriculum-sampler",
+        "experiment-api-streaming",
+    }
+
+    assert expected.issubset(EXAMPLES)
+
+
+def test_example_submission_summary_includes_sampler_invocation(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    from examples.getting_started._shared.launch import print_submission_summary
+
+    print_submission_summary(
+        experiment_id="exp-1",
+        sampler_invocation_id="sampler-1",
+        batch_id=None,
+        sample_ids=["sample-1"],
+    )
+
+    out = capsys.readouterr().out
+    assert "experiment_id" in out
+    assert "sampler_invocation_id" in out
+    assert "sample_ids" in out
+
+
 def test_examples_run_accepts_base_model_for_single_command_launch() -> None:
     args = build_parser().parse_args(
         [
@@ -160,7 +191,7 @@ def test_info_prints_prerequisites_options_and_script_path(
     assert "E2B_API_KEY" in out
     assert "--limit" in out
     assert "--base-url" in out
-    assert "examples/getting_started/01_minif2f_local_llamacpp/run.py" in out
+    assert "examples/getting_started/01_minif2f_local_llamacpp/submit.py" in out
 
 
 def test_check_runs_preflight_without_launching(
@@ -301,7 +332,7 @@ def fake_preflight(*, base_url: str) -> PreflightResult:
     assert (
         Path(command[5])
         .as_posix()
-        .endswith("examples/getting_started/01_minif2f_local_llamacpp/run.py")
+        .endswith("examples/getting_started/01_minif2f_local_llamacpp/submit.py")
     )
     assert command[6:] == [
         "--limit",
@@ -351,7 +382,7 @@ def test_run_can_use_configured_repo_root(
     tmp_path: Path,
 ) -> None:
     fake_repo = tmp_path / "repo"
-    script = fake_repo / "examples/getting_started/01_minif2f_local_llamacpp/run.py"
+    script = fake_repo / "examples/getting_started/01_minif2f_local_llamacpp/submit.py"
     script.parent.mkdir(parents=True)
     (fake_repo / "examples/pyproject.toml").write_text("[project]\nname='fake'\nversion='0'\n")
     script.write_text("print('ok')\n")

ergon_core/tests/unit/architecture/test_definition_identity_naming.py

@@ -19,6 +19,39 @@
 )
 EXCLUDED_FILES = {Path(__file__).resolve()}
 EXPERIMENT_ID_PATTERN = re.compile(r"\b(?:experiment" r"_id|experiment" r"Id)\b")
+ALLOWED_EXPERIMENT_ID_PATH_PREFIXES = (
+    ROOT / "ergon_core" / "ergon_core" / "api" / "experiment",
+    ROOT / "ergon_core" / "ergon_core" / "core" / "application" / "experiments",
+    ROOT / "ergon_core" / "ergon_core" / "core" / "persistence" / "experiments",
+    ROOT / "ergon_core" / "ergon_core" / "core" / "views" / "experiments",
+    ROOT / "ergon_core" / "ergon_core" / "core" / "views" / "samples",
+    ROOT / "ergon_core" / "tests" / "integration" / "experiments",
+    ROOT / "ergon_core" / "tests" / "unit" / "api",
+    ROOT / "ergon_core" / "tests" / "unit" / "core" / "application" / "experiments",
+    ROOT / "ergon_core" / "tests" / "unit" / "read_models",
+    ROOT / "ergon_core" / "tests" / "unit" / "rest_api",
+    ROOT / "tests" / "examples",
+)
+ALLOWED_EXPERIMENT_ID_FILES = {
+    ROOT
+    / "ergon_core"
+    / "ergon_core"
+    / "core"
+    / "infrastructure"
+    / "http"
+    / "routes"
+    / "experiments.py",
+    ROOT
+    / "ergon_core"
+    / "ergon_core"
+    / "core"
+    / "infrastructure"
+    / "http"
+    / "routes"
+    / "samples.py",
+    ROOT / "ergon_core" / "ergon_core" / "core" / "persistence" / "telemetry" / "models.py",
+    ROOT / "ergon_core" / "tests" / "unit" / "state" / "test_type_invariants.py",
+}
 ALLOWED_EXPERIMENT_ID_PATTERNS_BY_FILE = {
     ROOT / "ergon_core" / "ergon_core" / "api" / "experiment" / "experiment.py": (
         re.compile(r"experiment_id: UUID"),
@@ -90,6 +123,13 @@
 }
 
 
+def _allows_experiment_identity_path(path: Path) -> bool:
+    resolved = path.resolve()
+    if resolved in ALLOWED_EXPERIMENT_ID_FILES:
+        return True
+    return any(resolved.is_relative_to(prefix) for prefix in ALLOWED_EXPERIMENT_ID_PATH_PREFIXES)
+
+
 def test_definition_identity_uses_definition_name() -> None:
     hits: list[str] = []
     for root in DEFINITION_ROOTS:
@@ -107,6 +147,8 @@ def test_definition_identity_uses_definition_name() -> None:
                 text = path.read_text()
             except UnicodeDecodeError:
                 continue
+            if _allows_experiment_identity_path(path):
+                continue
             for line_number, line in enumerate(text.splitlines(), start=1):
                 if EXPERIMENT_ID_PATTERN.search(line):
                     allowed = ALLOWED_EXPERIMENT_ID_PATTERNS_BY_FILE.get(path.resolve(), ())

examples/experiment_api/01_minif2f_single_environment/submit.py

@@ -0,0 +1,37 @@
+"""Smallest complete MiniF2F experiment submission example."""
+
+from __future__ import annotations
+
+import asyncio
+
+from ergon_builtins.benchmarks.minif2f.sandbox import LeanSandbox
+from ergon_builtins.benchmarks.minif2f.worker_factory import (
+    make_minif2f_rubric,
+    make_minif2f_worker,
+)
+from ergon_builtins.environments import MiniF2FEnvironment
+from ergon_core.api import Experiment, RandomSampler
+from experiment_api._shared import experiment_submission_service
+
+
+async def main() -> None:
+    worker = make_minif2f_worker(model="openai:gpt-4o")
+    env = MiniF2FEnvironment(
+        name="mini-validation",
+        split="validation",
+        limit=10,
+        worker=worker,
+        evaluators=[make_minif2f_rubric()],
+        sandbox=LeanSandbox(),
+    )
+    experiment = Experiment(name="mini-validation-smoke", environments=[env])
+    result = await experiment.submit(
+        service=experiment_submission_service(),
+        k=10,
+        sampler=RandomSampler(seed=0),
+    )
+    print(result.model_dump_json(indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/experiment_api/02_mixed_environment_training/submit.py

@@ -0,0 +1,78 @@
+"""Heterogeneous training-style experiment with four builtin environments."""
+
+from __future__ import annotations
+
+import asyncio
+
+from ergon_builtins.benchmarks.gdpeval.benchmark import _default_gdpeval_sandbox
+from ergon_builtins.benchmarks.gdpeval.worker_factory import (
+    make_gdpeval_rubric,
+    make_gdpeval_worker,
+)
+from ergon_builtins.benchmarks.minif2f.sandbox import LeanSandbox
+from ergon_builtins.benchmarks.minif2f.worker_factory import make_minif2f_rubric
+from ergon_builtins.benchmarks.researchrubrics.benchmark import _default_research_sandbox
+from ergon_builtins.benchmarks.researchrubrics.worker_factory import make_research_rubric
+from ergon_builtins.benchmarks.swebench_verified.benchmark import _default_swebench_sandbox
+from ergon_builtins.benchmarks.swebench_verified.worker_factory import make_swebench_rubric
+from ergon_builtins.environments import (
+    GDPEvalEnvironment,
+    MiniF2FEnvironment,
+    ResearchRubricsEnvironment,
+    SweBenchVerifiedEnvironment,
+)
+from ergon_core.api import Experiment, RandomSampler
+from experiment_api._shared import experiment_submission_service
+
+
+async def main() -> None:
+    worker = make_gdpeval_worker(model="openai:gpt-4o")
+    experiment = Experiment(
+        name="generalist-mixed-training",
+        environments=[
+            MiniF2FEnvironment(
+                name="mini-validation",
+                split="validation",
+                limit=10,
+                worker=worker,
+                evaluators=[make_minif2f_rubric()],
+                sandbox=LeanSandbox(),
+            ),
+            ResearchRubricsEnvironment(
+                name="research-validation",
+                split="validation",
+                limit=10,
+                worker=worker,
+                evaluators=[make_research_rubric()],
+                sandbox=_default_research_sandbox(),
+            ),
+            GDPEvalEnvironment(
+                name="gdp-validation",
+                split="validation",
+                limit=10,
+                worker=worker,
+                evaluators=[make_gdpeval_rubric()],
+                sandbox=_default_gdpeval_sandbox(),
+            ),
+            SweBenchVerifiedEnvironment(
+                name="swebench-train",
+                split="train",
+                streaming=True,
+                limit=100,
+                worker=worker,
+                evaluators=[make_swebench_rubric()],
+                sandbox=_default_swebench_sandbox(),
+            ),
+        ],
+    )
+    result = await experiment.submit(
+        service=experiment_submission_service(),
+        k=32,
+        candidate_pool_size=128,
+        sampler=RandomSampler(seed=0),
+    )
+    print(result.model_dump_json(indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/experiment_api/03_streaming_hf_dataset/submit.py

@@ -0,0 +1,37 @@
+"""Streaming SWE-bench candidate-buffering example."""
+
+from __future__ import annotations
+
+import asyncio
+
+from ergon_builtins.benchmarks.swebench_verified.benchmark import _default_swebench_sandbox
+from ergon_builtins.benchmarks.swebench_verified.worker_factory import (
+    make_swebench_rubric,
+    make_swebench_worker,
+)
+from ergon_builtins.environments import SweBenchVerifiedEnvironment
+from ergon_core.api import Experiment, RandomSampler
+from experiment_api._shared import experiment_submission_service
+
+
+async def main() -> None:
+    env = SweBenchVerifiedEnvironment(
+        name="swebench-train",
+        split="train",
+        streaming=True,
+        worker=make_swebench_worker(model="openai:gpt-4o"),
+        evaluators=[make_swebench_rubric()],
+        sandbox=_default_swebench_sandbox(),
+    )
+    experiment = Experiment(name="swebench-streaming-buffer", environments=[env])
+    result = await experiment.submit(
+        service=experiment_submission_service(),
+        k=8,
+        candidate_pool_size=32,
+        sampler=RandomSampler(seed=0),
+    )
+    print(result.model_dump_json(indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())