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/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 a05e16ab1..efe917c87 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 @@ -1159,6 +1159,164 @@ 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, + expected: 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. + + ``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: + 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, 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=max(expected, len(submitted)) + ) + except (TimeoutError, asyncio.CancelledError): + for trace_id in submitted: + await self._cancel(platform, trace_id) + raise + # 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 = list_items(data) + if ( + items + and len(items) >= expected + and all( + item.get("status") in _TERMINAL_TRACE_STATUSES or item.get("end_time") + 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: Callable[[], dict[str, Any]] | None = 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 resolve_gateway_model + + try: + entry = resolve_gateway_model(model) + except Exception as exc: + logger.debug("batch submit skipped; model catalog unavailable: %s", exc) + 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: parts = urlsplit(runtime_url.rstrip("/")) diff --git a/hud/eval/taskset.py b/hud/eval/taskset.py index 2ef0d223a..3cb5dedbf 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: @@ -266,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 @@ -288,6 +298,34 @@ 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) + and (batch := _gateway_batch_spec(agent)) 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, + 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( id=uuid.uuid4().hex, diff --git a/hud/eval/tests/test_hosted.py b/hud/eval/tests/test_hosted.py index 374aa5481..9453f992a 100644 --- a/hud/eval/tests/test_hosted.py +++ b/hud/eval/tests/test_hosted.py @@ -503,3 +503,206 @@ 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"}], + [ + # 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"}, + ], + ] + ) + 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, expected=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.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 [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 + + +@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, "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) + ) + 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, 0.5, 0.5] + assert {run.group_id for run in job.runs} == {"v-a"} + + +@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] 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)]