From 479f2c91abff3f619d88ecb4ba65926d4db3af0b Mon Sep 17 00:00:00 2001 From: Daniel Mittelman Date: Thu, 9 Jul 2026 21:59:57 -0700 Subject: [PATCH 1/3] feat(eval): submit whole platform tasksets via /rollouts/run_list When a platform taskset is run wholesale on HostedRuntime with a stock gateway model, submit one batch request (/rollouts/run_list) instead of one /rollouts/submit per rollout. The server resolves the taskset's task versions itself, so every trace is born linked to its named task and the dashboard groups and labels them ("my-task (2)") instead of falling back to trace UUIDs. Anything the batch shape cannot express falls through to the existing per-rollout path: an open job, a task subset (which drops api_id), a custom model_client, agent config diverging from the type's defaults, or a model absent from the platform catalog. Fixes #488 Co-Authored-By: Claude Fable 5 --- hud/eval/runtime.py | 133 ++++++++++++++++++++++++ hud/eval/taskset.py | 39 ++++++- hud/eval/tests/test_hosted.py | 185 ++++++++++++++++++++++++++++++++++ 3 files changed, 356 insertions(+), 1 deletion(-) diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py index a05e16ab1..7d0f038f2 100644 --- a/hud/eval/runtime.py +++ b/hud/eval/runtime.py @@ -1159,6 +1159,139 @@ async def _cancel(self, platform: PlatformClient, trace_id: str) -> None: except Exception as exc: logger.warning("hosted rollout %s cancel failed: %s", trace_id, exc) + async def run_taskset( + self, + taskset_id: str, + model_id: str, + *, + group: int, + max_steps: int | None = None, + name: str | None = None, + ) -> tuple[str, list[Run]]: + """Submit a whole platform taskset as one batch via ``/rollouts/run_list``. + + Where :meth:`run` submits one rollout with the task's *content* (env, + scenario, args), this submits one request with the taskset's *identity*: + the platform resolves the current task versions itself, so every trace + is born linked to its named task and the dashboard groups and labels + them (``my-task (2)``) instead of falling back to trace UUIDs. + + Returns the server-minted job id and one folded ``Run`` per trace, in + the platform's trace order. + """ + platform = PlatformClient.from_settings() + if not platform.api_key: + raise RuntimeError("HUD-hosted execution requires HUD_API_KEY") + payload: dict[str, Any] = { + "taskset_id": taskset_id, + "model_ids": [model_id], + "group_size": group, + } + if max_steps is not None: + payload["max_steps"] = max_steps + if name: + payload["name"] = name + data = await platform.apost("/rollouts/run_list", json=payload) + job_id = str(data.get("job_id") or "") + if not job_id: + raise RuntimeError(f"run_list returned no job_id: {data}") + from hud.settings import settings as sdk_settings + + logger.info("job: %s/jobs/%s", sdk_settings.hud_web_url.rstrip("/"), job_id) + submitted = [ + str(t.get("trace_id") or "") for t in data.get("traces") or [] if t.get("trace_id") + ] + try: + states = await self._await_job_terminal(platform, job_id, expected=len(submitted)) + except asyncio.CancelledError: + for trace_id in submitted: + await self._cancel(platform, trace_id) + raise + runs: list[Run] = [] + for state in states: + trace_id = str(state.get("id") or state.get("trace_id") or "") + run = self._fold(state, trace_id) + run.trace.trace_id = trace_id + run.job_id = job_id + runs.append(run) + return job_id, runs + + async def _await_job_terminal( + self, platform: PlatformClient, job_id: str, *, expected: int + ) -> list[dict[str, Any]]: + loop = asyncio.get_event_loop() + deadline = loop.time() + self.run_timeout + while True: + data = await platform.aget(f"/jobs/{job_id}/traces") + items = data if isinstance(data, list) else (data.get("items") or []) + if ( + items + and len(items) >= expected + and all(item.get("status") in _TERMINAL_TRACE_STATUSES for item in items) + ): + return items + if loop.time() >= deadline: + raise TimeoutError( + f"hosted batch job {job_id} did not finish within {self.run_timeout:.0f}s" + ) + await asyncio.sleep(self.poll_interval) + + +def gateway_batch_spec(agent: Agent) -> tuple[str, int | None] | None: + """``(platform model id, max_steps)`` when ``agent`` fits the batch shape. + + ``/rollouts/run_list`` carries only a platform model id and a step budget, + so it can express exactly a stock gateway agent. Anything richer — a custom + ``model_client``, a non-gateway agent type, config diverging from the agent + type's defaults, or a model the platform catalog doesn't know — returns + None, and the caller keeps the per-rollout ``/rollouts/submit`` path that + carries the full agent spec. + """ + spec_of = getattr(agent, "hosted_spec", None) + if not callable(spec_of): + return None + try: + spec = spec_of() + except ValueError: + return None + config = dict(spec.get("config") or {}) + model = config.get("model") + if not isinstance(model, str) or not model: + return None + max_steps = config.get("max_steps") + try: + agent_config = agent.config # type: ignore[attr-defined] + default = type(agent_config)(model=model).model_dump( + mode="json", + exclude={"model_client", "api_key", "base_url", "hosted_tools"}, + ) + except Exception: + return None + # ``auto_respond`` is tolerated: it steers the *client* agent loop between + # steps, and the batch runner drives the platform's own loop either way — + # the same one behind the website's Batch Runs. + diverged = { + key + for key in set(config) | set(default) + if key not in ("model", "max_steps", "auto_respond") + and config.get(key) != default.get(key) + } + if diverged: + logger.debug("batch submit skipped; custom agent config: %s", sorted(diverged)) + return None + from hud.utils.gateway import list_gateway_models + + try: + models = list_gateway_models() + except Exception as exc: + logger.debug("batch submit skipped; model catalog unavailable: %s", exc) + return None + for entry in models: + if entry.id and model in (entry.model_name, entry.name, entry.id): + return entry.id, max_steps if isinstance(max_steps, int) else None + logger.debug("batch submit skipped; model %r not in the platform catalog", model) + return None + def _runtime_tunnel_ws_url(runtime_url: str, session_id: str) -> str: parts = urlsplit(runtime_url.rstrip("/")) diff --git a/hud/eval/taskset.py b/hud/eval/taskset.py index 2ef0d223a..770b661c2 100644 --- a/hud/eval/taskset.py +++ b/hud/eval/taskset.py @@ -23,7 +23,14 @@ from .job import Job, job_enter from .run import rollout -from .runtime import HostedRuntime, HUDRuntime, LocalRuntime, _declared_env, _declared_names +from .runtime import ( + HostedRuntime, + HUDRuntime, + LocalRuntime, + _declared_env, + _declared_names, + gateway_batch_spec, +) from .sync import fetch_taskset_tasks, resolve_taskset_id if TYPE_CHECKING: @@ -288,6 +295,36 @@ async def run( group_id = uuid.uuid4().hex expanded.extend((task, group_id) for _ in range(group)) + # A platform taskset run wholesale on the platform with a stock gateway + # model goes up as one batch (``/rollouts/run_list``): the server + # resolves the task versions itself, so every trace is born linked to + # its named task and the dashboard groups/labels them. Anything the + # batch shape cannot express — an open ``job``, a task subset (which + # drops ``api_id``), a custom agent config — falls through to the + # per-rollout submit below. + if ( + job is None + and task_list + and self.api_id is not None + and isinstance(runtime, HostedRuntime) + ): + batch = gateway_batch_spec(agent) + if batch is not None: + model_id, max_steps = batch + name = _job_name(self.name, task_list, group) + batch_job_id, runs = await runtime.run_taskset( + self.api_id, + model_id, + group=group, + max_steps=max_steps, + name=name, + ) + batch_job = Job( + id=batch_job_id, name=name, group=group, taskset_id=self.api_id + ) + batch_job.runs.extend(runs) + return batch_job + if job is None: job = Job( id=uuid.uuid4().hex, diff --git a/hud/eval/tests/test_hosted.py b/hud/eval/tests/test_hosted.py index 374aa5481..9761f6611 100644 --- a/hud/eval/tests/test_hosted.py +++ b/hud/eval/tests/test_hosted.py @@ -503,3 +503,188 @@ async def __anext__(self) -> bytes: cast("asyncio.StreamWriter", _Writer()), _WebSocket(), ) + + +# ─── batch submission (/rollouts/run_list) ────────────────────────────────── + + +class _FakeBatchPlatform: + """Scripted PlatformClient for the run_list path: one POST, then polls of + ``/jobs/{id}/traces`` served in order.""" + + api_key = "test-key" + + def __init__( + self, trace_pages: list[list[dict[str, Any]]], trace_ids: list[str] | None = None + ) -> None: + self.trace_pages = trace_pages + self.trace_ids = trace_ids or ["t-1", "t-2"] + self.posts: list[tuple[str, dict[str, Any]]] = [] + self.polled = 0 + + async def apost(self, path: str, *, json: Any | None = None) -> Any: + self.posts.append((path, json or {})) + return { + "job_id": "9a1b2c3d-0000-0000-0000-000000000001", + "accepted": len(self.trace_ids), + "traces": [{"trace_id": t} for t in self.trace_ids], + } + + async def aget(self, path: str, *, params: dict[str, Any] | None = None) -> Any: + page = self.trace_pages[min(self.polled, len(self.trace_pages) - 1)] + self.polled += 1 + return page + + +def _catalog_models() -> list[Any]: + from hud.utils.gateway import GatewayModelInfo + + return [ + GatewayModelInfo(id="model-uuid-1", model_name="test-model", name="Test Model"), + ] + + +def test_gateway_batch_spec_stock_agent(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.eval.runtime import gateway_batch_spec + from hud.utils import gateway + + monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) + agent = _agent() + agent.config.max_steps = 25 + + assert gateway_batch_spec(agent) == ("model-uuid-1", 25) + + +def test_gateway_batch_spec_rejects_custom_config(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.eval.runtime import gateway_batch_spec + from hud.utils import gateway + + monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) + agent = _agent() + agent.config.system_prompt = "be brief" # diverges from the type's defaults + + assert gateway_batch_spec(agent) is None + + +def test_gateway_batch_spec_rejects_unknown_model(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.eval.runtime import gateway_batch_spec + from hud.utils import gateway + + monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) + agent = OpenAIChatAgent( + OpenAIChatConfig(model="not-in-catalog", api_key="k", base_url="http://localhost") + ) + + assert gateway_batch_spec(agent) is None + + +@pytest.mark.asyncio +async def test_run_taskset_submits_batch_and_folds_runs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + platform = _FakeBatchPlatform( + [ + [{"id": "t-1", "status": "running"}, {"id": "t-2", "status": "queued"}], + [ + {"id": "t-1", "status": "completed", "reward": 1.0}, + {"id": "t-2", "status": "error", "reward": None, "error": "boom"}, + ], + ] + ) + monkeypatch.setattr( + "hud.eval.runtime.PlatformClient.from_settings", classmethod(lambda cls: platform) + ) + hosted = HostedRuntime(poll_interval=0.01) + + job_id, runs = await hosted.run_taskset( + "ts-123", "model-uuid-1", group=2, max_steps=40, name="My Batch" + ) + + assert job_id == "9a1b2c3d-0000-0000-0000-000000000001" + (path, payload) = platform.posts[0] + assert path == "/rollouts/run_list" + assert payload == { + "taskset_id": "ts-123", + "model_ids": ["model-uuid-1"], + "group_size": 2, + "max_steps": 40, + "name": "My Batch", + } + assert platform.polled == 2 # first page non-terminal, second page terminal + assert [run.reward for run in runs] == [1.0, 0.0] + assert runs[0].trace.status == "completed" + assert runs[1].trace.status == "error" + assert runs[0].trace.trace_id == "t-1" + assert runs[0].job_id == job_id + + +@pytest.mark.asyncio +async def test_taskset_run_dispatches_platform_taskset_to_run_list( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hud.eval.taskset import Taskset + from hud.utils import gateway + + platform = _FakeBatchPlatform( + [[{"id": "t-1", "status": "completed", "reward": 0.5}]], trace_ids=["t-1"] + ) + monkeypatch.setattr( + "hud.eval.runtime.PlatformClient.from_settings", classmethod(lambda cls: platform) + ) + monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) + + entered: list[str] = [] + + async def _fake_job_enter(job_id: str, **kwargs: Any) -> None: + entered.append(job_id) + + monkeypatch.setattr("hud.eval.taskset.job_enter", _fake_job_enter) + + taskset = Taskset( + "my-set", + [Task(env="coding", id="coding-bug", args={"x": 1}, slug="one")], + origin="api:ts-123", + ) + job = await taskset.run(_agent(), runtime=HostedRuntime(poll_interval=0.01), group=3) + + # One batch POST to run_list; the server-minted job comes back, and no + # client-side job is registered. + assert [p for (p, _) in platform.posts] == ["/rollouts/run_list"] + assert platform.posts[0][1]["taskset_id"] == "ts-123" + assert platform.posts[0][1]["group_size"] == 3 + assert job.id == "9a1b2c3d-0000-0000-0000-000000000001" + assert job.taskset_id == "ts-123" + assert entered == [] + assert [run.reward for run in job.runs] == [0.5] + + +@pytest.mark.asyncio +async def test_taskset_run_falls_back_to_submit_for_custom_agent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hud.eval.taskset import Taskset + from hud.utils import gateway + + platform = _FakePlatform([{"status": "completed", "reward": 1.0}]) + monkeypatch.setattr( + "hud.eval.runtime.PlatformClient.from_settings", classmethod(lambda cls: platform) + ) + monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) + + async def _fake_job_enter(job_id: str, **kwargs: Any) -> None: + return None + + monkeypatch.setattr("hud.eval.taskset.job_enter", _fake_job_enter) + + agent = _agent() + agent.config.system_prompt = "be brief" # not expressible by run_list + + taskset = Taskset( + "my-set", + [Task(env="coding", id="coding-bug", args={}, slug="one")], + origin="api:ts-123", + ) + job = await taskset.run(agent, runtime=HostedRuntime(poll_interval=0.01), group=1) + + assert [p for (p, _) in platform.posts] == ["/rollouts/submit"] + assert [run.reward for run in job.runs] == [1.0] From f4b47993fdc1f19752bff7dc0783a9b80e078462 Mon Sep 17 00:00:00 2001 From: Daniel Mittelman Date: Thu, 9 Jul 2026 22:50:43 -0700 Subject: [PATCH 2/3] fix(eval): harden the run_list batch path per review Four-lens review (conventions, correctness, concision, abstractions): - Cancel submitted traces on batch timeout, mirroring _await_terminal (previously leaked platform leases on TimeoutError). - Take the caller's tasks x group count as run_taskset(expected=...) so a lazily-populated submission response can't end the poll early; fold exactly the submitted traces (dedup strays) in submission order, and set run.group_id from each trace's task_version_id so group structure survives the batch path. Document the ordering deviation in Taskset.run's docstring. - Treat a trace with a stamped end_time as terminal even under an unrecognized status, so one odd status can't stall the whole batch. - Rename gateway_batch_spec -> _gateway_batch_spec (matches the module's private-helper convention, e.g. _declared_env/_declared_names). - Move catalog matching next to the catalog: resolve_gateway_model() in hud/utils/gateway.py; extract the platform list-envelope parsing into hud.utils.platform.list_items() and reuse it in cli/jobs.py. - Flatten the dispatch guard with a walrus; drop a dead 'or ""' branch; ruff format. Co-Authored-By: Claude Fable 5 --- hud/cli/jobs.py | 8 ++-- hud/eval/runtime.py | 69 ++++++++++++++++++++++++----------- hud/eval/taskset.py | 37 ++++++++++--------- hud/eval/tests/test_hosted.py | 42 +++++++++++++++------ hud/utils/gateway.py | 9 +++++ hud/utils/platform.py | 6 +++ 6 files changed, 115 insertions(+), 56 deletions(-) diff --git a/hud/cli/jobs.py b/hud/cli/jobs.py index 35f7dd4f5..1b45f94de 100644 --- a/hud/cli/jobs.py +++ b/hud/cli/jobs.py @@ -49,7 +49,7 @@ def jobs_command( def _list_jobs(*, json_output: bool, limit: int) -> None: - from hud.utils.platform import PlatformClient + from hud.utils.platform import PlatformClient, list_items client = PlatformClient.from_settings() try: @@ -58,7 +58,7 @@ def _list_jobs(*, json_output: bool, limit: int) -> None: console.print(f"[red]Failed to fetch jobs: {e}[/red]") raise typer.Exit(1) from e - items = data if isinstance(data, list) else (data.get("items") or []) + items = list_items(data) if json_output: console.print_json(json.dumps(items, indent=2, default=str)) @@ -99,7 +99,7 @@ def _list_jobs(*, json_output: bool, limit: int) -> None: def _show_job_traces(job_id: str, *, json_output: bool, limit: int) -> None: from hud.settings import settings - from hud.utils.platform import PlatformClient + from hud.utils.platform import PlatformClient, list_items client = PlatformClient.from_settings() try: @@ -108,7 +108,7 @@ def _show_job_traces(job_id: str, *, json_output: bool, limit: int) -> None: console.print(f"[red]Failed to fetch traces: {e}[/red]") raise typer.Exit(1) from e - items = data if isinstance(data, list) else (data.get("items") or []) + items = list_items(data) if json_output: console.print_json(json.dumps(items, indent=2, default=str)) diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py index 7d0f038f2..e45d6e784 100644 --- a/hud/eval/runtime.py +++ b/hud/eval/runtime.py @@ -47,7 +47,7 @@ from hud.telemetry.context import get_current_trace_id from hud.types import Step -from hud.utils.platform import PlatformClient +from hud.utils.platform import PlatformClient, list_items from hud.utils.process import ProcessGroup, create_process_group_exec from .run import Grade, Run, rollout @@ -1165,6 +1165,7 @@ async def run_taskset( model_id: str, *, group: int, + expected: int, max_steps: int | None = None, name: str | None = None, ) -> tuple[str, list[Run]]: @@ -1176,8 +1177,16 @@ async def run_taskset( is born linked to its named task and the dashboard groups and labels them (``my-task (2)``) instead of falling back to trace UUIDs. - Returns the server-minted job id and one folded ``Run`` per trace, in - the platform's trace order. + ``expected`` is the caller's tasks x group count — the completion bar + when the submission response under-reports its traces. ``run_timeout`` + bounds the whole batch here (the platform runs its traces + concurrently, so the wall-clock is set by the slowest rollout, not the + batch size); on timeout or cancel, every submitted trace is cancelled + platform-side before the error propagates. + + Returns the server-minted job id and one folded ``Run`` per submitted + trace in submission order, each ``group_id`` carrying the trace's task + version id (the ``group`` rollouts of one task share it). """ platform = PlatformClient.from_settings() if not platform.api_key: @@ -1197,37 +1206,55 @@ async def run_taskset( raise RuntimeError(f"run_list returned no job_id: {data}") from hud.settings import settings as sdk_settings - logger.info("job: %s/jobs/%s", sdk_settings.hud_web_url.rstrip("/"), job_id) - submitted = [ - str(t.get("trace_id") or "") for t in data.get("traces") or [] if t.get("trace_id") - ] + logger.info("job: %s/jobs/%s", sdk_settings.hud_web_url, job_id) + submitted = [str(t["trace_id"]) for t in data.get("traces") or [] if t.get("trace_id")] try: - states = await self._await_job_terminal(platform, job_id, expected=len(submitted)) - except asyncio.CancelledError: + states = await self._await_job_terminal( + platform, job_id, expected=max(expected, len(submitted)) + ) + except (TimeoutError, asyncio.CancelledError): for trace_id in submitted: await self._cancel(platform, trace_id) raise - runs: list[Run] = [] + # The poll page's order is arbitrary and may contain strays (retried or + # duplicated traces); keep one state per submitted trace, in submission + # order. Without a submission list, fall back to first-seen order. + by_id: dict[str, dict[str, Any]] = {} for state in states: trace_id = str(state.get("id") or state.get("trace_id") or "") + if trace_id and (not submitted or trace_id in set(submitted)): + by_id.setdefault(trace_id, state) + ordered = [t for t in submitted if t in by_id] if submitted else list(by_id) + runs: list[Run] = [] + for trace_id in ordered: + state = by_id[trace_id] run = self._fold(state, trace_id) run.trace.trace_id = trace_id run.job_id = job_id + version = state.get("task_version_id") + run.group_id = str(version) if version else None runs.append(run) return job_id, runs async def _await_job_terminal( self, platform: PlatformClient, job_id: str, *, expected: int ) -> list[dict[str, Any]]: + # Assumes the traces endpoint returns the full set in one page (a bare + # list today). A trace is over when its status is terminal or, for a + # status outside the known vocabulary, when its end_time is stamped — + # otherwise one unrecognized terminal status would stall the batch. loop = asyncio.get_event_loop() deadline = loop.time() + self.run_timeout while True: data = await platform.aget(f"/jobs/{job_id}/traces") - items = data if isinstance(data, list) else (data.get("items") or []) + items = list_items(data) if ( items and len(items) >= expected - and all(item.get("status") in _TERMINAL_TRACE_STATUSES for item in items) + and all( + item.get("status") in _TERMINAL_TRACE_STATUSES or item.get("end_time") + for item in items + ) ): return items if loop.time() >= deadline: @@ -1237,7 +1264,7 @@ async def _await_job_terminal( await asyncio.sleep(self.poll_interval) -def gateway_batch_spec(agent: Agent) -> tuple[str, int | None] | None: +def _gateway_batch_spec(agent: Agent) -> tuple[str, int | None] | None: """``(platform model id, max_steps)`` when ``agent`` fits the batch shape. ``/rollouts/run_list`` carries only a platform model id and a step budget, @@ -1273,24 +1300,22 @@ def gateway_batch_spec(agent: Agent) -> tuple[str, int | None] | None: diverged = { key for key in set(config) | set(default) - if key not in ("model", "max_steps", "auto_respond") - and config.get(key) != default.get(key) + if key not in ("model", "max_steps", "auto_respond") and config.get(key) != default.get(key) } if diverged: logger.debug("batch submit skipped; custom agent config: %s", sorted(diverged)) return None - from hud.utils.gateway import list_gateway_models + from hud.utils.gateway import resolve_gateway_model try: - models = list_gateway_models() + entry = resolve_gateway_model(model) except Exception as exc: logger.debug("batch submit skipped; model catalog unavailable: %s", exc) return None - for entry in models: - if entry.id and model in (entry.model_name, entry.name, entry.id): - return entry.id, max_steps if isinstance(max_steps, int) else None - logger.debug("batch submit skipped; model %r not in the platform catalog", model) - return None + if entry is None or not entry.id: + logger.debug("batch submit skipped; model %r not in the platform catalog", model) + return None + return entry.id, max_steps if isinstance(max_steps, int) else None def _runtime_tunnel_ws_url(runtime_url: str, session_id: str) -> str: diff --git a/hud/eval/taskset.py b/hud/eval/taskset.py index 770b661c2..3cb5dedbf 100644 --- a/hud/eval/taskset.py +++ b/hud/eval/taskset.py @@ -29,7 +29,7 @@ LocalRuntime, _declared_env, _declared_names, - gateway_batch_spec, + _gateway_batch_spec, ) from .sync import fetch_taskset_tasks, resolve_taskset_id @@ -273,7 +273,10 @@ async def run( an open ``job`` (:meth:`Job.start`), accumulates this batch into it instead, so a longer arc (a training session) spans many calls under one id. Returned ``job.runs`` preserves expansion order (task-major, - then group). + then group) — except on the one-request batch path (a platform + taskset run wholesale on ``HostedRuntime`` with a stock gateway + model), where runs come back in the platform's submission order, + each ``group_id`` carrying its trace's task version id. ``rollout_timeout`` is a hard per-rollout wall-clock cap (seconds) for the local (Provider) path: a rollout that exceeds it is cancelled and recorded @@ -307,23 +310,21 @@ async def run( and task_list and self.api_id is not None and isinstance(runtime, HostedRuntime) + and (batch := _gateway_batch_spec(agent)) is not None ): - batch = gateway_batch_spec(agent) - if batch is not None: - model_id, max_steps = batch - name = _job_name(self.name, task_list, group) - batch_job_id, runs = await runtime.run_taskset( - self.api_id, - model_id, - group=group, - max_steps=max_steps, - name=name, - ) - batch_job = Job( - id=batch_job_id, name=name, group=group, taskset_id=self.api_id - ) - batch_job.runs.extend(runs) - return batch_job + model_id, max_steps = batch + name = _job_name(self.name, task_list, group) + batch_job_id, runs = await runtime.run_taskset( + self.api_id, + model_id, + group=group, + expected=len(expanded), + max_steps=max_steps, + name=name, + ) + batch_job = Job(id=batch_job_id, name=name, group=group, taskset_id=self.api_id) + batch_job.runs.extend(runs) + return batch_job if job is None: job = Job( diff --git a/hud/eval/tests/test_hosted.py b/hud/eval/tests/test_hosted.py index 9761f6611..9453f992a 100644 --- a/hud/eval/tests/test_hosted.py +++ b/hud/eval/tests/test_hosted.py @@ -545,29 +545,29 @@ def _catalog_models() -> list[Any]: def test_gateway_batch_spec_stock_agent(monkeypatch: pytest.MonkeyPatch) -> None: - from hud.eval.runtime import gateway_batch_spec + from hud.eval.runtime import _gateway_batch_spec from hud.utils import gateway monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) agent = _agent() agent.config.max_steps = 25 - assert gateway_batch_spec(agent) == ("model-uuid-1", 25) + assert _gateway_batch_spec(agent) == ("model-uuid-1", 25) def test_gateway_batch_spec_rejects_custom_config(monkeypatch: pytest.MonkeyPatch) -> None: - from hud.eval.runtime import gateway_batch_spec + from hud.eval.runtime import _gateway_batch_spec from hud.utils import gateway monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) agent = _agent() agent.config.system_prompt = "be brief" # diverges from the type's defaults - assert gateway_batch_spec(agent) is None + assert _gateway_batch_spec(agent) is None def test_gateway_batch_spec_rejects_unknown_model(monkeypatch: pytest.MonkeyPatch) -> None: - from hud.eval.runtime import gateway_batch_spec + from hud.eval.runtime import _gateway_batch_spec from hud.utils import gateway monkeypatch.setattr(gateway, "list_gateway_models", _catalog_models) @@ -575,7 +575,7 @@ def test_gateway_batch_spec_rejects_unknown_model(monkeypatch: pytest.MonkeyPatc OpenAIChatConfig(model="not-in-catalog", api_key="k", base_url="http://localhost") ) - assert gateway_batch_spec(agent) is None + assert _gateway_batch_spec(agent) is None @pytest.mark.asyncio @@ -586,8 +586,17 @@ async def test_run_taskset_submits_batch_and_folds_runs( [ [{"id": "t-1", "status": "running"}, {"id": "t-2", "status": "queued"}], [ - {"id": "t-1", "status": "completed", "reward": 1.0}, - {"id": "t-2", "status": "error", "reward": None, "error": "boom"}, + # Reversed order plus a stray trace: the fold must restore + # submission order and keep only the submitted traces. + {"id": "t-stray", "status": "completed", "reward": 1.0}, + { + "id": "t-2", + "status": "error", + "reward": None, + "error": "boom", + "task_version_id": "v-a", + }, + {"id": "t-1", "status": "completed", "reward": 1.0, "task_version_id": "v-a"}, ], ] ) @@ -597,7 +606,7 @@ async def test_run_taskset_submits_batch_and_folds_runs( hosted = HostedRuntime(poll_interval=0.01) job_id, runs = await hosted.run_taskset( - "ts-123", "model-uuid-1", group=2, max_steps=40, name="My Batch" + "ts-123", "model-uuid-1", group=2, expected=2, max_steps=40, name="My Batch" ) assert job_id == "9a1b2c3d-0000-0000-0000-000000000001" @@ -611,10 +620,11 @@ async def test_run_taskset_submits_batch_and_folds_runs( "name": "My Batch", } assert platform.polled == 2 # first page non-terminal, second page terminal + assert [run.trace.trace_id for run in runs] == ["t-1", "t-2"] # submission order, stray dropped assert [run.reward for run in runs] == [1.0, 0.0] assert runs[0].trace.status == "completed" assert runs[1].trace.status == "error" - assert runs[0].trace.trace_id == "t-1" + assert [run.group_id for run in runs] == ["v-a", "v-a"] # task version id as the group key assert runs[0].job_id == job_id @@ -626,7 +636,14 @@ async def test_taskset_run_dispatches_platform_taskset_to_run_list( from hud.utils import gateway platform = _FakeBatchPlatform( - [[{"id": "t-1", "status": "completed", "reward": 0.5}]], trace_ids=["t-1"] + [ + [ + {"id": "t-1", "status": "completed", "reward": 0.5, "task_version_id": "v-a"}, + {"id": "t-2", "status": "completed", "reward": 0.5, "task_version_id": "v-a"}, + {"id": "t-3", "status": "completed", "reward": 0.5, "task_version_id": "v-a"}, + ] + ], + trace_ids=["t-1", "t-2", "t-3"], ) monkeypatch.setattr( "hud.eval.runtime.PlatformClient.from_settings", classmethod(lambda cls: platform) @@ -655,7 +672,8 @@ async def _fake_job_enter(job_id: str, **kwargs: Any) -> None: assert job.id == "9a1b2c3d-0000-0000-0000-000000000001" assert job.taskset_id == "ts-123" assert entered == [] - assert [run.reward for run in job.runs] == [0.5] + assert [run.reward for run in job.runs] == [0.5, 0.5, 0.5] + assert {run.group_id for run in job.runs} == {"v-a"} @pytest.mark.asyncio diff --git a/hud/utils/gateway.py b/hud/utils/gateway.py index 4fe5212bd..efd0ce266 100644 --- a/hud/utils/gateway.py +++ b/hud/utils/gateway.py @@ -111,3 +111,12 @@ def list_gateway_models() -> list[GatewayModelInfo]: """Models available through the HUD gateway (the platform model catalog).""" payload = PlatformClient.from_settings().get("/models") return GatewayModelsResponse.model_validate(payload).items + + +def resolve_gateway_model(model: str) -> GatewayModelInfo | None: + """The catalog entry whose id, name, or model name matches ``model``, if any.""" + wanted = normalize_gateway_model_id(model) + for entry in list_gateway_models(): + if wanted in (entry.model_name, entry.name, entry.id): + return entry + return None diff --git a/hud/utils/platform.py b/hud/utils/platform.py index 6184bbf4f..3907da940 100644 --- a/hud/utils/platform.py +++ b/hud/utils/platform.py @@ -60,3 +60,9 @@ async def aget(self, path: str, *, params: dict[str, Any] | None = None) -> Any: async def apost(self, path: str, *, json: Any | None = None) -> Any: return await make_request("POST", self.url(path), json=json, api_key=self.api_key) + + +def list_items(data: Any) -> list[dict[str, Any]]: + """Rows from a platform list response — a bare list or an ``{"items": [...]}`` envelope.""" + items = data if isinstance(data, list) else (data.get("items") or []) + return [item for item in items if isinstance(item, dict)] From 05b266f70db316fab05b644323e858280d662bd0 Mon Sep 17 00:00:00 2001 From: Daniel Mittelman Date: Fri, 10 Jul 2026 00:20:23 -0700 Subject: [PATCH 3/3] fix(eval): type the hosted_spec lookup; note batch-path ordering in Job.runs docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pyright flagged the untyped getattr result in _gateway_batch_spec, and the Job.runs reference table promised expansion order unconditionally — the hosted batch path can deviate, so say so where the contract is documented. Co-Authored-By: Claude Fable 5 --- docs/v6/reference/types.mdx | 2 +- hud/eval/runtime.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/v6/reference/types.mdx b/docs/v6/reference/types.mdx index 21c77c91c..bfd36c89c 100644 --- a/docs/v6/reference/types.mdx +++ b/docs/v6/reference/types.mdx @@ -60,7 +60,7 @@ run reports under a job, so even a single `task.run` returns a job of one. You g |--------|------|-------------| | `id` | `str` | Platform job id. | | `name` | `str` | Display name. | -| `runs` | `list[Run]` | The graded runs, in expansion order (task-major, then group). | +| `runs` | `list[Run]` | The graded runs, in expansion order (task-major, then group). When a full platform taskset is batch-submitted on `HostedRuntime`, order follows the platform's submission instead; use `results` when alignment matters. | | `group` | `int` | Rollouts per task. | | `taskset_id` | `str \| None` | Platform taskset id when the job runs a synced taskset (`Taskset.from_api`), else `None`. | | `reward` | `float` | Property: mean reward across runs (`0.0` for an empty job). | diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py index e45d6e784..efe917c87 100644 --- a/hud/eval/runtime.py +++ b/hud/eval/runtime.py @@ -1274,7 +1274,7 @@ def _gateway_batch_spec(agent: Agent) -> tuple[str, int | None] | None: None, and the caller keeps the per-rollout ``/rollouts/submit`` path that carries the full agent spec. """ - spec_of = getattr(agent, "hosted_spec", None) + spec_of: Callable[[], dict[str, Any]] | None = getattr(agent, "hosted_spec", None) if not callable(spec_of): return None try: