From c55b733fbe2c233d415f342ab3f56b1b73d95ddf Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Tue, 26 May 2026 23:54:40 +0100 Subject: [PATCH 01/15] Add experiment rollout trainer adapter path --- .../infrastructure/http/routes/rollouts.py | 20 +++ .../ergon_core/core/rl/rollout_service.py | 134 ++++++++++++++++++ .../ergon_core/core/rl/rollout_types.py | 19 ++- .../rl/test_rollout_experiment_submission.py | 64 +++++++++ .../ergon_infra/adapters/openrlhf_http.py | 25 ++-- ergon_infra/ergon_infra/adapters/trl_http.py | 17 ++- ergon_infra/ergon_infra/adapters/verl_http.py | 19 ++- ergon_infra/ergon_infra/training/config.py | 6 +- .../ergon_infra/training/trl_runner.py | 12 +- .../test_training_experiment_submission.py | 59 ++++++++ 10 files changed, 345 insertions(+), 30 deletions(-) create mode 100644 ergon_core/tests/unit/rl/test_rollout_experiment_submission.py create mode 100644 ergon_infra/tests/unit/test_training_experiment_submission.py diff --git a/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py b/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py index 89fc62633..50634eb8b 100644 --- a/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py +++ b/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py @@ -15,6 +15,7 @@ RolloutBatchSummary, SubmitRequest, SubmitResponse, + TrainingRolloutRequest, WeightSyncRequest, WeightSyncResponse, ) @@ -51,6 +52,25 @@ def submit_rollout( return service.submit(request) +@router.post( + "/experiments/{experiment_id}/rollout-batches", + response_model=RolloutBatchSummary, + status_code=202, +) +async def submit_experiment_rollout_batch( + experiment_id: UUID, + request: TrainingRolloutRequest, + service: Annotated[RolloutService, Depends(get_rollout_service)], +) -> RolloutBatchSummary: + """Start a trainer batch from a persisted experiment candidate pool.""" + if request.experiment_id != experiment_id: + raise HTTPException(400, "experiment_id mismatch") + try: + return await service.submit_experiment_batch(request) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + + @router.get("/{batch_id}", response_model=PollResponse) def poll_rollout( batch_id: UUID, diff --git a/ergon_core/ergon_core/core/rl/rollout_service.py b/ergon_core/ergon_core/core/rl/rollout_service.py index 481f07bed..a8ab87530 100644 --- a/ergon_core/ergon_core/core/rl/rollout_service.py +++ b/ergon_core/ergon_core/core/rl/rollout_service.py @@ -12,8 +12,20 @@ from uuid import UUID, uuid4 import inngest +from ergon_core.api.experiment.experiment import ExperimentRef +from ergon_core.core.application.experiments.candidate_pool import sample_from_pool_entry +from ergon_core.core.application.experiments.repositories import ( + ExperimentRepository, + record_sampler_invocation, +) +from ergon_core.core.application.samples.materialization import materialize_sample from ergon_core.core.persistence.context.models import SampleContextEvent from ergon_core.core.persistence.definitions.models import ExperimentDefinition +from ergon_core.core.persistence.experiments.models import ( + ExperimentRow, + ExperimentSamplerInvocationRow, + ExperimentSamplePoolEntryRow, +) from ergon_core.core.persistence.shared.enums import ( TERMINAL_SAMPLE_STATUSES, SampleStatus, @@ -39,6 +51,7 @@ SubmitRequest, SubmitResponse, Trajectory, + TrainingRolloutRequest, ) from ergon_core.core.application.events.runtime import WorkflowStartedEvent from sqlmodel import Session, select @@ -193,6 +206,110 @@ def get_rollout_batch_by_id(self, batch_id: UUID) -> RolloutBatchSummary | None: with self._session_factory() as session: return self.get_rollout_batch(session, batch_id) + async def submit_experiment_batch( + self, + request: TrainingRolloutRequest, + ) -> RolloutBatchSummary: + """Launch a trainer batch from already-buffered experiment candidates.""" + with self._session_factory() as session: + experiment = session.get(ExperimentRow, request.experiment_id) + if experiment is None: + raise ValueError(f"Experiment {request.experiment_id} not found") + + repository = ExperimentRepository(session) + pool_size = request.candidate_pool_size or request.k + candidates = repository.pending_unselected_pool_entries(request.experiment_id) + if len(candidates) < request.k: + raise ValueError( + "Experiment candidate pool does not contain enough unselected samples. " + "Submit through the Python experiment API to replenish streamed environments." + ) + selected_entries = _select_candidate_entries( + candidates[:pool_size], + k=request.k, + sampler=request.sampler, + sampler_config=request.sampler_config, + ) + invocation = record_sampler_invocation( + session=session, + experiment_ref=ExperimentRef( + id=experiment.id, + name=experiment.name, + environment_ids=repository.environment_ids_for_experiment(experiment.id), + created_at=experiment.created_at, + metadata=experiment.metadata_json, + ), + sampler_name=request.sampler, + requested_k=request.k, + candidate_pool_size=pool_size, + selected_count=len(selected_entries), + sampler_config=request.sampler_config, + ) + repository.mark_pool_entries_selected( + list(selected_entries), + sampler_invocation_id=invocation.id, + ) + sample_ids = await self._materialize_pool_entries( + session=session, + invocation=invocation, + entries=selected_entries, + ) + summary = self.create_rollout_batch( + session, + sample_ids=sample_ids, + sampler_invocation_id=invocation.id, + ) + session.commit() + + for sample_id in summary.sample_ids: + self._inngest_send( + inngest.Event( + name=WorkflowStartedEvent.name, + data=WorkflowStartedEvent(sample_id=sample_id).model_dump(mode="json"), + ) + ) + return summary + + async def _materialize_pool_entries( + self, + *, + session: Session, + invocation: ExperimentSamplerInvocationRow, + entries: Sequence[ExperimentSamplePoolEntryRow], + ) -> list[UUID]: + sample_ids: list[UUID] = [] + for entry in entries: + sample = await sample_from_pool_entry(entry) + row = SampleRecord( + experiment_id=entry.experiment_id, + environment_id=entry.environment_id, + sampler_invocation_id=invocation.id, + pool_entry_id=entry.id, + sample_key=sample.sample_key, + sample_ref_json=dict(sample.sample_ref), + benchmark_type="experiment", + instance_key=sample.sample_key, + sample_id=sample.sample_key, + worker_team_json={}, + dependency_extras_json={}, + assignment_json={ + "sample_key": sample.sample_key, + "sample_name": sample.name, + "environment_name": sample.environment_name, + "sample_ref": dict(sample.sample_ref), + "source_metadata": dict(sample.source_metadata), + "metadata": dict(sample.metadata), + }, + experiment=str(entry.experiment_id), + status=SampleStatus.PENDING, + ) + session.add(row) + session.flush() + materialize_sample(session=session, sample=sample, sample_row=row) + sample_ids.append(row.id) + session.flush() + return sample_ids + def poll(self, batch_id: UUID) -> PollResponse | None: """Non-blocking status check. Extracts trajectories when all done.""" with self._session_factory() as session: @@ -374,3 +491,20 @@ def _extract_trajectories(self, sample_ids: list[UUID]) -> list[Trajectory]: ) ) return result + + +def _select_candidate_entries( + entries: Sequence[ExperimentSamplePoolEntryRow], + *, + k: int, + sampler: str, + sampler_config: dict[str, object], +) -> list[ExperimentSamplePoolEntryRow]: + selected = list(entries) + if sampler == "random": + import random + + random.Random(sampler_config.get("seed")).shuffle(selected) + elif sampler not in {"sequential", "all"}: + raise ValueError(f"Unsupported trainer sampler: {sampler}") + return selected[: min(k, len(selected))] diff --git a/ergon_core/ergon_core/core/rl/rollout_types.py b/ergon_core/ergon_core/core/rl/rollout_types.py index 99b37d0cf..cefb5adc5 100644 --- a/ergon_core/ergon_core/core/rl/rollout_types.py +++ b/ergon_core/ergon_core/core/rl/rollout_types.py @@ -7,7 +7,12 @@ from uuid import UUID from ergon_core.core.shared.rollout_status import RolloutStatus as BatchStatus -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field + + +def _to_camel(value: str) -> str: + parts = value.split("_") + return parts[0] + "".join(part.capitalize() for part in parts[1:]) class SubmitRequest(BaseModel): @@ -27,6 +32,18 @@ class SubmitResponse(BaseModel): status: BatchStatus = BatchStatus.PENDING +class TrainingRolloutRequest(BaseModel): + """Trainer → Ergon: select and launch samples from a persisted experiment.""" + + model_config = ConfigDict(alias_generator=_to_camel, populate_by_name=True) + + experiment_id: UUID + k: int = Field(ge=1) + sampler: str = "random" + sampler_config: dict[str, object] = Field(default_factory=dict) + candidate_pool_size: int | None = None + + class RolloutBatchSummary(BaseModel): """Durable trainer batch membership exposed by sample id.""" diff --git a/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py b/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py new file mode 100644 index 000000000..66b1de1a4 --- /dev/null +++ b/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py @@ -0,0 +1,64 @@ +from uuid import uuid4 + +from ergon_core.core.infrastructure.http.routes.rollouts import router +from ergon_core.core.rl.rollout_types import BatchStatus, RolloutBatchSummary +from fastapi import FastAPI +from fastapi.testclient import TestClient + + +class _FakeRolloutService: + def __init__(self) -> None: + self.request = None + + async def submit_experiment_batch(self, request): + self.request = request + return RolloutBatchSummary( + batch_id=uuid4(), + sample_ids=[uuid4()], + status=BatchStatus.PENDING, + sampler_invocation_id=uuid4(), + ) + + +def test_experiment_rollout_route_accepts_camel_case_training_request() -> None: + experiment_id = uuid4() + service = _FakeRolloutService() + app = FastAPI() + app.state.rollout_service = service + app.include_router(router) + client = TestClient(app) + + response = client.post( + f"/rollouts/experiments/{experiment_id}/rollout-batches", + json={ + "experimentId": str(experiment_id), + "k": 32, + "sampler": "random", + "samplerConfig": {"seed": 7}, + "candidatePoolSize": 256, + }, + ) + + assert response.status_code == 202 + assert service.request.experiment_id == experiment_id + assert service.request.k == 32 + assert service.request.sampler == "random" + assert service.request.sampler_config == {"seed": 7} + assert service.request.candidate_pool_size == 256 + + +def test_experiment_rollout_route_rejects_mismatched_experiment_id() -> None: + app = FastAPI() + app.state.rollout_service = _FakeRolloutService() + app.include_router(router) + client = TestClient(app) + + response = client.post( + f"/rollouts/experiments/{uuid4()}/rollout-batches", + json={ + "experimentId": str(uuid4()), + "k": 1, + }, + ) + + assert response.status_code == 400 diff --git a/ergon_infra/ergon_infra/adapters/openrlhf_http.py b/ergon_infra/ergon_infra/adapters/openrlhf_http.py index 557fea678..0333ea5ba 100644 --- a/ergon_infra/ergon_infra/adapters/openrlhf_http.py +++ b/ergon_infra/ergon_infra/adapters/openrlhf_http.py @@ -4,7 +4,7 @@ # OpenRLHF CLI: --agent_func_path ergon_infra/adapters/openrlhf_http.py - --agent_func_kwargs '{"ergon_url": "http://macbook:9000/api", "definition_id": ""}' + --agent_func_kwargs '{"ergon_url": "http://macbook:9000/api", "experiment_id": ""}' """ import asyncio @@ -17,7 +17,9 @@ logger = logging.getLogger(__name__) _client: httpx.AsyncClient | None = None -_definition_id: str = "" +_experiment_id: str = "" +_sampler: str = "random" +_candidate_pool_size: int | None = None _poll_interval_s: float = 2.0 _timeout_s: float = 300.0 @@ -28,14 +30,18 @@ class OpenRLHFCallbackContext(Protocol): def configure( ergon_url: str, - definition_id: str, + experiment_id: str, + sampler: str = "random", + candidate_pool_size: int | None = None, poll_interval_s: float = 2.0, timeout_s: float = 300.0, ) -> None: """Module-level configuration (called by OpenRLHF before agent_func).""" - global _client, _definition_id, _poll_interval_s, _timeout_s + global _client, _experiment_id, _sampler, _candidate_pool_size, _poll_interval_s, _timeout_s _client = httpx.AsyncClient(base_url=ergon_url, timeout=30.0) - _definition_id = definition_id + _experiment_id = experiment_id + _sampler = sampler + _candidate_pool_size = candidate_pool_size _poll_interval_s = poll_interval_s _timeout_s = timeout_s @@ -54,10 +60,13 @@ async def agent_func( raise RuntimeError("Call configure() before agent_func()") resp = await _client.post( - "/rollouts/submit", + f"/rollouts/experiments/{_experiment_id}/rollout-batches", json={ - "definition_id": _definition_id, - "num_episodes": 1, + "experimentId": _experiment_id, + "k": 1, + "sampler": _sampler, + "samplerConfig": {}, + "candidatePoolSize": _candidate_pool_size, }, ) resp.raise_for_status() diff --git a/ergon_infra/ergon_infra/adapters/trl_http.py b/ergon_infra/ergon_infra/adapters/trl_http.py index 9d82663a5..d692665ae 100644 --- a/ergon_infra/ergon_infra/adapters/trl_http.py +++ b/ergon_infra/ergon_infra/adapters/trl_http.py @@ -9,7 +9,7 @@ rollout_func = make_ergon_http_rollout_func( ergon_url="http://macbook:9000/api", - definition_id="", + experiment_id="", ) trainer = GRPOTrainer(..., rollout_func=rollout_func) """ @@ -40,7 +40,9 @@ class TRLTrainerContext(Protocol): def make_ergon_http_rollout_func( ergon_url: str, - definition_id: str, + experiment_id: str, + sampler: str = "random", + candidate_pool_size: int | None = None, poll_interval_s: float = 2.0, timeout_s: float = 300.0, ) -> Callable[[list, TRLTrainerContext], "RolloutBatch"]: @@ -48,7 +50,7 @@ def make_ergon_http_rollout_func( Args: ergon_url: base URL of the Ergon API (e.g. ``http://localhost:9000/api``). - definition_id: ExperimentDefinition UUID to run episodes against. + experiment_id: persisted experiment UUID to sample from. poll_interval_s: seconds between poll requests. timeout_s: max wall-clock seconds to wait for a batch to complete. @@ -59,10 +61,13 @@ def make_ergon_http_rollout_func( def rollout_func(prompts: list, trainer: TRLTrainerContext) -> RolloutBatch: resp = client.post( - "/rollouts/submit", + f"/rollouts/experiments/{experiment_id}/rollout-batches", json={ - "definition_id": definition_id, - "num_episodes": len(prompts), + "experimentId": experiment_id, + "k": len(prompts), + "sampler": sampler, + "samplerConfig": {}, + "candidatePoolSize": candidate_pool_size, }, ) resp.raise_for_status() diff --git a/ergon_infra/ergon_infra/adapters/verl_http.py b/ergon_infra/ergon_infra/adapters/verl_http.py index ab3c803b6..0915fdbe5 100644 --- a/ergon_infra/ergon_infra/adapters/verl_http.py +++ b/ergon_infra/ergon_infra/adapters/verl_http.py @@ -6,7 +6,7 @@ agent_loop: ergon agent_loop_kwargs: ergon_url: http://macbook:9000/api - definition_id: + experiment_id: """ import logging @@ -26,23 +26,30 @@ class ErgonAgentLoop(AgentLoopBase): def __init__( self, ergon_url: str, - definition_id: str, + experiment_id: str, + sampler: str = "random", + candidate_pool_size: int | None = None, poll_interval_s: float = 2.0, timeout_s: float = 300.0, **kwargs: object, ) -> None: super().__init__(**kwargs) self._client = httpx.AsyncClient(base_url=ergon_url, timeout=30.0) - self._definition_id = definition_id + self._experiment_id = experiment_id + self._sampler = sampler + self._candidate_pool_size = candidate_pool_size self._poll_interval_s = poll_interval_s self._timeout_s = timeout_s async def run(self, sampling_params: dict, **kwargs: object) -> AgentLoopOutput: resp = await self._client.post( - "/rollouts/submit", + f"/rollouts/experiments/{self._experiment_id}/rollout-batches", json={ - "definition_id": self._definition_id, - "num_episodes": 1, + "experimentId": self._experiment_id, + "k": 1, + "sampler": self._sampler, + "samplerConfig": {}, + "candidatePoolSize": self._candidate_pool_size, }, ) resp.raise_for_status() diff --git a/ergon_infra/ergon_infra/training/config.py b/ergon_infra/ergon_infra/training/config.py index b9817abcb..d9d37e229 100644 --- a/ergon_infra/ergon_infra/training/config.py +++ b/ergon_infra/ergon_infra/training/config.py @@ -19,7 +19,7 @@ class TrainingConfig(BaseModel): benchmark: str evaluator: str = "stub-rubric" limit: int | None = None - definition_id: str | None = None + experiment_id: str | None = None # -- Model ---------------------------------------------------------------- model: str = "Qwen/Qwen2.5-1.5B" @@ -68,7 +68,7 @@ def _build_training_parser() -> argparse.ArgumentParser: p.add_argument("--evaluator", type=str, default="stub-rubric", help="Evaluator slug") p.add_argument("--limit", type=int, default=None, help="Max tasks per episode") p.add_argument( - "--definition-id", type=str, default=None, help="Existing ExperimentDefinition UUID" + "--experiment-id", type=str, default=None, help="Existing experiment UUID" ) p.add_argument("--model", type=str, default="Qwen/Qwen2.5-1.5B", help="HuggingFace model ID") @@ -134,7 +134,7 @@ def training_config_from_args(argv: list[str] | None = None) -> TrainingConfig: benchmark=args.benchmark, evaluator=args.evaluator, limit=args.limit, - definition_id=args.definition_id, + experiment_id=args.experiment_id, model=args.model, device=args.device, vllm_mode=vllm_mode, diff --git a/ergon_infra/ergon_infra/training/trl_runner.py b/ergon_infra/ergon_infra/training/trl_runner.py index 13b9c5126..24b18fc6a 100644 --- a/ergon_infra/ergon_infra/training/trl_runner.py +++ b/ergon_infra/ergon_infra/training/trl_runner.py @@ -46,11 +46,11 @@ def run_trl_training(config: TrainingConfig) -> int: "(e.g. http://localhost:9000/api for local dev)." ) - definition_id = config.definition_id - if not definition_id: + experiment_id = config.experiment_id + if not experiment_id: raise ValueError( - "--definition-id is required. Create one via: " - "ergon experiment define --limit 1 --worker --model " + "--experiment-id is required. Create one via the Python experiment API " + "and persist its candidate pool before starting trainer rollouts." ) tokenizer = AutoTokenizer.from_pretrained(config.model) @@ -59,7 +59,7 @@ def run_trl_training(config: TrainingConfig) -> int: rollout_func = make_ergon_http_rollout_func( ergon_url=config.ergon_url, - definition_id=definition_id, + experiment_id=experiment_id, timeout_s=config.timeout_s, ) @@ -99,7 +99,7 @@ def reward_fn(completions: list[str], **kwargs: object) -> list[float]: logger.info("Starting TRL GRPO training") logger.info(" Ergon API: %s", config.ergon_url) logger.info(" Model: %s", config.model) - logger.info(" Definition: %s", definition_id) + logger.info(" Experiment: %s", experiment_id) logger.info(" Device: %s", config.device) logger.info(" Output dir: %s", config.output_dir) diff --git a/ergon_infra/tests/unit/test_training_experiment_submission.py b/ergon_infra/tests/unit/test_training_experiment_submission.py new file mode 100644 index 000000000..f55aca724 --- /dev/null +++ b/ergon_infra/tests/unit/test_training_experiment_submission.py @@ -0,0 +1,59 @@ +from uuid import uuid4 + +import ergon_infra.adapters.trl_http as trl_http + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def raise_for_status(self) -> None: + return None + + def json(self): + return self._payload + + +class _FakeClient: + def __init__(self, *args, **kwargs) -> None: + self.requests = [] + self._polls = 0 + + def post(self, path, *, json): + self.requests.append(("POST", path, json)) + return _FakeResponse({"batch_id": "batch-1"}) + + def get(self, path): + self.requests.append(("GET", path, None)) + return _FakeResponse({"status": "complete", "trajectories": []}) + + def delete(self, path): + self.requests.append(("DELETE", path, None)) + return _FakeResponse({}) + + +def test_trl_rollout_func_submits_experiment_batch(monkeypatch) -> None: + fake = _FakeClient() + monkeypatch.setattr(trl_http.httpx, "Client", lambda *args, **kwargs: fake) + experiment_id = uuid4() + + rollout_func = trl_http.make_ergon_http_rollout_func( + ergon_url="http://ergon.test", + experiment_id=str(experiment_id), + sampler="random", + candidate_pool_size=256, + ) + + rollout_func([{"prompt": "a"}, {"prompt": "b"}], object()) + + assert fake.requests[0] == ( + "POST", + f"/rollouts/experiments/{experiment_id}/rollout-batches", + { + "experimentId": str(experiment_id), + "k": 2, + "sampler": "random", + "samplerConfig": {}, + "candidatePoolSize": 256, + }, + ) From f2a91be92567280b5a2398879c36a95208b774b3 Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 00:25:19 +0100 Subject: [PATCH 02/15] Tighten experiment rollout trainer path --- .../ergon_core/core/rl/rollout_service.py | 8 +++++--- .../test_definition_identity_naming.py | 6 ++++++ .../tests/unit/rl/test_rollout_service.py | 17 +++++++++++++++++ ergon_infra/ergon_infra/training/config.py | 4 +--- 4 files changed, 29 insertions(+), 6 deletions(-) diff --git a/ergon_core/ergon_core/core/rl/rollout_service.py b/ergon_core/ergon_core/core/rl/rollout_service.py index a8ab87530..4a17942a5 100644 --- a/ergon_core/ergon_core/core/rl/rollout_service.py +++ b/ergon_core/ergon_core/core/rl/rollout_service.py @@ -14,7 +14,7 @@ import inngest from ergon_core.api.experiment.experiment import ExperimentRef from ergon_core.core.application.experiments.candidate_pool import sample_from_pool_entry -from ergon_core.core.application.experiments.repositories import ( +from ergon_core.core.application.experiments.repository import ( ExperimentRepository, record_sampler_invocation, ) @@ -99,7 +99,9 @@ def submit(self, request: SubmitRequest) -> SubmitResponse: with self._session_factory() as session: definition = session.get(ExperimentDefinition, request.definition_id) - benchmark_type = definition.benchmark_type if definition else "rl-rollout" + if definition is None: + raise ValueError(f"Definition {request.definition_id} not found") + benchmark_type = definition.benchmark_type session.add( RolloutBatch( id=batch_id, @@ -235,7 +237,7 @@ async def submit_experiment_batch( experiment_ref=ExperimentRef( id=experiment.id, name=experiment.name, - environment_ids=repository.environment_ids_for_experiment(experiment.id), + environment_ids={}, created_at=experiment.created_at, metadata=experiment.metadata_json, ), diff --git a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py index 3de9e4c39..4b8c0b794 100644 --- a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py +++ b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py @@ -25,14 +25,19 @@ 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" / "ergon_core" / "core" / "rl", 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 / "ergon_core" / "tests" / "unit" / "rl", ROOT / "ergon_cli" / "ergon_cli" / "domains" / "experiments", ROOT / "ergon_cli" / "ergon_cli" / "domains" / "samples", ROOT / "ergon_cli" / "tests" / "unit" / "cli", + ROOT / "ergon_infra" / "ergon_infra" / "adapters", + ROOT / "ergon_infra" / "ergon_infra" / "training", + ROOT / "ergon_infra" / "tests" / "unit", ROOT / "ergon-dashboard" / "src" / "app" / "experiments", ROOT / "ergon-dashboard" / "src" / "app" / "samples", ROOT / "ergon-dashboard" / "src" / "components" / "experiments", @@ -66,6 +71,7 @@ ROOT / "ergon-dashboard" / "src" / "lib" / "contracts" / "rest.ts", ROOT / "ergon-dashboard" / "src" / "lib" / "server-data" / "experiments.ts", ROOT / "ergon-dashboard" / "src" / "lib" / "server-data" / "samples.ts", + ROOT / "ergon_core" / "ergon_core" / "core" / "infrastructure" / "http" / "routes" / "rollouts.py", } ALLOWED_EXPERIMENT_ID_PATTERNS_BY_FILE = { ROOT / "ergon_core" / "ergon_core" / "api" / "experiment" / "experiment.py": ( diff --git a/ergon_core/tests/unit/rl/test_rollout_service.py b/ergon_core/tests/unit/rl/test_rollout_service.py index c8c271a8a..d4e6c6cdb 100644 --- a/ergon_core/tests/unit/rl/test_rollout_service.py +++ b/ergon_core/tests/unit/rl/test_rollout_service.py @@ -78,3 +78,20 @@ def test_rollout_submit_uses_rollout_batch_and_run_definition_without_legacy_rec assert {run.model_target for run in runs} == {"openai:test"} assert {membership.sample_id for membership in memberships} == {run.id for run in runs} assert len(sent_events) == 2 + + +def test_rollout_submit_rejects_missing_definition_without_fallback(session_factory) -> None: + service = RolloutService( + session_factory=session_factory, + inngest_send=lambda event: None, + tokenizer_name="unused-in-submit", + ) + definition_id = uuid4() + + with pytest.raises(ValueError, match=f"Definition {definition_id} not found"): + service.submit( + SubmitRequest( + definition_id=definition_id, + num_episodes=1, + ) + ) diff --git a/ergon_infra/ergon_infra/training/config.py b/ergon_infra/ergon_infra/training/config.py index d9d37e229..f52c32642 100644 --- a/ergon_infra/ergon_infra/training/config.py +++ b/ergon_infra/ergon_infra/training/config.py @@ -67,9 +67,7 @@ def _build_training_parser() -> argparse.ArgumentParser: p.add_argument("--benchmark", type=str, required=True, help="Benchmark slug") p.add_argument("--evaluator", type=str, default="stub-rubric", help="Evaluator slug") p.add_argument("--limit", type=int, default=None, help="Max tasks per episode") - p.add_argument( - "--experiment-id", type=str, default=None, help="Existing experiment UUID" - ) + p.add_argument("--experiment-id", type=str, default=None, help="Existing experiment UUID") p.add_argument("--model", type=str, default="Qwen/Qwen2.5-1.5B", help="HuggingFace model ID") From b68bd04cae25245ddce2a5b7722df9d039278589 Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 00:26:48 +0100 Subject: [PATCH 03/15] Format trainer identity guard --- .../unit/architecture/test_definition_identity_naming.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py index 4b8c0b794..34e0c7160 100644 --- a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py +++ b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py @@ -71,7 +71,14 @@ ROOT / "ergon-dashboard" / "src" / "lib" / "contracts" / "rest.ts", ROOT / "ergon-dashboard" / "src" / "lib" / "server-data" / "experiments.ts", ROOT / "ergon-dashboard" / "src" / "lib" / "server-data" / "samples.ts", - ROOT / "ergon_core" / "ergon_core" / "core" / "infrastructure" / "http" / "routes" / "rollouts.py", + ROOT + / "ergon_core" + / "ergon_core" + / "core" + / "infrastructure" + / "http" + / "routes" + / "rollouts.py", } ALLOWED_EXPERIMENT_ID_PATTERNS_BY_FILE = { ROOT / "ergon_core" / "ergon_core" / "api" / "experiment" / "experiment.py": ( From b2deec81765b64556ad8c4781020ef66f89434f7 Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 00:40:30 +0100 Subject: [PATCH 04/15] Align trainer rollout with experiment samples --- .../ergon_cli/domains/training/commands.py | 2 +- .../ergon_cli/domains/training/models.py | 2 +- .../ergon_cli/domains/training/parser.py | 2 +- .../ergon_cli/domains/training/service.py | 2 +- ergon_cli/tests/unit/cli/test_training_cli.py | 7 +- .../application/experiments/candidate_pool.py | 86 ++++++++++++++++++- .../ergon_core/core/rl/rollout_service.py | 63 ++------------ ergon_infra/ergon_infra/training/__main__.py | 2 +- 8 files changed, 102 insertions(+), 64 deletions(-) diff --git a/ergon_cli/ergon_cli/domains/training/commands.py b/ergon_cli/ergon_cli/domains/training/commands.py index d99f2dca6..9a92a94ab 100644 --- a/ergon_cli/ergon_cli/domains/training/commands.py +++ b/ergon_cli/ergon_cli/domains/training/commands.py @@ -20,7 +20,7 @@ def handle_train(args: Namespace) -> int: benchmark=args.benchmark, evaluator=args.evaluator, limit=args.limit, - definition_id=args.definition_id, + experiment_id=args.experiment_id, model=args.model, device=args.device, vllm_mode=None if args.device == "cpu" else args.vllm_mode, diff --git a/ergon_cli/ergon_cli/domains/training/models.py b/ergon_cli/ergon_cli/domains/training/models.py index 94af1d887..070ef90fa 100644 --- a/ergon_cli/ergon_cli/domains/training/models.py +++ b/ergon_cli/ergon_cli/domains/training/models.py @@ -11,7 +11,7 @@ class TrainingCommand(BaseModel): benchmark: str evaluator: str limit: int | None - definition_id: str | None + experiment_id: str | None model: str device: Literal["cpu", "cuda"] vllm_mode: Literal["colocate", "server"] | None diff --git a/ergon_cli/ergon_cli/domains/training/parser.py b/ergon_cli/ergon_cli/domains/training/parser.py index a359701cc..905fbfbb2 100644 --- a/ergon_cli/ergon_cli/domains/training/parser.py +++ b/ergon_cli/ergon_cli/domains/training/parser.py @@ -17,7 +17,7 @@ def register_train_parser(subparsers: argparse._SubParsersAction) -> None: train_local.add_argument("--benchmark", required=True, help="Benchmark slug") train_local.add_argument("--evaluator", default="stub-rubric", help="Evaluator slug") train_local.add_argument("--limit", type=int, default=None, help="Max tasks per episode") - train_local.add_argument("--definition-id", default=None, help="ExperimentDefinition UUID") + train_local.add_argument("--experiment-id", default=None, help="Experiment UUID") train_local.add_argument("--model", default="Qwen/Qwen2.5-1.5B", help="HuggingFace model ID") train_local.add_argument( "--device", default="cuda", choices=["cpu", "cuda"], help="Device type" diff --git a/ergon_cli/ergon_cli/domains/training/service.py b/ergon_cli/ergon_cli/domains/training/service.py index 76a7394a9..1333422f1 100644 --- a/ergon_cli/ergon_cli/domains/training/service.py +++ b/ergon_cli/ergon_cli/domains/training/service.py @@ -26,7 +26,7 @@ def run_training(command: TrainingCommand) -> int: benchmark=command.benchmark, evaluator=command.evaluator, limit=command.limit, - definition_id=command.definition_id, + experiment_id=command.experiment_id, model=command.model, device=command.device, vllm_mode=None if command.device == "cpu" else command.vllm_mode, diff --git a/ergon_cli/tests/unit/cli/test_training_cli.py b/ergon_cli/tests/unit/cli/test_training_cli.py index 4da99067b..b57eb65e0 100644 --- a/ergon_cli/tests/unit/cli/test_training_cli.py +++ b/ergon_cli/tests/unit/cli/test_training_cli.py @@ -10,7 +10,7 @@ def _train_args(**overrides: object) -> Namespace: "benchmark": "bench", "evaluator": "stub-rubric", "limit": None, - "definition_id": None, + "experiment_id": None, "model": "Qwen/Qwen2.5-1.5B", "device": "cuda", "vllm_mode": "server", @@ -43,8 +43,11 @@ def fake_run_training(command): monkeypatch.setattr(training_commands, "run_training", fake_run_training) - rc = training_commands.handle_train(_train_args(device="cpu", vllm_mode="server")) + rc = training_commands.handle_train( + _train_args(device="cpu", vllm_mode="server", experiment_id="experiment-1") + ) assert rc == 0 + assert captured["command"].experiment_id == "experiment-1" assert captured["command"].device == "cpu" assert captured["command"].vllm_mode is None diff --git a/ergon_core/ergon_core/core/application/experiments/candidate_pool.py b/ergon_core/ergon_core/core/application/experiments/candidate_pool.py index a5d236563..aa6239139 100644 --- a/ergon_core/ergon_core/core/application/experiments/candidate_pool.py +++ b/ergon_core/ergon_core/core/application/experiments/candidate_pool.py @@ -3,6 +3,7 @@ from collections.abc import Sequence from uuid import UUID, uuid4 +from pydantic import JsonValue from sqlmodel import Session from ergon_core.api.benchmark import Task @@ -11,8 +12,13 @@ from ergon_core.api.experiment.sample import Sample from ergon_core.core.application.experiments.repository import ( ExperimentRepository, + record_sampler_invocation, +) +from ergon_core.core.persistence.experiments.models import ( + ExperimentRow, + ExperimentSamplerInvocationRow, + ExperimentSamplePoolEntryRow, ) -from ergon_core.core.persistence.experiments.models import ExperimentSamplePoolEntryRow class SampleCandidatePool: @@ -95,6 +101,64 @@ def mark_selected( ) +def reserve_sample_pool_entries_for_sampler( + *, + session: Session, + experiment_id: UUID, + k: int, + candidate_pool_size: int | None, + sampler_name: str, + sampler_config: dict[str, JsonValue] | None = None, +) -> tuple[ExperimentSamplerInvocationRow, list[ExperimentSamplePoolEntryRow]]: + """Reserve existing candidate-pool rows for a trainer sampler. + + Trainer rollout services can ask the experiment application domain for + already-buffered samples without importing the experiment repository + directly. Stream replenishment remains the Python authoring API's job. + """ + + experiment = session.get(ExperimentRow, experiment_id) + if experiment is None: + raise ValueError(f"Experiment {experiment_id} not found") + + repository = ExperimentRepository(session) + pool_size = candidate_pool_size or k + candidates = repository.pending_unselected_pool_entries(experiment_id) + if len(candidates) < k: + raise ValueError( + "Experiment candidate pool does not contain enough unselected samples. " + "Submit through the Python experiment API to replenish streamed environments." + ) + + config = dict(sampler_config or {}) + selected_entries = _select_candidate_entries( + candidates[:pool_size], + k=k, + sampler_name=sampler_name, + sampler_config=config, + ) + invocation = record_sampler_invocation( + session=session, + experiment_ref=ExperimentRef( + id=experiment.id, + name=experiment.name, + environment_ids={}, + created_at=experiment.created_at, + metadata=experiment.metadata_json, + ), + sampler_name=sampler_name, + requested_k=k, + candidate_pool_size=pool_size, + selected_count=len(selected_entries), + sampler_config=config, + ) + repository.mark_pool_entries_selected( + selected_entries, + sampler_invocation_id=invocation.id, + ) + return invocation, selected_entries + + async def sample_from_pool_entry(entry: ExperimentSamplePoolEntryRow) -> Sample: """Rehydrate retained candidate JSON into an authored, unmaterialized Sample.""" @@ -133,3 +197,23 @@ async def _task_from_candidate_snapshot(task_json: object) -> Task: task.task_payload = EmptyTaskPayload() task._task_id = None return task + + +def _select_candidate_entries( + entries: Sequence[ExperimentSamplePoolEntryRow], + *, + k: int, + sampler_name: str, + sampler_config: dict[str, JsonValue], +) -> list[ExperimentSamplePoolEntryRow]: + selected = list(entries) + if sampler_name == "random": + import random + + seed = sampler_config.get("seed") + if seed is not None and not isinstance(seed, (str, bytes, bytearray, int, float)): + raise ValueError("Random trainer sampler seed must be a scalar value") + random.Random(seed).shuffle(selected) + elif sampler_name not in {"sequential", "all"}: + raise ValueError(f"Unsupported trainer sampler: {sampler_name}") + return selected[: min(k, len(selected))] diff --git a/ergon_core/ergon_core/core/rl/rollout_service.py b/ergon_core/ergon_core/core/rl/rollout_service.py index 4a17942a5..8c7e78b9b 100644 --- a/ergon_core/ergon_core/core/rl/rollout_service.py +++ b/ergon_core/ergon_core/core/rl/rollout_service.py @@ -12,17 +12,14 @@ from uuid import UUID, uuid4 import inngest -from ergon_core.api.experiment.experiment import ExperimentRef -from ergon_core.core.application.experiments.candidate_pool import sample_from_pool_entry -from ergon_core.core.application.experiments.repository import ( - ExperimentRepository, - record_sampler_invocation, +from ergon_core.core.application.experiments.candidate_pool import ( + reserve_sample_pool_entries_for_sampler, + sample_from_pool_entry, ) from ergon_core.core.application.samples.materialization import materialize_sample from ergon_core.core.persistence.context.models import SampleContextEvent from ergon_core.core.persistence.definitions.models import ExperimentDefinition from ergon_core.core.persistence.experiments.models import ( - ExperimentRow, ExperimentSamplerInvocationRow, ExperimentSamplePoolEntryRow, ) @@ -214,43 +211,14 @@ async def submit_experiment_batch( ) -> RolloutBatchSummary: """Launch a trainer batch from already-buffered experiment candidates.""" with self._session_factory() as session: - experiment = session.get(ExperimentRow, request.experiment_id) - if experiment is None: - raise ValueError(f"Experiment {request.experiment_id} not found") - - repository = ExperimentRepository(session) - pool_size = request.candidate_pool_size or request.k - candidates = repository.pending_unselected_pool_entries(request.experiment_id) - if len(candidates) < request.k: - raise ValueError( - "Experiment candidate pool does not contain enough unselected samples. " - "Submit through the Python experiment API to replenish streamed environments." - ) - selected_entries = _select_candidate_entries( - candidates[:pool_size], - k=request.k, - sampler=request.sampler, - sampler_config=request.sampler_config, - ) - invocation = record_sampler_invocation( + invocation, selected_entries = reserve_sample_pool_entries_for_sampler( session=session, - experiment_ref=ExperimentRef( - id=experiment.id, - name=experiment.name, - environment_ids={}, - created_at=experiment.created_at, - metadata=experiment.metadata_json, - ), + experiment_id=request.experiment_id, + k=request.k, + candidate_pool_size=request.candidate_pool_size, sampler_name=request.sampler, - requested_k=request.k, - candidate_pool_size=pool_size, - selected_count=len(selected_entries), sampler_config=request.sampler_config, ) - repository.mark_pool_entries_selected( - list(selected_entries), - sampler_invocation_id=invocation.id, - ) sample_ids = await self._materialize_pool_entries( session=session, invocation=invocation, @@ -493,20 +461,3 @@ def _extract_trajectories(self, sample_ids: list[UUID]) -> list[Trajectory]: ) ) return result - - -def _select_candidate_entries( - entries: Sequence[ExperimentSamplePoolEntryRow], - *, - k: int, - sampler: str, - sampler_config: dict[str, object], -) -> list[ExperimentSamplePoolEntryRow]: - selected = list(entries) - if sampler == "random": - import random - - random.Random(sampler_config.get("seed")).shuffle(selected) - elif sampler not in {"sequential", "all"}: - raise ValueError(f"Unsupported trainer sampler: {sampler}") - return selected[: min(k, len(selected))] diff --git a/ergon_infra/ergon_infra/training/__main__.py b/ergon_infra/ergon_infra/training/__main__.py index e1529a765..fabf9ce2e 100644 --- a/ergon_infra/ergon_infra/training/__main__.py +++ b/ergon_infra/ergon_infra/training/__main__.py @@ -10,7 +10,7 @@ python -m ergon_infra.training \\ --benchmark smoke-test \\ - --definition-id \\ + --experiment-id \\ --model Qwen/Qwen2.5-1.5B \\ --vllm-mode colocate \\ --output-dir /checkpoints/test From 92db2d79732d0e31f9bb237fb74d907588380f7a Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 00:43:51 +0100 Subject: [PATCH 05/15] Allow trainer CLI experiment identity --- .../tests/unit/architecture/test_definition_identity_naming.py | 1 + 1 file changed, 1 insertion(+) diff --git a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py index 34e0c7160..7e21e7884 100644 --- a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py +++ b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py @@ -34,6 +34,7 @@ ROOT / "ergon_core" / "tests" / "unit" / "rl", ROOT / "ergon_cli" / "ergon_cli" / "domains" / "experiments", ROOT / "ergon_cli" / "ergon_cli" / "domains" / "samples", + ROOT / "ergon_cli" / "ergon_cli" / "domains" / "training", ROOT / "ergon_cli" / "tests" / "unit" / "cli", ROOT / "ergon_infra" / "ergon_infra" / "adapters", ROOT / "ergon_infra" / "ergon_infra" / "training", From 95d02e46afe0c0ea3e8c5dc8924cf2a3587f4bae Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 01:05:43 +0100 Subject: [PATCH 06/15] Remove stale definition rollout launch surfaces --- ergon_builtins/AGENTS.md | 18 +-- .../ergon_builtins/benchmarks/README.md | 5 +- ergon_cli/ergon_cli/domains/eval/service.py | 3 +- .../infrastructure/http/routes/rollouts.py | 11 -- .../http/routes/test_harness.py | 6 +- ergon_core/ergon_core/core/rl/eval_runner.py | 112 ++++-------------- .../ergon_core/core/rl/rollout_service.py | 74 +----------- .../ergon_core/core/rl/rollout_types.py | 17 --- .../test_definition_identity_naming.py | 1 + .../tests/unit/rest_api/test_rollouts_di.py | 15 ++- ergon_core/tests/unit/rl/test_eval_runner.py | 20 ++++ .../rl/test_rollout_experiment_submission.py | 17 +++ .../tests/unit/rl/test_rollout_service.py | 97 --------------- .../ergon_infra/templates/eval-only.yaml | 22 ++-- .../templates/train-trl-multinode.yaml | 6 +- .../ergon_infra/templates/train-trl.yaml | 6 +- .../templates/train-with-eval.yaml | 12 +- .../ergon_infra/training/trl_runner.py | 2 +- tests/e2e/conftest.py | 110 ----------------- 19 files changed, 117 insertions(+), 437 deletions(-) create mode 100644 ergon_core/tests/unit/rl/test_eval_runner.py delete mode 100644 ergon_core/tests/unit/rl/test_rollout_service.py diff --git a/ergon_builtins/AGENTS.md b/ergon_builtins/AGENTS.md index ec20e7845..39593da36 100644 --- a/ergon_builtins/AGENTS.md +++ b/ergon_builtins/AGENTS.md @@ -14,13 +14,13 @@ update this doc in the same PR. | Goal | Command | |---|---| -| Populate **SANDBOX** panel (stdin/stdout events) with no LLM | `ergon experiment define researchrubrics-smoke --worker canonical-smoke --model stub:constant --limit 1 && ergon experiment run ` | -| Populate **GENERATIONS** without calling a model | `ergon experiment define smoke-test --worker training-stub --model stub:constant --limit 1 && ergon experiment run ` | -| Populate **EVALUATION** with a passing gate, no LLM | any benchmark + `--evaluator stub-rubric` | -| Populate **EVALUATION** with varied scores (RL reward-shape test) | any benchmark + `--evaluator varied-stub-rubric` | -| Test a real ReAct agent end-to-end | `ergon experiment define swebench-verified --worker swebench-react --model openai:gpt-4o --limit 1 && ergon experiment run ` | -| Test manager → researcher delegation with a real LLM | `ergon experiment define researchrubrics-smoke --worker researchrubrics-researcher --model openai:gpt-4o --limit 1 && ergon experiment run ` | -| Test Lean 4 proof verification | `ergon experiment define minif2f --worker minif2f-react --model openai:gpt-4o --limit 1 && ergon experiment run ` (needs Lean sandbox) | +| Populate **SANDBOX** panel (stdin/stdout events) with no LLM | Use a Python submit script that builds a `ResearchRubricsEnvironment` with the canonical smoke worker and calls `Experiment.submit(...)`. | +| Populate **GENERATIONS** without calling a model | Use a Python submit script with an environment bound to `TrainingStubWorker`; inspect with `ergon sample show `. | +| Populate **EVALUATION** with a passing gate, no LLM | Bind `stub-rubric` in the environment's evaluator list. | +| Populate **EVALUATION** with varied scores (RL reward-shape test) | Bind `varied-stub-rubric` in the environment's evaluator list. | +| Test a real ReAct agent end-to-end | Build the relevant builtin environment in Python with the ReAct worker and submit it through `Experiment.submit(...)`. | +| Test manager -> researcher delegation with a real LLM | Build `ResearchRubricsEnvironment` in Python with `researchrubrics-researcher` and submit it through `Experiment.submit(...)`. | +| Test Lean 4 proof verification | Build `MiniF2FEnvironment` in Python with the MiniF2F worker/rubric and submit it through `Experiment.submit(...)` (needs Lean sandbox). | --- @@ -36,8 +36,8 @@ Which worker emits what. `—` = not applicable, `✗` = nothing emitted. | `minif2f-react` | ✓ | ✓ (proof artifact) | ✓ (Lean files) | ✓ | | `researchrubrics-researcher` | ✓ | ✓ (SampleResource kind=REPORT) | ✓ (writes `final_output/report.md`) | ✗ | -EVALUATION is populated by whichever **evaluator** you pass with -`--evaluator`; see table below. +EVALUATION is populated by the **evaluators** bound to the submitted +environment; see table below. --- diff --git a/ergon_builtins/ergon_builtins/benchmarks/README.md b/ergon_builtins/ergon_builtins/benchmarks/README.md index d464dae97..3c655ea5b 100644 --- a/ergon_builtins/ergon_builtins/benchmarks/README.md +++ b/ergon_builtins/ergon_builtins/benchmarks/README.md @@ -1,8 +1,7 @@ # Builtin Benchmarks Each subdirectory is one benchmark. Import from Python; **there is no -CLI authoring path** (PR 6.5 deleted `ergon experiment define` / -`ergon run `). Authoring is Python-only; the CLI is for +CLI authoring path**. Authoring is Python-only; the CLI is for observation (`ergon experiment show`, `ergon sample status`, and related read commands). @@ -95,6 +94,6 @@ Or just `ls ergon_builtins/ergon_builtins/benchmarks/`. After persisting and launching a benchmark from Python, use the CLI to observe its state: - `ergon run status ` — current state of one run -- `ergon run list [--status=S] [--definition-id=]` — list runs, optionally filtered +- `ergon sample list [--status=S] [--experiment-id=]` — list samples, optionally filtered - `ergon experiment show ` — full experiment detail (UUID-based) - `ergon experiment list` — list recent experiments diff --git a/ergon_cli/ergon_cli/domains/eval/service.py b/ergon_cli/ergon_cli/domains/eval/service.py index df1e23eaa..99cdf2805 100644 --- a/ergon_cli/ergon_cli/domains/eval/service.py +++ b/ergon_cli/ergon_cli/domains/eval/service.py @@ -20,11 +20,10 @@ async def run_eval(command: EvalCommand) -> int: if command.checkpoint is None: raise CliUsageError("Usage: ergon eval checkpoint --checkpoint ") - await evaluate_checkpoint( + return await evaluate_checkpoint( checkpoint_path=command.checkpoint, benchmark_type=command.benchmark, evaluator_type=command.evaluator, model_base=command.model_base, eval_limit=command.eval_limit, ) - return 0 diff --git a/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py b/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py index 50634eb8b..9b0894dda 100644 --- a/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py +++ b/ergon_core/ergon_core/core/infrastructure/http/routes/rollouts.py @@ -13,8 +13,6 @@ from ergon_core.core.rl.rollout_types import ( PollResponse, RolloutBatchSummary, - SubmitRequest, - SubmitResponse, TrainingRolloutRequest, WeightSyncRequest, WeightSyncResponse, @@ -43,15 +41,6 @@ def get_vllm_manager(request: Request) -> VLLMManager | None: return cast(VLLMManager, manager) -@router.post("/submit", response_model=SubmitResponse, status_code=202) -def submit_rollout( - request: SubmitRequest, - service: Annotated[RolloutService, Depends(get_rollout_service)], -) -> SubmitResponse: - """Start a batch of episodes. Returns immediately with batch_id.""" - return service.submit(request) - - @router.post( "/experiments/{experiment_id}/rollout-batches", response_model=RolloutBatchSummary, diff --git a/ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py b/ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py index bfd08f3ab..1424dfc55 100644 --- a/ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py +++ b/ergon_core/ergon_core/core/infrastructure/http/routes/test_harness.py @@ -33,7 +33,11 @@ SweBenchSmokeBenchmark, ) -router = APIRouter(prefix="/api/__danger__/test-harness", tags=["danger-test-harness"]) +router = APIRouter( + prefix="/api/__danger__/test-harness", + tags=["danger-test-harness"], + include_in_schema=False, +) _SMOKE_BENCHMARKS = { benchmark.type_slug: benchmark diff --git a/ergon_core/ergon_core/core/rl/eval_runner.py b/ergon_core/ergon_core/core/rl/eval_runner.py index 2f2181413..ab30cb04a 100644 --- a/ergon_core/ergon_core/core/rl/eval_runner.py +++ b/ergon_core/ergon_core/core/rl/eval_runner.py @@ -1,8 +1,9 @@ -"""Eval watcher: score checkpoints on Ergon benchmarks. +"""Eval watcher: score checkpoints through Pythonic experiment submit scripts. -Watches a checkpoint directory, detects new checkpoints, runs -``ergon experiment define`` + ``ergon experiment run`` against each, -and optionally reports results. +Watches a checkpoint directory, detects new checkpoints, and delegates each +checkpoint to an explicit command supplied by the caller. The CLI no longer +constructs or submits experiment work itself; benchmark/environment composition +stays in Python authoring code. The watcher runs on CPU. For vLLM-based evaluation, use ``--on-checkpoint`` to spawn a SkyPilot GPU job per checkpoint. @@ -10,17 +11,13 @@ import asyncio import logging -import re import shlex import subprocess from ergon_core.core.rl.checkpoint import CheckpointInfo, discover_checkpoints logger = logging.getLogger(__name__) -_UUID_RE = re.compile( - r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", - re.IGNORECASE, -) +LOCAL_EVAL_UNSUPPORTED_EXIT_CODE = 2 async def watch_and_evaluate( @@ -103,83 +100,24 @@ async def _run_local_eval( model_base: str, eval_limit: int | None, ) -> int: - """Run checkpoint evaluation locally via the experiment CLI. Returns exit code. + """Reject local CLI-launched evaluation. - Uses the checkpoint path as the vLLM model target so each checkpoint - is actually evaluated (not just the base model). + The removed CLI launch path used to hide benchmark composition inside the + eval CLI. Callers should use ``--on-checkpoint`` with + a Python entrypoint that builds the desired Experiment/Environment/Sampler + objects and submits them directly. """ - if eval_limit is None: - raise ValueError("--eval-limit is required for local checkpoint evaluation") - - model_target = f"vllm:{ckpt.path}" - - define_cmd = [ - "ergon", - "experiment", - "define", + logger.error( + "Local checkpoint evaluation is no longer supported by the CLI. " + "Use --on-checkpoint with a Python experiment submission command " + "(checkpoint=%s benchmark=%s evaluator=%s model_base=%s eval_limit=%s).", + ckpt.path, benchmark_type, - "--worker", - "training-stub", - "--evaluator", evaluator_type, - "--model", - model_target, - "--limit", - str(eval_limit), - ] - - logger.info("Defining local eval experiment for step %d: %s", ckpt.step, " ".join(define_cmd)) - - env = dict(__import__("os").environ) - env["ERGON_CHECKPOINT_STEP"] = str(ckpt.step) - env["ERGON_CHECKPOINT_PATH"] = ckpt.path - - try: - define_proc = await asyncio.create_subprocess_exec( - *define_cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - env=env, - ) - stdout, stderr = await define_proc.communicate() - - exit_code = 0 if define_proc.returncode is None else define_proc.returncode - output = stdout.decode() + stderr.decode() - if exit_code != 0: - logger.warning( - "Eval experiment definition failed for step %d (exit %d): %s", - ckpt.step, - exit_code, - output[:500], - ) - return exit_code - - definition_id = _parse_uuid_line("EXPERIMENT_ID=", output) - run_cmd = ["ergon", "experiment", "run", definition_id] - logger.info("Running local eval for step %d: %s", ckpt.step, " ".join(run_cmd)) - - run_proc = await asyncio.create_subprocess_exec( - *run_cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - env=env, - ) - _stdout, stderr = await run_proc.communicate() - - exit_code = 0 if run_proc.returncode is None else run_proc.returncode - if exit_code == 0: - logger.info("Eval complete for step %d", ckpt.step) - else: - logger.warning( - "Eval failed for step %d (exit %d): %s", - ckpt.step, - exit_code, - stderr.decode()[:500], - ) - return exit_code - except Exception: # slopcop: ignore[no-broad-except] - logger.exception("Eval crashed for step %d", ckpt.step) - return 1 + model_base, + eval_limit, + ) + return LOCAL_EVAL_UNSUPPORTED_EXIT_CODE async def evaluate_checkpoint( @@ -205,13 +143,3 @@ async def evaluate_checkpoint( model_base=model_base, eval_limit=eval_limit, ) - - -def _parse_uuid_line(prefix: str, output: str) -> str: - for line in output.splitlines(): - if not line.startswith(prefix): - continue - match = _UUID_RE.search(line) - if match is not None: - return match.group(0) - raise RuntimeError(f"missing {prefix} line in CLI output:\n{output}") diff --git a/ergon_core/ergon_core/core/rl/rollout_service.py b/ergon_core/ergon_core/core/rl/rollout_service.py index 8c7e78b9b..0ead10e2e 100644 --- a/ergon_core/ergon_core/core/rl/rollout_service.py +++ b/ergon_core/ergon_core/core/rl/rollout_service.py @@ -1,7 +1,7 @@ """Rollout-as-a-Service: orchestrate episode batches for RL trainers. Encapsulates all logic previously inline in trl_adapter.py. Both the -HTTP endpoint (/rollouts/) and any in-process callers delegate here. +HTTP endpoints (/rollouts/) and any in-process callers delegate here. Batch state is durable in PG — survives API restarts. """ @@ -9,7 +9,7 @@ import logging from collections import defaultdict from collections.abc import Callable, Sequence -from uuid import UUID, uuid4 +from uuid import UUID import inngest from ergon_core.core.application.experiments.candidate_pool import ( @@ -18,7 +18,6 @@ ) from ergon_core.core.application.samples.materialization import materialize_sample from ergon_core.core.persistence.context.models import SampleContextEvent -from ergon_core.core.persistence.definitions.models import ExperimentDefinition from ergon_core.core.persistence.experiments.models import ( ExperimentSamplerInvocationRow, ExperimentSamplePoolEntryRow, @@ -27,7 +26,6 @@ TERMINAL_SAMPLE_STATUSES, SampleStatus, ) -from ergon_core.core.persistence.shared.ids import new_id from ergon_core.core.persistence.telemetry.models import ( RolloutBatch, RolloutBatchSampleMembership, @@ -45,8 +43,6 @@ EpisodeFailure, PollResponse, RolloutBatchSummary, - SubmitRequest, - SubmitResponse, Trajectory, TrainingRolloutRequest, ) @@ -60,7 +56,7 @@ class RolloutService: """Orchestrate rollout batches: create runs, fire events, poll, extract. Lifecycle: - 1. Trainer calls ``submit()`` → SampleRecords + RolloutBatch created, Inngest events fired + 1. Trainer calls ``submit_experiment_batch()`` → SampleRecords + RolloutBatch created, Inngest events fired 2. Trainer polls ``poll()`` → returns RUNNING until all episodes finish 3. When all terminal → ``poll()`` extracts trajectories and returns COMPLETE @@ -89,70 +85,6 @@ def _get_tokenizer(self) -> Tokenizer: self._tokenizer = AutoTokenizer.from_pretrained(self._tokenizer_name) return self._tokenizer - def submit(self, request: SubmitRequest) -> SubmitResponse: - """Create SampleRecords, RolloutBatch, and fire Inngest workflow/started events.""" - batch_id = uuid4() - sample_ids: list[UUID] = [] - - with self._session_factory() as session: - definition = session.get(ExperimentDefinition, request.definition_id) - if definition is None: - raise ValueError(f"Definition {request.definition_id} not found") - benchmark_type = definition.benchmark_type - session.add( - RolloutBatch( - id=batch_id, - status=BatchStatus.PENDING, - ) - ) - - for index in range(request.num_episodes): - sample_id = new_id() - session.add( - SampleRecord( - id=sample_id, - definition_id=request.definition_id, - benchmark_type=benchmark_type, - instance_key=f"episode-{index}", - worker_team_json={"primary": "rl-rollout"}, - model_target=request.model_target_override, - status=SampleStatus.PENDING, - ) - ) - session.add( - RolloutBatchSampleMembership( - batch_id=batch_id, - sample_id=sample_id, - ordinal=index, - ) - ) - sample_ids.append(sample_id) - - session.commit() - - for sample_id in sample_ids: - self._inngest_send( - inngest.Event( - name=WorkflowStartedEvent.name, - data=WorkflowStartedEvent( - sample_id=sample_id, - definition_id=request.definition_id, - ).model_dump(mode="json"), - ) - ) - - logger.info( - "Submitted batch %s: %d episodes for definition %s", - batch_id, - request.num_episodes, - request.definition_id, - ) - return SubmitResponse( - batch_id=batch_id, - sample_ids=sample_ids, - status=BatchStatus.PENDING, - ) - def create_rollout_batch( self, session: Session, diff --git a/ergon_core/ergon_core/core/rl/rollout_types.py b/ergon_core/ergon_core/core/rl/rollout_types.py index cefb5adc5..95bea5036 100644 --- a/ergon_core/ergon_core/core/rl/rollout_types.py +++ b/ergon_core/ergon_core/core/rl/rollout_types.py @@ -15,23 +15,6 @@ def _to_camel(value: str) -> str: return parts[0] + "".join(part.capitalize() for part in parts[1:]) -class SubmitRequest(BaseModel): - """Trainer → Ergon: start a batch of episodes.""" - - definition_id: UUID - num_episodes: int = Field(ge=1) - policy_version: int | None = None - model_target_override: str | None = None - - -class SubmitResponse(BaseModel): - """Ergon → Trainer: batch accepted.""" - - batch_id: UUID - sample_ids: list[UUID] - status: BatchStatus = BatchStatus.PENDING - - class TrainingRolloutRequest(BaseModel): """Trainer → Ergon: select and launch samples from a persisted experiment.""" diff --git a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py index 7e21e7884..a5a03c337 100644 --- a/ergon_core/tests/unit/architecture/test_definition_identity_naming.py +++ b/ergon_core/tests/unit/architecture/test_definition_identity_naming.py @@ -44,6 +44,7 @@ ROOT / "ergon-dashboard" / "src" / "components" / "experiments", ROOT / "ergon-dashboard" / "src" / "components" / "indexes", ROOT / "ergon-dashboard" / "src" / "components" / "samples", + ROOT / "ergon-dashboard" / "src" / "generated" / "rest", ROOT / "ergon-dashboard" / "src" / "lib" / "sample-state", ROOT / "ergon-dashboard" / "tests" / "components", ROOT / "ergon-dashboard" / "tests" / "contracts", diff --git a/ergon_core/tests/unit/rest_api/test_rollouts_di.py b/ergon_core/tests/unit/rest_api/test_rollouts_di.py index 6dab21ad4..89bd199f2 100644 --- a/ergon_core/tests/unit/rest_api/test_rollouts_di.py +++ b/ergon_core/tests/unit/rest_api/test_rollouts_di.py @@ -9,12 +9,15 @@ class _FakeRolloutService: def __init__(self) -> None: self.batch_id = uuid4() self.sample_id = uuid4() + self.submitted_request = None - def submit(self, _request: object) -> dict[str, object]: + async def submit_experiment_batch(self, request: object) -> dict[str, object]: + self.submitted_request = request return { "batch_id": self.batch_id, "sample_ids": [self.sample_id], "status": "pending", + "sampler_invocation_id": str(uuid4()), } def get_rollout_batch_by_id(self, _batch_id: object) -> dict[str, object]: @@ -33,21 +36,23 @@ def restart(self, checkpoint_path: str) -> None: self.restarted_with = checkpoint_path -def test_rollout_router_gets_service_from_app_state() -> None: +def test_rollout_router_gets_experiment_service_from_app_state() -> None: + experiment_id = uuid4() app = FastAPI() app.state.rollout_service = _FakeRolloutService() app.include_router(router) client = TestClient(app) resp = client.post( - "/rollouts/submit", + f"/rollouts/experiments/{experiment_id}/rollout-batches", json={ - "definition_id": str(uuid4()), - "num_episodes": 1, + "experimentId": str(experiment_id), + "k": 1, }, ) assert resp.status_code == 202 + assert app.state.rollout_service.submitted_request.experiment_id == experiment_id def test_rollout_batch_route_exposes_sample_ids() -> None: diff --git a/ergon_core/tests/unit/rl/test_eval_runner.py b/ergon_core/tests/unit/rl/test_eval_runner.py new file mode 100644 index 000000000..b8b69e334 --- /dev/null +++ b/ergon_core/tests/unit/rl/test_eval_runner.py @@ -0,0 +1,20 @@ +import asyncio + +from ergon_core.core.rl.eval_runner import ( + LOCAL_EVAL_UNSUPPORTED_EXIT_CODE, + evaluate_checkpoint, +) + + +def test_checkpoint_eval_does_not_submit_experiments_through_cli() -> None: + exit_code = asyncio.run( + evaluate_checkpoint( + checkpoint_path="/tmp/checkpoint-1", + benchmark_type="gdpeval", + evaluator_type="stub-rubric", + model_base="base-model", + eval_limit=1, + ) + ) + + assert exit_code == LOCAL_EVAL_UNSUPPORTED_EXIT_CODE diff --git a/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py b/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py index 66b1de1a4..c090dfecf 100644 --- a/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py +++ b/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py @@ -62,3 +62,20 @@ def test_experiment_rollout_route_rejects_mismatched_experiment_id() -> None: ) assert response.status_code == 400 + + +def test_definition_rollout_submit_route_does_not_accept_public_submissions() -> None: + app = FastAPI() + app.state.rollout_service = _FakeRolloutService() + app.include_router(router) + client = TestClient(app) + + response = client.post( + "/rollouts/submit", + json={ + "definitionId": str(uuid4()), + "numEpisodes": 1, + }, + ) + + assert response.status_code in {404, 405} diff --git a/ergon_core/tests/unit/rl/test_rollout_service.py b/ergon_core/tests/unit/rl/test_rollout_service.py deleted file mode 100644 index d4e6c6cdb..000000000 --- a/ergon_core/tests/unit/rl/test_rollout_service.py +++ /dev/null @@ -1,97 +0,0 @@ -from uuid import uuid4 - -import inngest -import pytest -from ergon_core.core.persistence.definitions.models import ExperimentDefinition -from ergon_core.core.persistence.telemetry.models import ( - RolloutBatch, - RolloutBatchSampleMembership, - SampleRecord, -) -from ergon_core.core.rl.rollout_service import RolloutService -from ergon_core.core.rl.rollout_types import SubmitRequest -from sqlalchemy.pool import StaticPool -from sqlmodel import Session, SQLModel, create_engine, select - - -@pytest.fixture() -def session_factory(): - engine = create_engine( - "sqlite://", - connect_args={"check_same_thread": False}, - poolclass=StaticPool, - ) - SQLModel.metadata.create_all( - engine, - tables=[ - ExperimentDefinition.__table__, - SampleRecord.__table__, - RolloutBatch.__table__, - RolloutBatchSampleMembership.__table__, - ], - ) - - def _get_session() -> Session: - return Session(engine) - - return _get_session - - -def test_rollout_submit_uses_rollout_batch_and_run_definition_without_legacy_record( - session_factory, -) -> None: - sent_events: list[inngest.Event] = [] - definition_id = uuid4() - with session_factory() as session: - session.add( - ExperimentDefinition( - id=definition_id, - benchmark_type="ci-benchmark", - name="ci definition", - metadata_json={}, - ) - ) - session.commit() - - service = RolloutService( - session_factory=session_factory, - inngest_send=sent_events.append, - tokenizer_name="unused-in-submit", - ) - - response = service.submit( - SubmitRequest( - definition_id=definition_id, - num_episodes=2, - model_target_override="openai:test", - ) - ) - - with session_factory() as session: - batch = session.get(RolloutBatch, response.batch_id) - runs = list(session.exec(select(SampleRecord)).all()) - memberships = list(session.exec(select(RolloutBatchSampleMembership)).all()) - - assert batch is not None - assert batch.experiment_id is None - assert {run.definition_id for run in runs} == {definition_id} - assert {run.model_target for run in runs} == {"openai:test"} - assert {membership.sample_id for membership in memberships} == {run.id for run in runs} - assert len(sent_events) == 2 - - -def test_rollout_submit_rejects_missing_definition_without_fallback(session_factory) -> None: - service = RolloutService( - session_factory=session_factory, - inngest_send=lambda event: None, - tokenizer_name="unused-in-submit", - ) - definition_id = uuid4() - - with pytest.raises(ValueError, match=f"Definition {definition_id} not found"): - service.submit( - SubmitRequest( - definition_id=definition_id, - num_episodes=1, - ) - ) diff --git a/ergon_infra/ergon_infra/templates/eval-only.yaml b/ergon_infra/ergon_infra/templates/eval-only.yaml index 0c9a13745..cc1529db7 100644 --- a/ergon_infra/ergon_infra/templates/eval-only.yaml +++ b/ergon_infra/ergon_infra/templates/eval-only.yaml @@ -6,7 +6,8 @@ # # sky launch ergon_infra/ergon_infra/templates/eval-only.yaml -y \ # --env CHECKPOINT=s3://bucket/checkpoint-500 \ -# --env BENCHMARK=gdpeval +# --env BENCHMARK=gdpeval \ +# --env EVAL_COMMAND='python examples/my_eval_submit.py' name: ergon-eval @@ -19,6 +20,7 @@ envs: BENCHMARK: EVALUATOR: ${EVALUATOR:-stub-rubric} EVAL_LIMIT: ${EVAL_LIMIT:-50} + EVAL_COMMAND: setup: | git clone https://github.com/DeepFlow-research/ergon.git /tmp/ergon @@ -39,9 +41,15 @@ run: | sleep 1 done - EXPERIMENT_ID="$(ergon experiment define "$BENCHMARK" \ - --model "vllm:http://localhost:8000" \ - --worker "training-stub" \ - --evaluator "$EVALUATOR" \ - --limit "$EVAL_LIMIT" 2>&1 | sed -n 's/^EXPERIMENT_ID=//p')" - ergon experiment run "$EXPERIMENT_ID" + if [ -z "$EVAL_COMMAND" ]; then + echo "EVAL_COMMAND must be a Python experiment submission command." + echo "The command receives MODEL_TARGET, BENCHMARK, EVALUATOR, EVAL_LIMIT, and CHECKPOINT." + exit 2 + fi + + export MODEL_TARGET="vllm:http://localhost:8000" + export ERGON_EVAL_MODEL_TARGET="$MODEL_TARGET" + export ERGON_EVAL_BENCHMARK="$BENCHMARK" + export ERGON_EVAL_EVALUATOR="$EVALUATOR" + export ERGON_EVAL_LIMIT="$EVAL_LIMIT" + eval "$EVAL_COMMAND" diff --git a/ergon_infra/ergon_infra/templates/train-trl-multinode.yaml b/ergon_infra/ergon_infra/templates/train-trl-multinode.yaml index 0bde52727..8ca8d75a0 100644 --- a/ergon_infra/ergon_infra/templates/train-trl-multinode.yaml +++ b/ergon_infra/ergon_infra/templates/train-trl-multinode.yaml @@ -7,7 +7,7 @@ # Weight sync between training and vLLM happens via NCCL automatically. # # sky launch ergon_infra/ergon_infra/templates/train-trl-multinode.yaml -y \ -# --env DEFINITION_ID= \ +# --env EXPERIMENT_ID= \ # --env MODEL=Qwen/Qwen2.5-1.5B name: ergon-training-multinode @@ -23,7 +23,7 @@ file_mounts: mode: MOUNT_CACHED envs: - DEFINITION_ID: + EXPERIMENT_ID: MODEL: BENCHMARK: ${BENCHMARK:-smoke-test} VLLM_TP_SIZE: ${VLLM_TP_SIZE:-1} @@ -60,7 +60,7 @@ run: | ergon train local \ --benchmark "$BENCHMARK" \ - --definition-id "$DEFINITION_ID" \ + --experiment-id "$EXPERIMENT_ID" \ --model "$MODEL" \ --vllm-mode server \ --vllm-server-url "http://$HEAD_IP:8000" \ diff --git a/ergon_infra/ergon_infra/templates/train-trl.yaml b/ergon_infra/ergon_infra/templates/train-trl.yaml index f7d3d8f68..8bae5ebf2 100644 --- a/ergon_infra/ergon_infra/templates/train-trl.yaml +++ b/ergon_infra/ergon_infra/templates/train-trl.yaml @@ -3,7 +3,7 @@ # file_mounts to sync local source (see /tmp/ac2-gpu-test.yaml pattern). # # sky launch ergon_infra/ergon_infra/templates/train-trl.yaml -y \ -# --env DEFINITION_ID= \ +# --env EXPERIMENT_ID= \ # --env MODEL=Qwen/Qwen2.5-7B \ # --env BENCHMARK=smoke-test @@ -19,7 +19,7 @@ file_mounts: mode: MOUNT_CACHED envs: - DEFINITION_ID: + EXPERIMENT_ID: MODEL: BENCHMARK: ${BENCHMARK:-smoke-test} VLLM_MODE: ${VLLM_MODE:-server} @@ -56,7 +56,7 @@ run: | ergon train local \ --benchmark "$BENCHMARK" \ - --definition-id "$DEFINITION_ID" \ + --experiment-id "$EXPERIMENT_ID" \ --model "$MODEL" \ --vllm-mode "$VLLM_MODE" \ --vllm-server-url "http://localhost:$VLLM_PORT" \ diff --git a/ergon_infra/ergon_infra/templates/train-with-eval.yaml b/ergon_infra/ergon_infra/templates/train-with-eval.yaml index be0c7fff5..ea43f2a8f 100644 --- a/ergon_infra/ergon_infra/templates/train-with-eval.yaml +++ b/ergon_infra/ergon_infra/templates/train-with-eval.yaml @@ -3,9 +3,10 @@ # file_mounts to sync local source (see /tmp/ac2-gpu-test.yaml pattern). # # sky jobs launch ergon_infra/ergon_infra/templates/train-with-eval.yaml -y \ -# --env DEFINITION_ID= \ +# --env EXPERIMENT_ID= \ # --env MODEL=Qwen/Qwen2.5-7B \ -# --env BENCHMARK=gdpeval +# --env BENCHMARK=gdpeval \ +# --env EVAL_COMMAND='python examples/my_eval_submit.py' name: ergon-rl-full execution: parallel @@ -22,7 +23,7 @@ file_mounts: mode: MOUNT_CACHED envs: - DEFINITION_ID: + EXPERIMENT_ID: MODEL: BENCHMARK: ${BENCHMARK:-smoke-test} @@ -35,7 +36,7 @@ setup: | run: | ergon train local \ --benchmark "$BENCHMARK" \ - --definition-id "$DEFINITION_ID" \ + --experiment-id "$EXPERIMENT_ID" \ --model "$MODEL" \ --vllm-mode server \ --output-dir /checkpoints \ @@ -55,6 +56,7 @@ file_mounts: envs: BENCHMARK: EVALUATOR: ${EVALUATOR:-stub-rubric} + EVAL_COMMAND: setup: | git clone https://github.com/DeepFlow-research/ergon.git /tmp/ergon @@ -68,4 +70,4 @@ run: | --benchmark "$BENCHMARK" \ --evaluator "$EVALUATOR" \ --poll-interval "${POLL_INTERVAL:-120}" \ - --on-checkpoint "sky launch ergon_infra/ergon_infra/templates/eval-only.yaml -y --env CHECKPOINT={path} --env BENCHMARK=$BENCHMARK" + --on-checkpoint "sky launch ergon_infra/ergon_infra/templates/eval-only.yaml -y --env CHECKPOINT={path} --env BENCHMARK=$BENCHMARK --env EVALUATOR=$EVALUATOR --env EVAL_COMMAND='$EVAL_COMMAND'" diff --git a/ergon_infra/ergon_infra/training/trl_runner.py b/ergon_infra/ergon_infra/training/trl_runner.py index 24b18fc6a..562c44ee6 100644 --- a/ergon_infra/ergon_infra/training/trl_runner.py +++ b/ergon_infra/ergon_infra/training/trl_runner.py @@ -8,7 +8,7 @@ GPU Node (this process) Ergon API (MacBook / cloud) ─────────────────────── ────────────────────────── - TRL GRPOTrainer POST /rollouts/submit + TRL GRPOTrainer POST /rollouts/experiments/{id}/rollout-batches → rollout_func() ────HTTP────────► → Inngest pipeline ← trajectories ◄──HTTP──────── → workers call vLLM → forward/backward (GPU) → eval + score diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 1bbc9f5e2..93ddcd8bc 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -5,19 +5,12 @@ """ import os -import re import socket -import subprocess from urllib.parse import urlparse import pytest from ergon_core.core.shared.settings import settings -_UUID_RE = re.compile( - r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", - re.IGNORECASE, -) - # NOTE: smoke fixture registration now lives exclusively inside the local API # composition target. # Host-side pytest is a black-box client (``_submit.py`` → HTTP) and @@ -73,106 +66,3 @@ def _require_infra(): "", ] pytest.fail("\n".join(lines)) - - -def run_benchmark( - slug: str, - *, - worker: str, - evaluator: str, - sandbox: str, - extras: str = "none", - model: str = "stub:constant", - limit: int = 1, - experiment: str = "ci", - timeout: int = 120, -) -> subprocess.CompletedProcess: - """Define and run an experiment via the ergon CLI.""" - define_cmd = [ - "ergon", - "experiment", - "define", - slug, - "--worker", - worker, - "--model", - model, - "--evaluator", - evaluator, - "--sandbox", - sandbox, - "--extras", - extras, - "--limit", - str(limit), - "--experiment", - experiment, - ] - env = {**os.environ, "PYTHONUNBUFFERED": "1"} - define = subprocess.run( - define_cmd, - capture_output=True, - text=True, - env=env, - timeout=timeout + 30, - ) - if define.returncode != 0: - return define - - definition_id = _parse_uuid_line("EXPERIMENT_ID=", define.stdout + define.stderr) - return subprocess.run( - ["ergon", "experiment", "run", definition_id, "--timeout", str(timeout)], - capture_output=True, - text=True, - env=env, - timeout=timeout + 30, - ) - - -def _parse_uuid_line(prefix: str, output: str) -> str: - for line in output.splitlines(): - if not line.startswith(prefix): - continue - match = _UUID_RE.search(line) - if match is not None: - return match.group(0) - raise AssertionError(f"missing {prefix} line in CLI output:\n{output}") - - -@pytest.fixture(scope="session") -def benchmarked(): - """Memoize `run_benchmark` calls by explicit runtime configuration. - - The stubbed E2E tests each assert against the *latest* SampleRecord; re-running - the same benchmark per-test burned ~4× subprocess launches with identical - outcomes. This fixture runs each unique config exactly once per session and - returns the cached `CompletedProcess`. - """ - cache: dict[tuple[str, str, str, str, str, str], subprocess.CompletedProcess] = {} - - def _run( - slug: str, - *, - worker: str, - evaluator: str, - sandbox: str, - extras: str = "none", - limit: int = 1, - experiment: str = "ci", - timeout: int = 120, - ) -> subprocess.CompletedProcess: - key = (slug, worker, evaluator, sandbox, extras, experiment) - if key not in cache: - cache[key] = run_benchmark( - slug, - worker=worker, - evaluator=evaluator, - sandbox=sandbox, - extras=extras, - limit=limit, - experiment=experiment, - timeout=timeout, - ) - return cache[key] - - return _run From b4401546aaa6810cc78922a8c6095bfe92e2f8c2 Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 14:59:40 +0100 Subject: [PATCH 07/15] Fix experiment ref identity in trainer candidate pool --- .../ergon_core/core/application/experiments/candidate_pool.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ergon_core/ergon_core/core/application/experiments/candidate_pool.py b/ergon_core/ergon_core/core/application/experiments/candidate_pool.py index aa6239139..60833a3ba 100644 --- a/ergon_core/ergon_core/core/application/experiments/candidate_pool.py +++ b/ergon_core/ergon_core/core/application/experiments/candidate_pool.py @@ -140,7 +140,7 @@ def reserve_sample_pool_entries_for_sampler( invocation = record_sampler_invocation( session=session, experiment_ref=ExperimentRef( - id=experiment.id, + experiment_id=experiment.id, name=experiment.name, environment_ids={}, created_at=experiment.created_at, From b6a180e04dd694930afefb67d7676d1f5ffc2eda Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 15:43:20 +0100 Subject: [PATCH 08/15] Preserve experiment identity on trainer rollout batches --- ergon_core/ergon_core/core/rl/rollout_service.py | 1 + ergon_core/ergon_core/core/rl/rollout_types.py | 2 ++ ergon_core/tests/unit/rest_api/test_rollouts_di.py | 4 ++-- .../tests/unit/rl/test_rollout_experiment_submission.py | 2 ++ 4 files changed, 7 insertions(+), 2 deletions(-) diff --git a/ergon_core/ergon_core/core/rl/rollout_service.py b/ergon_core/ergon_core/core/rl/rollout_service.py index 0ead10e2e..65cc70653 100644 --- a/ergon_core/ergon_core/core/rl/rollout_service.py +++ b/ergon_core/ergon_core/core/rl/rollout_service.py @@ -159,6 +159,7 @@ async def submit_experiment_batch( summary = self.create_rollout_batch( session, sample_ids=sample_ids, + experiment_id=request.experiment_id, sampler_invocation_id=invocation.id, ) session.commit() diff --git a/ergon_core/ergon_core/core/rl/rollout_types.py b/ergon_core/ergon_core/core/rl/rollout_types.py index 95bea5036..9593b78d2 100644 --- a/ergon_core/ergon_core/core/rl/rollout_types.py +++ b/ergon_core/ergon_core/core/rl/rollout_types.py @@ -30,6 +30,8 @@ class TrainingRolloutRequest(BaseModel): class RolloutBatchSummary(BaseModel): """Durable trainer batch membership exposed by sample id.""" + model_config = ConfigDict(alias_generator=_to_camel, populate_by_name=True) + batch_id: UUID sample_ids: list[UUID] status: BatchStatus diff --git a/ergon_core/tests/unit/rest_api/test_rollouts_di.py b/ergon_core/tests/unit/rest_api/test_rollouts_di.py index 89bd199f2..f8c653c74 100644 --- a/ergon_core/tests/unit/rest_api/test_rollouts_di.py +++ b/ergon_core/tests/unit/rest_api/test_rollouts_di.py @@ -65,8 +65,8 @@ def test_rollout_batch_route_exposes_sample_ids() -> None: assert resp.status_code == 200 body = resp.json() - assert body["sample_ids"] == [str(app.state.rollout_service.sample_id)] - assert "run_ids" not in body + assert body["sampleIds"] == [str(app.state.rollout_service.sample_id)] + assert "runIds" not in body def test_sync_weights_gets_vllm_manager_from_app_state() -> None: diff --git a/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py b/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py index c090dfecf..90efa095f 100644 --- a/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py +++ b/ergon_core/tests/unit/rl/test_rollout_experiment_submission.py @@ -16,6 +16,7 @@ async def submit_experiment_batch(self, request): batch_id=uuid4(), sample_ids=[uuid4()], status=BatchStatus.PENDING, + experiment_id=request.experiment_id, sampler_invocation_id=uuid4(), ) @@ -40,6 +41,7 @@ def test_experiment_rollout_route_accepts_camel_case_training_request() -> None: ) assert response.status_code == 202 + assert response.json()["experimentId"] == str(experiment_id) assert service.request.experiment_id == experiment_id assert service.request.k == 32 assert service.request.sampler == "random" From 2a1bb44f4e3f3aea869bcbe0360d449d1a63c9b7 Mon Sep 17 00:00:00 2001 From: Charlie Masters <69640669+cm2435@users.noreply.github.com> Date: Wed, 27 May 2026 23:15:00 +0100 Subject: [PATCH 09/15] stabilize sample runtime stack --- .../scripts/generate-event-contracts.mjs | 9 - .../app/samples/[sampleId]/detail/page.tsx | 21 - .../src/app/samples/[sampleId]/page.tsx | 13 +- ergon-dashboard/src/app/samples/page.tsx | 22 +- .../components/experiments/SampleTable.tsx | 2 +- .../sampleRunMetricExplorerModel.test.ts | 4 +- .../components/indexes/SampleIndexTable.tsx | 13 +- .../components/sample/SampleWorkspacePage.tsx | 10 +- .../sample/replayNavigation.test.ts | 4 +- .../src/components/sample/replayNavigation.ts | 4 +- .../sample/useSampleDisplayState.ts | 10 +- .../sample/useSampleKeyboardShortcuts.ts | 4 +- .../activity/buildSampleActivities.test.ts | 8 +- .../activity/buildSampleActivities.ts | 6 +- .../components/ActivityStackTimeline.tsx | 4 +- .../features/activity/goldenFixture.test.ts | 4 +- .../activity/snapshotSequence.test.ts | 4 +- .../src/features/activity/snapshotSequence.ts | 6 +- .../src/features/evaluation/selectors.test.ts | 2 +- .../graph/components/MutationTimeline.tsx | 4 +- .../graph/contracts/graphMutations.test.ts | 18 +- .../graph/contracts/graphMutations.ts | 12 +- .../features/graph/hooks/useGraphMutations.ts | 6 +- .../graph/layout/goldenLayout.test.ts | 4 +- .../graph/state/graphMutationReducer.ts | 22 +- .../events/DashboardWorkflowStartedEvent.ts | 2 +- .../DashboardWorkflowStartedEvent.schema.json | 18 - .../src/generated/rest/contracts.ts | 46 +- .../src/generated/rest/openapi.json | 82 ++-- .../src/hooks/useSampleWorkspaceState.ts | 8 +- .../src/inngest/functions/index.ts | 10 +- ergon-dashboard/src/lib/contracts/events.ts | 128 +++++- .../src/lib/sample-state/hydrate.ts | 2 +- .../src/lib/server-data/samples.ts | 10 +- ergon-dashboard/src/lib/socket/server.ts | 6 +- ergon-dashboard/src/lib/state/store.ts | 6 +- .../src/lib/testing/dashboardHarness.ts | 4 +- ergon-dashboard/src/lib/types.ts | 16 +- .../tests/components/ExperimentDetail.test.ts | 2 +- .../tests/contracts/contracts.test.ts | 133 +++++- .../frontend-regressions.contract.test.ts | 6 +- .../contracts/server-data.contract.test.ts | 9 +- ergon-dashboard/tests/e2e/_shared/smoke.ts | 14 +- .../mas-samples/concurrent-mas-run.json | 2 +- .../tests/helpers/backendHarnessClient.ts | 14 +- .../tests/helpers/dashboardFixtures.ts | 12 +- .../tests/helpers/testHarnessClient.ts | 16 +- .../benchmarks/gdpeval/benchmark.py | 6 +- .../benchmarks/minif2f/benchmark.py | 3 +- .../benchmarks/researchrubrics/benchmark.py | 6 +- .../benchmarks/swebench_verified/benchmark.py | 6 +- .../toolkits/subagents/toolkit.py | 7 +- ergon_builtins/tests/fixtures/toy_workflow.py | 10 +- .../tools/test_subtask_lifecycle_toolkit.py | 8 +- .../tools/test_workflow_command_adapter.py | 4 +- .../unit/state/test_benchmark_contract.py | 39 +- .../workers/test_react_worker_contract.py | 1 - ergon_core/ergon_core/api/__init__.py | 2 - .../ergon_core/api/benchmark/__init__.py | 3 +- .../ergon_core/api/benchmark/requirements.py | 11 - ergon_core/ergon_core/api/worker/context.py | 11 - .../core/application/evaluation/service.py | 57 +-- .../core/application/events/runtime.py | 8 - .../core/application/experiments/__init__.py | 10 +- .../experiments/definition_writer.py | 247 ----------- .../core/application/experiments/errors.py | 34 +- .../core/application/experiments/handles.py | 5 - .../core/application/experiments/launch.py | 70 --- .../core/application/experiments/models.py | 31 -- .../core/application/experiments/service.py | 79 +--- .../core/application/runtime/events.py | 6 +- .../application/runtime/graph_repository.py | 190 -------- .../core/application/runtime/lifecycle.py | 25 +- .../core/application/runtime/models.py | 2 - .../core/application/runtime/orchestration.py | 8 - .../application/runtime/sample_identity.py | 17 - .../application/runtime/sample_lifecycle.py | 112 +---- .../application/runtime/sample_records.py | 53 --- .../core/application/runtime/task_errors.py | 6 +- .../application/runtime/task_execution.py | 74 +--- .../application/runtime/task_management.py | 30 +- .../application/samples/materialization.py | 5 +- .../application/testing/suppression_budget.py | 11 +- .../testing/test_harness_service.py | 82 ++-- .../infrastructure/http/routes/samples.py | 2 - .../http/routes/test_harness.py | 138 +++--- .../resources/persist_outputs/contract.py | 1 - .../core/jobs/sandbox/setup/contract.py | 1 - .../core/jobs/task/cancel_orphans/job.py | 3 - .../ergon_core/core/jobs/task/evaluate/job.py | 22 +- .../ergon_core/core/jobs/task/execute/job.py | 8 - .../core/jobs/task/propagate/job.py | 6 - .../core/jobs/task/worker_execute/contract.py | 1 - .../core/jobs/task/worker_execute/job.py | 8 +- .../core/jobs/workflow/complete/job.py | 3 - .../ergon_core/core/jobs/workflow/fail/job.py | 2 - .../core/jobs/workflow/start/job.py | 34 +- .../core/persistence/definitions/__init__.py | 0 .../core/persistence/definitions/models.py | 280 ------------ .../core/persistence/graph/models.py | 17 +- .../core/persistence/shared/types.py | 1 - .../core/persistence/telemetry/models.py | 39 +- .../core/views/dashboard_events/contracts.py | 7 +- .../core/views/experiments/models.py | 121 +----- .../core/views/experiments/service.py | 411 +----------------- .../ergon_core/core/views/samples/models.py | 8 +- .../ergon_core/core/views/samples/service.py | 53 +-- .../ergon_core/core/views/samples/snapshot.py | 18 +- ergon_core/migrations/env.py | 1 - .../versions/00000000_initial_v2.py | 9 - ...000002_add_sample_experiment_provenance.py | 12 - ...00003_delete_definition_runtime_columns.py | 276 ++++++++++++ .../test_experiment_persistence_roundtrip.py | 2 +- .../test_public_experiment_submit_smoke.py | 1 - .../test_submit_starts_materialized_sample.py | 34 +- .../api/worker/test_worker_context_facade.py | 14 - .../architecture/test_core_schema_sources.py | 47 +- .../test_model_field_descriptions.py | 6 +- .../test_no_deleted_v2_symbols.py | 38 ++ .../test_persistence_boundaries.py | 6 +- .../test_public_api_boundaries.py | 6 +- .../test_public_api_target_structure.py | 2 - .../test_runtime_application_ownership.py | 1 - .../architecture/test_single_alembic_head.py | 3 +- .../test_inline_evaluator_persistence.py | 92 +--- .../experiments/test_candidate_pool.py | 1 - ...est_definition_writer_inline_evaluators.py | 107 ----- .../test_experiment_persistence.py | 1 - .../experiments/test_experiment_submit.py | 1 - .../test_sample_materialization.py | 3 +- .../test_execute_task_object_bound_fanout.py | 11 +- .../test_persist_outputs_public_sandbox.py | 1 - .../jobs/test_sandbox_setup_public_sandbox.py | 1 - ...test_worker_execute_live_sandbox_attach.py | 7 +- .../tasks/test_spawn_dynamic_task_dispatch.py | 19 +- .../evaluation/test_evaluation_dto_mapping.py | 2 +- .../evaluation/test_evaluation_persistence.py | 43 +- .../test_context_event_repository.py | 13 - .../test_experiment_read_service.py | 385 ++-------------- .../read_models/test_sample_read_service.py | 48 +- .../unit/rest_api/test_samples_routes.py | 15 +- .../rest_api/test_samples_routes_existing.py | 14 +- .../tests/unit/rest_api/test_test_harness.py | 124 +++++- .../test_rollout_batch_sample_memberships.py | 22 +- .../rl/test_rollout_experiment_submission.py | 4 +- .../runtime/test_child_function_payloads.py | 10 +- .../runtime/test_cleanup_cancelled_task.py | 1 - ...st_definition_repository_and_latest_run.py | 134 ------ .../test_definition_task_payload_typing.py | 25 -- .../test_dynamic_task_evaluation_mapping.py | 2 +- .../test_execute_task_evaluator_fanout.py | 9 +- .../test_experiment_definition_service.py | 57 --- .../runtime/test_experiment_launch_service.py | 152 ------- .../unit/runtime/test_experiment_schemas.py | 14 - .../test_failed_task_sandbox_cleanup.py | 2 - .../runtime/test_graph_mutation_contracts.py | 5 +- .../runtime/test_graph_worker_identity.py | 144 ++---- .../unit/runtime/test_identity_invariants.py | 158 +++---- .../unit/runtime/test_persist_outputs.py | 1 - .../runtime/test_persist_outputs_resources.py | 1 - .../runtime/test_propagation_contracts.py | 2 - .../runtime/test_sample_annotation_events.py | 3 +- .../test_sample_graph_task_snapshot.py | 143 ++---- .../test_sample_record_missing_error.py | 18 +- .../runtime/test_sample_record_service.py | 60 --- .../tests/unit/runtime/test_sample_service.py | 60 --- .../unit/runtime/test_sample_state_replay.py | 3 +- .../unit/runtime/test_sandbox_cleanup.py | 3 - .../unit/runtime/test_spawn_dynamic_task.py | 16 +- .../runtime/test_task_execution_repository.py | 1 - .../unit/runtime/test_typed_sample_wal.py | 3 +- .../runtime/test_walkthrough_smoketest.py | 170 ++------ .../test_worker_context_containment.py | 18 +- .../unit/runtime/test_workflow_service.py | 4 - .../unit/state/test_event_schema_phase0.py | 8 +- .../tests/unit/state/test_type_invariants.py | 25 +- .../ergon_ingestion/exports/sharded.py | 2 +- .../writers/external_run_writer.py | 60 +-- .../tests/unit/test_external_run_writer.py | 9 +- .../tests/unit/test_sharded_export.py | 2 +- tests/e2e/_asserts.py | 4 +- tests/e2e/_submit.py | 8 +- tests/e2e/test_minif2f_smoke.py | 20 +- tests/e2e/test_researchrubrics_smoke.py | 20 +- tests/e2e/test_swebench_smoke.py | 20 +- tests/fixtures/smoke_components/benchmarks.py | 2 - tests/integration/propagation/_helpers.py | 14 +- .../propagation/test_propagation_blocked.py | 35 +- .../propagation/test_propagation_cancel.py | 19 +- .../test_propagation_edge_cases.py | 22 +- .../propagation/test_propagation_happy.py | 14 +- .../propagation/test_propagation_restart.py | 13 +- tests/integration/restart/_helpers.py | 6 +- .../restart/test_downstream_invalidation.py | 26 +- .../restart/test_manager_dag_scenario.py | 8 +- .../integration/restart/test_reactivation.py | 24 +- .../integration/restart/test_restart_task.py | 44 +- .../integration/smokes/test_smoke_harness.py | 41 +- .../integration/swebench_verified/conftest.py | 121 ------ 199 files changed, 1662 insertions(+), 4721 deletions(-) delete mode 100644 ergon-dashboard/src/app/samples/[sampleId]/detail/page.tsx delete mode 100644 ergon_core/ergon_core/api/benchmark/requirements.py delete mode 100644 ergon_core/ergon_core/core/application/experiments/definition_writer.py delete mode 100644 ergon_core/ergon_core/core/application/experiments/handles.py delete mode 100644 ergon_core/ergon_core/core/application/experiments/launch.py delete mode 100644 ergon_core/ergon_core/core/application/runtime/sample_identity.py delete mode 100644 ergon_core/ergon_core/core/persistence/definitions/__init__.py delete mode 100644 ergon_core/ergon_core/core/persistence/definitions/models.py create mode 100644 ergon_core/migrations/versions/00000003_delete_definition_runtime_columns.py delete mode 100644 ergon_core/tests/unit/core/application/experiments/test_definition_writer_inline_evaluators.py delete mode 100644 ergon_core/tests/unit/runtime/test_definition_repository_and_latest_run.py delete mode 100644 ergon_core/tests/unit/runtime/test_definition_task_payload_typing.py delete mode 100644 ergon_core/tests/unit/runtime/test_experiment_definition_service.py delete mode 100644 ergon_core/tests/unit/runtime/test_experiment_launch_service.py delete mode 100644 ergon_core/tests/unit/runtime/test_experiment_schemas.py delete mode 100644 ergon_core/tests/unit/runtime/test_sample_record_service.py delete mode 100644 ergon_core/tests/unit/runtime/test_sample_service.py delete mode 100644 tests/integration/swebench_verified/conftest.py diff --git a/ergon-dashboard/scripts/generate-event-contracts.mjs b/ergon-dashboard/scripts/generate-event-contracts.mjs index 607f4fa4c..ec32502c0 100644 --- a/ergon-dashboard/scripts/generate-event-contracts.mjs +++ b/ergon-dashboard/scripts/generate-event-contracts.mjs @@ -41,15 +41,6 @@ function patchZod4Compat(source) { } function customEventContractSource(entry) { - if (entry.modelName === "DashboardGraphMutationEvent") { - return `import { z } from "zod"; -import { GraphMutationDtoSchema } from "@/features/graph/contracts/graphMutations"; - -export const DashboardGraphMutationEventSchema = z.object({ - mutation: GraphMutationDtoSchema, -}).catchall(z.any()); -`; - } if (entry.modelName === "DashboardSampleRuntimeEvent") { return `import { z } from "zod"; import { SampleRuntimeEventViewSchema } from "@/lib/contracts/rest"; diff --git a/ergon-dashboard/src/app/samples/[sampleId]/detail/page.tsx b/ergon-dashboard/src/app/samples/[sampleId]/detail/page.tsx deleted file mode 100644 index 13ca16277..000000000 --- a/ergon-dashboard/src/app/samples/[sampleId]/detail/page.tsx +++ /dev/null @@ -1,21 +0,0 @@ -import { notFound } from "next/navigation"; - -import { SampleDetail } from "@/components/samples/SampleDetail"; -import { loadSampleState } from "@/lib/server-data/samples"; - -interface SampleDetailPageProps { - params: Promise<{ - sampleId: string; - }>; -} - -export default async function SampleDetailPage({ params }: SampleDetailPageProps) { - const { sampleId } = await params; - const result = await loadSampleState(sampleId); - if (!result.ok) { - if (result.status === 404) notFound(); - throw new Error(`Failed to load sample ${sampleId}: ${result.status}`); - } - - return ; -} diff --git a/ergon-dashboard/src/app/samples/[sampleId]/page.tsx b/ergon-dashboard/src/app/samples/[sampleId]/page.tsx index 0b5e79c52..96769d641 100644 --- a/ergon-dashboard/src/app/samples/[sampleId]/page.tsx +++ b/ergon-dashboard/src/app/samples/[sampleId]/page.tsx @@ -1,4 +1,7 @@ -import { redirect } from "next/navigation"; +import { notFound } from "next/navigation"; + +import { SampleWorkspacePage } from "@/components/sample/SampleWorkspacePage"; +import { loadSampleSnapshot } from "@/lib/server-data/samples"; interface SamplePageProps { params: Promise<{ @@ -8,5 +11,11 @@ interface SamplePageProps { export default async function SamplePage({ params }: SamplePageProps) { const { sampleId } = await params; - redirect(`/samples/${sampleId}/detail`); + const result = await loadSampleSnapshot(sampleId); + if (!result.ok) { + if (result.status === 404) notFound(); + return ; + } + + return ; } diff --git a/ergon-dashboard/src/app/samples/page.tsx b/ergon-dashboard/src/app/samples/page.tsx index 42d8ce6ca..fac9d73f0 100644 --- a/ergon-dashboard/src/app/samples/page.tsx +++ b/ergon-dashboard/src/app/samples/page.tsx @@ -1,29 +1,33 @@ import { SampleIndexTable } from "@/components/indexes/SampleIndexTable"; import { loadSampleList, type SampleSummary } from "@/lib/server-data/samples"; -export default async function RunsPage() { - let runs: SampleSummary[] = []; +export default async function SamplesPage() { + let samples: SampleSummary[] = []; let error: string | null = null; const result = await loadSampleList({ limit: 100 }); if (result.ok) { - runs = result.data; + samples = result.data; } else { const detail = (result.body as { detail?: string })?.detail; error = detail ?? `API returned ${result.status}`; } - const runningCount = runs.filter((run) => ["executing", "evaluating"].includes(run.status)).length; - const failedCount = runs.filter((run) => run.status === "failed" || run.failed_tasks > 0).length; - const completedCount = runs.filter((run) => run.status === "completed").length; + const runningCount = samples.filter((sample) => + ["executing", "evaluating"].includes(sample.status), + ).length; + const failedCount = samples.filter( + (sample) => sample.status === "failed" || sample.failed_tasks > 0, + ).length; + const completedCount = samples.filter((sample) => sample.status === "completed").length; return (

- Run Index + Sample Index

-

Runs

+

Samples

{error ? ( @@ -47,7 +51,7 @@ export default async function RunsPage() { - +
); } diff --git a/ergon-dashboard/src/components/experiments/SampleTable.tsx b/ergon-dashboard/src/components/experiments/SampleTable.tsx index fb0ca5ad2..b72c6fb79 100644 --- a/ergon-dashboard/src/components/experiments/SampleTable.tsx +++ b/ergon-dashboard/src/components/experiments/SampleTable.tsx @@ -26,7 +26,7 @@ export function SampleTable({ samples }: { samples: ExperimentSampleSummaryView[ > {sample.sampleKey} diff --git a/ergon-dashboard/src/components/experiments/sampleRunMetricExplorerModel.test.ts b/ergon-dashboard/src/components/experiments/sampleRunMetricExplorerModel.test.ts index 4837b3f5f..1d7685e95 100644 --- a/ergon-dashboard/src/components/experiments/sampleRunMetricExplorerModel.test.ts +++ b/ergon-dashboard/src/components/experiments/sampleRunMetricExplorerModel.test.ts @@ -11,7 +11,7 @@ import { type RunMetricPoint, } from "./sampleRunMetricExplorerModel"; -const definitionId = "11111111-1111-4111-8111-111111111111"; +const experimentId = "11111111-1111-4111-8111-111111111111"; const defaultSampleId = "22222222-2222-4222-8222-222222222222"; function metrics(overrides: Partial = {}): ExperimentRunRow["metrics"] { @@ -28,7 +28,7 @@ function metrics(overrides: Partial = {}): Experime function runRow(overrides: Partial = {}): ExperimentRunRow { return { sample_id: defaultSampleId, - definition_id: definitionId, + experiment_id: experimentId, benchmark_type: "minif2f", instance_key: "sample-a", status: "completed", diff --git a/ergon-dashboard/src/components/indexes/SampleIndexTable.tsx b/ergon-dashboard/src/components/indexes/SampleIndexTable.tsx index ecd1fabe4..7d2ade3a1 100644 --- a/ergon-dashboard/src/components/indexes/SampleIndexTable.tsx +++ b/ergon-dashboard/src/components/indexes/SampleIndexTable.tsx @@ -23,7 +23,6 @@ export function SampleIndexTable({ runs }: { runs: SampleSummary[] }) { const matchesStatus = status === "all" || run.status === status; const text = [ run.name, - run.definition_name, run.experiment, run.benchmark_type, run.instance_key, @@ -45,7 +44,7 @@ export function SampleIndexTable({ runs }: { runs: SampleSummary[] }) { setQuery(event.target.value)} - placeholder="Search runs" + placeholder="Search samples" className="h-9 w-full max-w-sm rounded-md border border-[var(--line)] bg-[var(--card)] px-3 text-sm text-[var(--ink)] outline-none focus:border-[var(--ink-2)]" />