From 49dfac24baca7e3fa0580ccc45cdfe72595cc8a9 Mon Sep 17 00:00:00 2001 From: CK0607 Date: Mon, 17 Aug 2026 17:27:18 +0530 Subject: [PATCH] feat(cli): add deterministic task preflight checks Embed environment, task, grader, and reward validation in existing task and sync commands so broken definitions fail before rollout or upload. Co-authored-by: Cursor --- docs/v6/reference/cli.mdx | 15 +- hud/cli/eval.py | 45 +-- hud/cli/sync.py | 81 ++++++ hud/cli/task.py | 418 ++++++++++++++++++++++----- hud/cli/task_runtime.py | 118 ++++++++ hud/cli/tests/test_sync_export.py | 114 +++++++- hud/cli/tests/test_task.py | 243 ++++++++++++++++ hud/cli/utils/registry.py | 3 + hud/cli/utils/source.py | 39 +++ hud/cli/utils/tests/test_registry.py | 7 +- hud/cli/utils/tests/test_source.py | 11 + 11 files changed, 974 insertions(+), 120 deletions(-) create mode 100644 hud/cli/task_runtime.py create mode 100644 hud/cli/tests/test_task.py diff --git a/docs/v6/reference/cli.mdx b/docs/v6/reference/cli.mdx index dd7659443..634b20758 100644 --- a/docs/v6/reference/cli.mdx +++ b/docs/v6/reference/cli.mdx @@ -110,19 +110,24 @@ For a platform taskset, pass its name or id directly: `hud eval "My Tasks" claud ## Run a packaged image -`hud task start` / `hud task grade` attach to an env already serving locally (e.g. inside a built image, or alongside `hud serve`), or load one from source with `--source`. `hud task list` always reads from source (default `.`) - it doesn't attach. +`hud task start` / `hud task grade` attach to an env already serving locally (e.g. inside a built image, or alongside `hud serve`), or load one from source with `--source`. Both verify that the live environment exposes the requested task before running it. ```bash hud task list # what tasks are exposed +hud task list --env browser-v2 # boot a deployed env and read its live manifest hud task start fix_bug # -> the prompt (stdout) hud task grade fix_bug --answer "..." # -> the reward (stdout) +hud task grade fix_bug --dry-run # start + empty-answer grade, no agent rollout +hud task grade task/fix_bug --env env/browser-v2 --dry-run # check a deployed env ``` +`--dry-run` reports the environment, task, grader, and reward-shape phases separately. A valid low reward passes: the check verifies that grading works and returns finite values in the supported range, not that an empty answer solves the task. + | Command | Key options | |---------|-------------| -| `hud task start ` | `--source`/`-s`, `--args` (JSON), `--url`/`-u`, `--out`/`-o` | -| `hud task grade ` | `--answer`, `--answer-file`, `--source`, `--args`, `--url`, `--out` | -| `hud task list` | `--source`/`-s` | +| `hud task start ` | `--source`/`-s`, `--args` (JSON), `--url`/`-u`, `--env`, `--out`/`-o` | +| `hud task grade ` | `--answer`, `--answer-file`, `--dry-run`, `--timeout`, `--source`, `--args`, `--url`, `--env`, `--out` | +| `hud task list` | `--source`/`-s`, `--env`, `--url`/`-u` | ## Platform @@ -131,6 +136,8 @@ hud sync tasks my-taskset # publish tasks as a named taskset hud sync env # sync environment metadata ``` +Before uploading tasks, `hud sync tasks` rejects macOS archive metadata, broken links, and references to deployed environments or task ids that do not exist. `hud deploy` runs the same source-fixture checks before building. + External benchmark formats can be adapted into runnable `Taskset`s through the experimental [Harbor integration](/v6/experimental/harbor). diff --git a/hud/cli/eval.py b/hud/cli/eval.py index 8b3f4ccd1..3b773ea9e 100644 --- a/hud/cli/eval.py +++ b/hud/cli/eval.py @@ -5,7 +5,6 @@ from __future__ import annotations -import ast import asyncio import logging import os @@ -23,6 +22,7 @@ from rich import box from rich.table import Table +from hud.cli.task_runtime import spawn_target as _spawn_target from hud.cli.utils.api import require_api_key from hud.cli.utils.config import parse_key_value from hud.settings import settings @@ -686,49 +686,6 @@ def _build_agent(cfg: EvalConfig) -> Any: return cast("Any", cfg.agent_type.cls)(config=config) -def _python_defines_environment(path: Path) -> bool: - """Return True when ``path`` constructs a v6 :class:`~hud.environment.Environment`.""" - try: - tree = ast.parse(path.read_text(encoding="utf-8")) - except (OSError, SyntaxError): - return False - for node in ast.walk(tree): - if not isinstance(node, ast.Call): - continue - callee = node.func - callee_name = ( - callee.id - if isinstance(callee, ast.Name) - else callee.attr - if isinstance(callee, ast.Attribute) - else None - ) - if callee_name == "Environment": - return True - return False - - -def _spawn_target(source: Path) -> Path: - """The path the ``SubprocessRuntime`` provider serves. - - Directories and env-defining ``.py`` files are served as-is. Task-only - sources (``tasks.py`` importing from ``env.py``) resolve to a sibling - ``env.py`` or the containing directory. JSON/JSONL data files use the - surrounding directory (the env source lives next to the tasks file). - """ - resolved = source.resolve() - if resolved.is_dir(): - return resolved - if resolved.suffix != ".py": - return resolved.parent - if _python_defines_environment(resolved): - return resolved - env_py = resolved.parent / "env.py" - if env_py.is_file(): - return env_py - return resolved.parent - - def _resolve_placement(cfg: EvalConfig, source_path: Path | None, taskset: Any) -> Any: """Map the config's ``runtime`` onto a placement for ``Taskset.run``. diff --git a/hud/cli/sync.py b/hud/cli/sync.py index c0aaf9275..bad5a574e 100644 --- a/hud/cli/sync.py +++ b/hud/cli/sync.py @@ -172,6 +172,85 @@ def _warn_on_linked_environment_mismatch( ) +def _resolve_registry_environment_detail( + platform: PlatformClient, + ref: str, +) -> tuple[RegistryEnvironment | None, str | None]: + matches = [ + env + for env in resolve_registry_environments(platform, ref) + if env.name == ref or env.id == ref + ] + if not matches: + return None, f"environment {ref!r} is not deployed" + if len(matches) > 1: + return None, f"environment name {ref!r} is ambiguous" + registry_env = get_registry_environment(platform, matches[0].id) + if registry_env is None: + return None, f"environment {ref!r} is not deployed" + return registry_env, None + + +def _validate_task_manifests( + taskset: Taskset, + platform: PlatformClient, + console: HUDConsole, +) -> None: + """Fail before upload when a deployed environment cannot expose a task.""" + errors: list[str] = [] + for env_name in sorted(taskset.environment_names()): + try: + registry_env, error = _resolve_registry_environment_detail(platform, env_name) + except (HudException, ValueError) as exc: + errors.append(f"could not validate environment {env_name!r}: {exc}") + continue + if error is not None: + errors.append(error) + continue + assert registry_env is not None + manifest = registry_env.manifest + raw_tasks = manifest.get("tasks") if manifest is not None else None + if not isinstance(raw_tasks, list): + errors.append(f"environment {env_name!r} has no successful build manifest") + continue + exposed = { + task["id"] + for task in raw_tasks + if isinstance(task, dict) and isinstance(task.get("id"), str) + } + required = {task.id for task in taskset if task.env == env_name} + required.update( + task.verifier.id + for task in taskset + if task.verifier is not None and task.verifier.env == env_name + ) + missing = sorted(required - exposed) + if missing: + errors.append(f"environment {env_name!r} does not expose task(s): {', '.join(missing)}") + + if errors: + console.error("Task validation failed:") + for error in errors: + console.error(f" {error}") + console.hint("Deploy the environment or fix the task ids before syncing.") + raise typer.Exit(1) + console.success("Task environment manifests validated") + + +def _validate_source_fixtures(source: str, console: HUDConsole) -> None: + errors = [ + issue + for issue in EnvironmentSource.open(source).validate_fixture_quality() + if issue.severity == "error" + ] + if not errors: + return + console.error("Source fixture validation failed:") + for issue in errors: + console.error(f" {issue.message} ({issue.file})") + raise typer.Exit(1) + + def _fetch_remote_taskset( platform: PlatformClient, target_ref: str, @@ -310,6 +389,8 @@ def sync_tasks_command( exclude=exclude, console=hud_console, ) + _validate_source_fixtures(source, hud_console) + _validate_task_manifests(local_taskset, platform, hud_console) _warn_on_linked_environment_mismatch(local_taskset, platform, hud_console) # Creating a new taskset is only allowed when targeting an explicit name diff --git a/hud/cli/task.py b/hud/cli/task.py index 3e15470b2..54b69e48d 100644 --- a/hud/cli/task.py +++ b/hud/cli/task.py @@ -12,19 +12,32 @@ from __future__ import annotations import asyncio +import contextlib import json -import socket -from pathlib import Path -from typing import TYPE_CHECKING, Any -from urllib.parse import urlsplit +import math +from dataclasses import dataclass +from pathlib import Path # noqa: TC003 - Typer resolves annotations at runtime +from typing import TYPE_CHECKING, Any, Literal, TypeVar import typer +from hud.cli.task_runtime import ( + TaskResolutionError, + collect_taskset, + find_local_env_url, + normalize_control_url, + parse_task_args, + select_local_task, + spawn_target, +) from hud.utils.hud_console import HUDConsole if TYPE_CHECKING: + from collections.abc import Callable from contextlib import AbstractAsyncContextManager + from hud.clients import HudClient + from hud.eval import Taskset from hud.eval.runtime import Runtime hud_console = HUDConsole() @@ -35,56 +48,57 @@ ) -def _parse_args(args: str) -> dict[str, Any]: - try: - parsed = json.loads(args or "{}") - except json.JSONDecodeError as exc: - hud_console.error(f"--args must be valid JSON: {exc}") - raise typer.Exit(1) from None - if not isinstance(parsed, dict): - hud_console.error("--args must be a JSON object") - raise typer.Exit(1) - return parsed +PhaseStatus = Literal["pass", "fail", "skip"] +T = TypeVar("T") -def _collect(source: str) -> Any: - """Collect a Taskset from a source (``.py``/dir or JSON/JSONL), like ``hud eval``.""" - from hud.eval import Taskset +@dataclass(frozen=True, slots=True) +class CheckPhase: + name: str + status: PhaseStatus + detail: str + +def _resolution_or_exit(operation: Callable[[], T]) -> T: try: - return Taskset.from_file(source) - except FileNotFoundError as exc: - hud_console.error(str(exc)) - raise typer.Exit(1) from None + return operation() + except TaskResolutionError as exc: + message = str(exc) + hud_console.error(message) + raise typer.Exit(1) -def _local_env_url(port: int = 8765) -> str | None: - """Return a control-channel URL if an env is already serving locally on ``port`` - (e.g. ``hud serve``, or a built image whose CMD serves on :8765), else ``None``.""" - try: - with socket.create_connection(("127.0.0.1", port), timeout=0.25): - return f"tcp://127.0.0.1:{port}" - except OSError: - return None +def _parse_args(args: str) -> dict[str, Any]: + return _resolution_or_exit(lambda: parse_task_args(args)) + + +def _collect(source: str) -> Taskset: + """Collect a Taskset from a source (``.py``/dir or JSON/JSONL), like ``hud eval``.""" + return _resolution_or_exit(lambda: collect_taskset(source)) -def _spawn_target(source: str) -> Path: - """The path ``spawn`` serves: ``.py``/dir as-is, JSON/JSONL's parent directory.""" - resolved = Path(source).resolve() - if resolved.is_dir() or resolved.suffix == ".py": - return resolved - return resolved.parent +def _environment_name(value: str) -> str: + return value.removeprefix("env/").removeprefix("environment/") + + +def _task_id(value: str) -> str: + return value.removeprefix("task/") def _resolve( - task: str, source: str | None, url: str | None, args: dict[str, Any] + task: str, + source: str | None, + url: str | None, + env: str | None, + args: dict[str, Any], ) -> tuple[str, dict[str, Any], AbstractAsyncContextManager[Runtime]]: """Resolve ``(task_id, args, placement)``, choosing a substrate in priority order: - 1. ``--url`` — attach to that control channel; - 2. no ``--source`` and a local env already serving on :8765 — attach to it + 1. ``--env`` — boot that deployed environment through the HUD runtime; + 2. ``--url`` — attach to that control channel; + 3. no ``--source`` and a local env already serving on :8765 — attach to it (e.g. inside a built image, or alongside ``hud serve``); - 3. otherwise — introspect local source for the task id/slug, and spawn that + 4. otherwise — introspect local source for the task id/slug, and spawn that source as the substrate. The placement decision is made *here*, so this returns the acquisition @@ -94,32 +108,26 @@ def _resolve( """ from contextlib import nullcontext + from hud.eval import HUDRuntime, Task from hud.eval.runtime import Runtime, SubprocessRuntime + if sum(value is not None for value in (source, url, env)) > 1: + hud_console.error("choose only one placement: --source, --url, or --env") + raise typer.Exit(1) + if env is not None: + selected = Task(env=_environment_name(env), id=_task_id(task), args=args) + return selected.id, selected.args, HUDRuntime()(selected) + attach = url if attach is None and source is None: - attach = _local_env_url() + attach = find_local_env_url() if attach is not None: - parts = urlsplit(attach if "://" in attach else f"tcp://{attach}") - endpoint = f"tcp://{parts.hostname or '127.0.0.1'}:{parts.port or 8765}" - return task, args, nullcontext(Runtime(endpoint)) + endpoint = _resolution_or_exit(lambda: normalize_control_url(attach)) + return _task_id(task), args, nullcontext(Runtime(endpoint)) - taskset = _collect(source or ".") - if not taskset: - hud_console.error(f"No tasks found in {source or '.'}") - raise typer.Exit(1) - matches = [ - candidate - for index, (slug, candidate) in enumerate(taskset.items()) - if task in (slug, candidate.id, str(index)) - ] - if not matches: - available = ", ".join(sorted({t.id for t in taskset})) - hud_console.error(f"No task matching {task!r} (available: {available})") - raise typer.Exit(1) - selected = matches[0] - placement = SubprocessRuntime(_spawn_target(source or "."))(selected) - return selected.id, args or selected.args, placement + selected = _resolution_or_exit(lambda: select_local_task(task, source or ".", args)) + placement = SubprocessRuntime(spawn_target(source or "."))(selected) + return selected.id, selected.args, placement def _emit(result: dict[str, Any], headline: str, out: Path | None) -> None: @@ -132,11 +140,236 @@ def _emit(result: dict[str, Any], headline: str, out: Path | None) -> None: typer.echo(value if isinstance(value, str) else json.dumps(value, default=str)) +async def _require_task(client: HudClient, task_id: str) -> None: + tasks = await client.list_tasks() + available = [ + task["id"] for task in tasks if isinstance(task, dict) and isinstance(task.get("id"), str) + ] + if task_id not in available: + joined = ", ".join(available) or "" + raise TaskResolutionError(f"task {task_id!r} is not exposed by the environment ({joined})") + + +def _count_subscores(subscores: object) -> int: + if subscores is None: + return 0 + if not isinstance(subscores, list): + raise ValueError("grade subscores must be a list") + + count = 0 + for subscore in subscores: + if not isinstance(subscore, dict): + raise ValueError("each grade subscore must be an object") + count += 1 + _count_subscores(subscore.get("children")) + return count + + +def _validate_grade_result(result: dict[str, Any]) -> tuple[float, int]: + from hud.clients.client import HudProtocolError + from hud.eval.run import Grade + + try: + grade = Grade.from_dict(result) + except (HudProtocolError, ValueError) as exc: + raise ValueError(str(exc)) from None + if grade.is_error: + raise ValueError(grade.content or "grader returned isError=true") + score = grade.reward + if not math.isfinite(score): + raise ValueError(f"grade score must be finite, got {score}") + if not 0 <= score <= 1: + raise ValueError(f"grade score must be finite and within [0, 1], got {score}") + return score, _count_subscores(grade.raw.get("subscores")) + + +def _render_check(phases: list[CheckPhase]) -> None: + for phase in phases: + typer.echo(f"[{phase.status}] {phase.name:<8} {phase.detail}") + passed = all(phase.status != "fail" for phase in phases) + typer.echo(f"\nresult: {'PASS' if passed else 'FAIL'}") + if not passed: + raise typer.Exit(1) + + +def _phase_error(exc: Exception, action: str, timeout: float) -> str: + return f"{action} timed out after {timeout:g}s" if isinstance(exc, TimeoutError) else str(exc) + + +async def _dry_run_grade( + task_id: str, + task_args: dict[str, Any], + placement: AbstractAsyncContextManager[Runtime], + phase_timeout: float, +) -> tuple[list[CheckPhase], dict[str, Any] | None]: + phases: list[CheckPhase] = [] + from hud.clients import connect + + async with contextlib.AsyncExitStack() as stack: + try: + async with asyncio.timeout(phase_timeout): + runtime = await stack.enter_async_context(placement) + client = await stack.enter_async_context( + connect(runtime, ready_timeout=phase_timeout) + ) + except Exception as exc: + phases.append( + CheckPhase( + "env", + "fail", + _phase_error(exc, "environment startup", phase_timeout), + ) + ) + phases.extend( + [ + CheckPhase("task", "skip", "environment check failed"), + CheckPhase("grader", "skip", "task did not start"), + CheckPhase("reward", "skip", "grader did not run"), + ] + ) + return phases, None + + try: + tasks = await asyncio.wait_for(client.list_tasks(), phase_timeout) + except Exception as exc: + phases.append(CheckPhase("env", "fail", _phase_error(exc, "tasks.list", phase_timeout))) + phases.extend( + [ + CheckPhase("task", "skip", "environment check failed"), + CheckPhase("grader", "skip", "task did not start"), + CheckPhase("reward", "skip", "grader did not run"), + ] + ) + return phases, None + + manifest = client.manifest + env_name = manifest.server_info.name if manifest is not None else "environment" + phases.append(CheckPhase("env", "pass", f"{env_name} ready, {len(tasks)} task(s)")) + available = { + task["id"] + for task in tasks + if isinstance(task, dict) and isinstance(task.get("id"), str) + } + if task_id not in available: + phases.extend( + [ + CheckPhase("task", "fail", f"task {task_id!r} is not exposed"), + CheckPhase("grader", "skip", "task did not start"), + CheckPhase("reward", "skip", "grader did not run"), + ] + ) + return phases, None + + session_active = False + try: + try: + session_active = True + started = await asyncio.wait_for( + client.start_task(task_id, task_args), phase_timeout + ) + except Exception as exc: + phases.append( + CheckPhase( + "task", + "fail", + _phase_error(exc, "tasks.start", phase_timeout), + ) + ) + phases.extend( + [ + CheckPhase("grader", "skip", "task did not start"), + CheckPhase("reward", "skip", "grader did not run"), + ] + ) + return phases, None + + prompt = started.get("prompt") + detail = ( + f"start returned prompt ({len(prompt)} chars)" + if isinstance(prompt, str) + else "start returned a prompt" + ) + phases.append(CheckPhase("task", "pass", detail)) + + try: + graded = await asyncio.wait_for(client.grade({"answer": ""}), phase_timeout) + session_active = False + except Exception as exc: + phases.append( + CheckPhase( + "grader", + "fail", + _phase_error(exc, "tasks.grade", phase_timeout), + ) + ) + phases.append(CheckPhase("reward", "skip", "grader failed")) + return phases, None + + if graded.get("isError") is True: + detail = str(graded.get("content") or "grader returned isError=true") + phases.append(CheckPhase("grader", "fail", detail)) + phases.append(CheckPhase("reward", "skip", "grader failed")) + return phases, graded + phases.append(CheckPhase("grader", "pass", "empty-answer grade completed")) + try: + score, subscore_count = _validate_grade_result(graded) + except ValueError as exc: + phases.append(CheckPhase("reward", "fail", str(exc))) + return phases, graded + + detail = f"score {score:g} is valid" + if subscore_count: + detail += f", {subscore_count} subscore(s) valid" + phases.append(CheckPhase("reward", "pass", detail)) + return phases, graded + finally: + if session_active: + with contextlib.suppress(Exception): + await asyncio.wait_for(client.cancel(), 2.0) + + @task_app.command("list") def list_command( source: str = typer.Option(".", "--source", "-s", help="Env source (.py/dir/JSON)."), + env: str | None = typer.Option( + None, + "--env", + help="Boot this deployed environment and list its live task manifest.", + ), + url: str | None = typer.Option( + None, + "--url", + "-u", + help="List tasks from a served control channel instead of local source.", + ), ) -> None: - """List the tasks (slug + task id + args) exposed by a source.""" + """List tasks from local definitions or a live environment.""" + if env is not None and url is not None: + hud_console.error("choose either --env or --url") + raise typer.Exit(1) + if env is not None or url is not None: + from hud.eval import HUDRuntime, Runtime, Task + + env_name = _environment_name(env) if env else "attached" + task = Task(env=env_name, id="__hud_task_list__") + provider = ( + HUDRuntime() + if env is not None + else Runtime(_resolution_or_exit(lambda: normalize_control_url(url or ""))) + ) + + async def _run() -> list[dict[str, Any]]: + from hud.clients import connect + + async with provider(task) as runtime, connect(runtime) as client: + return await client.list_tasks() + + for task_manifest in asyncio.run(_run()): + typer.echo( + f"{task_manifest.get('id', '')}\t" + f"{task_manifest.get('description', '')}".rstrip() + ) + return + for slug, task in _collect(source).items(): args = f" {json.dumps(task.args)}" if task.args else "" typer.echo(f"{slug}\t{task.id}{args}") @@ -152,12 +385,17 @@ def start_command( url: str | None = typer.Option( None, "--url", "-u", help="Attach to a served control channel instead of loading source." ), + env: str | None = typer.Option( + None, + "--env", + help="Boot this deployed environment instead of loading local source.", + ), out: Path | None = typer.Option( # noqa: B008 None, "--out", "-o", help="Write the prompt here instead of stdout." ), ) -> None: """Start a task and return its prompt (the env's first yield).""" - task_id, task_args, placement = _resolve(task, source, url, _parse_args(args)) + task_id, task_args, placement = _resolve(task, source, url, env, _parse_args(args)) async def _run() -> dict[str, Any]: from hud.clients import connect @@ -165,9 +403,11 @@ async def _run() -> dict[str, Any]: # Start and disconnect without grading; an attached (persistent) env keeps # the session for a later `hud task grade` to resume. async with placement as runtime, connect(runtime) as client: + await _require_task(client, task_id) return await client.start_task(task_id, task_args) - _emit(asyncio.run(_run()), "prompt", out) + result = _resolution_or_exit(lambda: asyncio.run(_run())) + _emit(result, "prompt", out) @task_app.command("grade") @@ -184,27 +424,65 @@ def grade_command( url: str | None = typer.Option( None, "--url", "-u", help="Attach to a served control channel instead of loading source." ), + env: str | None = typer.Option( + None, + "--env", + help="Boot this deployed environment instead of loading local source.", + ), out: Path | None = typer.Option( # noqa: B008 None, "--out", "-o", help="Write the full JSON result here (else print the reward)." ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Start fresh, grade an empty answer, and report deterministic lifecycle checks.", + ), + timeout: float = typer.Option( + 120.0, + "--timeout", + min=0.1, + help="Per-phase timeout in seconds for --dry-run.", + ), ) -> None: """Grade an answer for a task and return its reward.""" + if dry_run and (answer or answer_file is not None): + hud_console.error("--dry-run uses an empty answer; omit --answer and --answer-file") + raise typer.Exit(1) answer_text = answer_file.read_text(encoding="utf-8") if answer_file is not None else answer - task_id, task_args, placement = _resolve(task, source, url, _parse_args(args)) + task_id, task_args, placement = _resolve(task, source, url, env, _parse_args(args)) + + if dry_run: + phases, result = asyncio.run(_dry_run_grade(task_id, task_args, placement, timeout)) + if out is not None and result is not None: + _emit(result, "score", out) + _render_check(phases) + return async def _run() -> dict[str, Any]: from hud.clients import connect from hud.clients.client import HudProtocolError async with placement as runtime, connect(runtime) as client: + session_active = False try: - return await client.grade({"answer": answer_text}) # resume a prior start - except HudProtocolError: - # No held session: run the whole lifecycle here (start then grade). - await client.start_task(task_id, task_args) - return await client.grade({"answer": answer_text}) - - _emit(asyncio.run(_run()), "score", out) + try: + result = await client.grade({"answer": answer_text}) # resume a prior start + except HudProtocolError as exc: + if exc.code != -32600 or exc.message != "no task in progress": + raise + await _require_task(client, task_id) + session_active = True + await client.start_task(task_id, task_args) + result = await client.grade({"answer": answer_text}) + session_active = False + return result + finally: + if session_active: + with contextlib.suppress(Exception): + await asyncio.wait_for(client.cancel(), 2.0) + + result = _resolution_or_exit(lambda: asyncio.run(_run())) + _emit(result, "score", out) __all__ = ["task_app"] diff --git a/hud/cli/task_runtime.py b/hud/cli/task_runtime.py new file mode 100644 index 000000000..429d27460 --- /dev/null +++ b/hud/cli/task_runtime.py @@ -0,0 +1,118 @@ +"""Shared task resolution for CLI commands.""" + +from __future__ import annotations + +import ast +import json +import socket +from pathlib import Path +from typing import TYPE_CHECKING, Any +from urllib.parse import urlsplit + +if TYPE_CHECKING: + from hud.eval import Task, Taskset + + +class TaskResolutionError(ValueError): + """The requested task or source cannot be resolved.""" + + +def parse_task_args(value: str) -> dict[str, Any]: + try: + parsed = json.loads(value or "{}") + except json.JSONDecodeError as exc: + raise TaskResolutionError(f"--args must be valid JSON: {exc}") from None + if not isinstance(parsed, dict): + raise TaskResolutionError("--args must be a JSON object") + return parsed + + +def collect_taskset(source: str) -> Taskset: + from hud.eval import Taskset + + try: + return Taskset.from_file(source) + except (FileNotFoundError, ValueError) as exc: + raise TaskResolutionError(str(exc)) from None + + +def find_local_env_url(port: int = 8765) -> str | None: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.25): + return f"tcp://127.0.0.1:{port}" + except OSError: + return None + + +def _python_defines_environment(path: Path) -> bool: + try: + tree = ast.parse(path.read_text(encoding="utf-8")) + except (OSError, SyntaxError): + return False + return any( + isinstance(node, ast.Call) + and ( + (isinstance(node.func, ast.Name) and node.func.id == "Environment") + or (isinstance(node.func, ast.Attribute) and node.func.attr == "Environment") + ) + for node in ast.walk(tree) + ) + + +def spawn_target(source: str | Path) -> Path: + resolved = Path(source).resolve() + if resolved.is_dir(): + return resolved + if resolved.suffix != ".py": + return resolved.parent + if _python_defines_environment(resolved): + return resolved + env_py = resolved.parent / "env.py" + return env_py if env_py.is_file() else resolved.parent + + +def select_local_task( + task: str, + source: str, + args: dict[str, Any], +) -> Task: + taskset = collect_taskset(source) + if not taskset: + raise TaskResolutionError(f"No tasks found in {source}") + matches = [ + candidate + for index, (slug, candidate) in enumerate(taskset.items()) + if task in (slug, candidate.id, str(index)) + ] + if not matches: + available = ", ".join(sorted({candidate.id for candidate in taskset})) + raise TaskResolutionError(f"No task matching {task!r} (available: {available})") + selected = matches[0] + if args: + selected = selected.model_copy(update={"args": args}) + return selected + + +def normalize_control_url(value: str) -> str: + parts = urlsplit(value if "://" in value else f"tcp://{value}") + if parts.scheme != "tcp": + raise TaskResolutionError("--url must use the tcp:// control-channel scheme") + if parts.hostname is None: + raise TaskResolutionError("--url must include a host") + try: + port = parts.port or 8765 + except ValueError as exc: + raise TaskResolutionError(f"--url has an invalid port: {exc}") from None + host = f"[{parts.hostname}]" if ":" in parts.hostname else parts.hostname + return f"tcp://{host}:{port}" + + +__all__ = [ + "TaskResolutionError", + "collect_taskset", + "find_local_env_url", + "normalize_control_url", + "parse_task_args", + "select_local_task", + "spawn_target", +] diff --git a/hud/cli/tests/test_sync_export.py b/hud/cli/tests/test_sync_export.py index a2be7c7b4..f05ead584 100644 --- a/hud/cli/tests/test_sync_export.py +++ b/hud/cli/tests/test_sync_export.py @@ -11,7 +11,9 @@ import hud.cli.sync as sync_module from hud.cli.sync import _write_csv from hud.cli.utils.registry import RegistryEnvironment -from hud.eval import Task +from hud.eval import Task, Taskset +from hud.utils.hud_console import HUDConsole +from hud.utils.platform import PlatformClient if TYPE_CHECKING: from pathlib import Path @@ -56,3 +58,113 @@ def closed_input(_: str) -> str: assert exc_info.value.exit_code == 0 assert "Aborted." in capsys.readouterr().err + + +def test_task_sync_validates_deployed_manifest_before_upload( + monkeypatch: pytest.MonkeyPatch, +) -> None: + taskset = Taskset("checks", [Task(env="demo", id="solve")]) + deployed = RegistryEnvironment( + id="env-1", + name="demo", + manifest={"tasks": [{"id": "solve"}]}, + ) + summary = RegistryEnvironment(id=deployed.id, name=deployed.name) + monkeypatch.setattr(sync_module, "resolve_registry_environments", lambda *_: [summary]) + monkeypatch.setattr(sync_module, "get_registry_environment", lambda *_: deployed) + + sync_module._validate_task_manifests( + taskset, + PlatformClient("https://api.example", "key"), + HUDConsole(), + ) + + +def test_task_sync_accepts_a_registry_environment_id( + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry_id = "11111111-1111-1111-1111-111111111111" + taskset = Taskset("checks", [Task(env=registry_id, id="solve")]) + deployed = RegistryEnvironment( + id=registry_id, + name="demo", + manifest={"tasks": [{"id": "solve"}]}, + ) + monkeypatch.setattr(sync_module, "get_registry_environment", lambda *_: deployed) + + sync_module._validate_task_manifests( + taskset, + PlatformClient("https://api.example", "key"), + HUDConsole(), + ) + + +def test_task_sync_rejects_unknown_task_before_upload( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + taskset = Taskset("checks", [Task(env="demo", id="missing")]) + deployed = RegistryEnvironment( + id="env-1", + name="demo", + manifest={"tasks": [{"id": "solve"}]}, + ) + summary = RegistryEnvironment(id=deployed.id, name=deployed.name) + monkeypatch.setattr(sync_module, "resolve_registry_environments", lambda *_: [summary]) + monkeypatch.setattr(sync_module, "get_registry_environment", lambda *_: deployed) + + with pytest.raises(typer.Exit): + sync_module._validate_task_manifests( + taskset, + PlatformClient("https://api.example", "key"), + HUDConsole(), + ) + + assert "does not expose task(s): missing" in capsys.readouterr().err + + +def test_task_sync_validates_verifier_task_ids( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + taskset = Taskset( + "checks", + [ + Task( + env="actor", + id="solve", + verifier=Task(env="judge", id="verify"), + ) + ], + ) + deployed = { + "env-actor": RegistryEnvironment( + id="env-actor", + name="actor", + manifest={"tasks": [{"id": "solve"}]}, + ), + "env-judge": RegistryEnvironment( + id="env-judge", + name="judge", + manifest={"tasks": [{"id": "other"}]}, + ), + } + monkeypatch.setattr( + sync_module, + "resolve_registry_environments", + lambda _, name: [RegistryEnvironment(id=f"env-{name}", name=name)], + ) + monkeypatch.setattr( + sync_module, + "get_registry_environment", + lambda _, registry_id: deployed[registry_id], + ) + + with pytest.raises(typer.Exit): + sync_module._validate_task_manifests( + taskset, + PlatformClient("https://api.example", "key"), + HUDConsole(), + ) + + assert "environment 'judge' does not expose task(s): verify" in capsys.readouterr().err diff --git a/hud/cli/tests/test_task.py b/hud/cli/tests/test_task.py new file mode 100644 index 000000000..bb1f303e9 --- /dev/null +++ b/hud/cli/tests/test_task.py @@ -0,0 +1,243 @@ +"""Behavioral coverage for deterministic ``hud task`` checks.""" + +from __future__ import annotations + +import asyncio +import contextlib +import sys +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest +from typer.testing import CliRunner + +import hud.clients as hud_clients +import hud.eval as hud_eval +from hud.cli import app + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Iterator + from pathlib import Path + + +@pytest.fixture(autouse=True) +def _clear_loaded_env_module() -> Iterator[None]: + sys.modules.pop("env", None) + yield + sys.modules.pop("env", None) + + +def _write_task_source( + tmp_path: Path, + *, + score: float, + grade_expression: str | None = None, +) -> Path: + grade = grade_expression or repr(score) + (tmp_path / "env.py").write_text( + "from hud import Environment\n\n" + "from hud.graders import EvaluationResult\n\n" + 'env = Environment("checks")\n\n' + '@env.template(id="solve")\n' + "async def solve(case: str):\n" + ' yield f"solve {case}"\n' + f" yield {grade}\n", + encoding="utf-8", + ) + tasks = tmp_path / "tasks.py" + tasks.write_text( + 'from env import solve\n\ntask = solve(case="demo")\n', + encoding="utf-8", + ) + return tasks + + +def test_dry_run_accepts_a_low_but_valid_reward(tmp_path: Path) -> None: + tasks = _write_task_source(tmp_path, score=0.0) + + result = CliRunner().invoke( + app, + ["task", "grade", "solve", "--source", str(tasks), "--dry-run"], + ) + + assert result.exit_code == 0, result.output + assert "[pass] env" in result.output + assert "[pass] task" in result.output + assert "[pass] grader" in result.output + assert "[pass] reward" in result.output + assert "score 0 is valid" in result.output + assert "result: PASS" in result.output + + +def test_dry_run_rejects_an_out_of_range_reward(tmp_path: Path) -> None: + tasks = _write_task_source(tmp_path, score=2.0) + + result = CliRunner().invoke( + app, + ["task", "grade", "solve", "--source", str(tasks), "--dry-run"], + ) + + assert result.exit_code == 1, result.output + assert "[pass] grader" in result.output + assert "[fail] reward" in result.output + assert "within [0, 1]" in result.output + assert "result: FAIL" in result.output + + +def test_dry_run_reports_grader_errors_separately(tmp_path: Path) -> None: + tasks = _write_task_source( + tmp_path, + score=0.0, + grade_expression=( + 'EvaluationResult(reward=0.0, isError=True, content="grader dependency failed")' + ), + ) + + result = CliRunner().invoke( + app, + ["task", "grade", "solve", "--source", str(tasks), "--dry-run"], + ) + + assert result.exit_code == 1, result.output + assert "[fail] grader" in result.output + assert "grader dependency failed" in result.output + assert "[skip] reward" in result.output + + +def test_start_resolves_a_task_only_source_through_its_sibling_env(tmp_path: Path) -> None: + tasks = _write_task_source(tmp_path, score=1.0) + + result = CliRunner().invoke( + app, + ["task", "start", "solve", "--source", str(tasks)], + ) + + assert result.exit_code == 0, result.output + assert result.output.strip() == "solve demo" + + +def test_start_reports_task_missing_from_live_environment(tmp_path: Path) -> None: + tasks = _write_task_source(tmp_path, score=1.0) + tasks.write_text( + 'from hud import Task\n\ntask = Task(env="checks", id="missing")\n', + encoding="utf-8", + ) + + result = CliRunner().invoke( + app, + ["task", "start", "missing", "--source", str(tasks)], + ) + + assert result.exit_code == 1 + assert "task 'missing' is not exposed by the environment (solve)" in result.output + assert "Traceback" not in result.output + + +@pytest.mark.parametrize( + ("url", "message"), + [ + ("https://example.com", "--url must use the tcp:// control-channel scheme"), + ("tcp://localhost:not-a-port", "--url has an invalid port"), + ], +) +def test_list_rejects_invalid_control_channel_url(url: str, message: str) -> None: + result = CliRunner().invoke( + app, + ["task", "list", "--url", url], + ) + + assert result.exit_code == 1 + assert message in result.output + assert "Traceback" not in result.output + + +def test_dry_run_times_out_task_start(monkeypatch: pytest.MonkeyPatch) -> None: + class Placement: + async def __aenter__(self) -> object: + return object() + + async def __aexit__(self, *exc: object) -> None: + return None + + class Runtime: + def __call__(self, task: object) -> Placement: + return Placement() + + class Client: + manifest = SimpleNamespace(server_info=SimpleNamespace(name="checks")) + + def __init__(self) -> None: + self.cancelled = False + + async def list_tasks(self) -> list[dict[str, str]]: + return [{"id": "solve"}] + + async def start_task(self, task_id: str, args: dict[str, object]) -> None: + await asyncio.sleep(1) + + async def cancel(self) -> None: + self.cancelled = True + + client = Client() + + @contextlib.asynccontextmanager + async def connect(*args: object, **kwargs: object) -> AsyncIterator[Client]: + yield client + + monkeypatch.setattr(hud_eval, "HUDRuntime", Runtime) + monkeypatch.setattr(hud_clients, "connect", connect) + + result = CliRunner().invoke( + app, + [ + "task", + "grade", + "solve", + "--env", + "checks", + "--dry-run", + "--timeout", + "0.1", + ], + ) + + assert result.exit_code == 1, result.output + assert "[fail] task" in result.output + assert "tasks.start timed out after 0.1s" in result.output + assert "[skip] grader" in result.output + assert client.cancelled + + +def test_dry_run_times_out_environment_startup(monkeypatch: pytest.MonkeyPatch) -> None: + class SlowPlacement: + async def __aenter__(self) -> object: + await asyncio.sleep(1) + return object() + + async def __aexit__(self, *exc: object) -> None: + return None + + class SlowRuntime: + def __call__(self, task: object) -> SlowPlacement: + return SlowPlacement() + + monkeypatch.setattr(hud_eval, "HUDRuntime", SlowRuntime) + + result = CliRunner().invoke( + app, + [ + "task", + "grade", + "solve", + "--env", + "checks", + "--dry-run", + "--timeout", + "0.1", + ], + ) + + assert result.exit_code == 1, result.output + assert "[fail] env" in result.output + assert "environment startup timed out after 0.1s" in result.output + assert "[skip] task" in result.output diff --git a/hud/cli/utils/registry.py b/hud/cli/utils/registry.py index d74f63824..828a6c5aa 100644 --- a/hud/cli/utils/registry.py +++ b/hud/cli/utils/registry.py @@ -17,6 +17,7 @@ class RegistryEnvironment: id: str name: str version: str = "" + manifest: dict[str, Any] | None = None @classmethod def from_record(cls, data: dict[str, Any]) -> RegistryEnvironment: @@ -26,10 +27,12 @@ def from_record(cls, data: dict[str, Any]) -> RegistryEnvironment: raise ValueError("registry environment record needs an id") latest_build = data.get("latest_build") version = latest_build.get("version") if isinstance(latest_build, dict) else None + manifest = latest_build.get("manifest") if isinstance(latest_build, dict) else None return cls( id=env_id, name=str(data.get("name") or "unnamed"), version=str(version) if version is not None else "", + manifest=manifest if isinstance(manifest, dict) else None, ) @property diff --git a/hud/cli/utils/source.py b/hud/cli/utils/source.py index 33fcbaac3..f58ef0e70 100644 --- a/hud/cli/utils/source.py +++ b/hud/cli/utils/source.py @@ -258,6 +258,45 @@ def validate(self) -> list[ValidationIssue]: issues: list[ValidationIssue] = [] issues.extend(self.validate_pyproject_references()) issues.extend(self.validate_dockerfile()) + issues.extend(self.validate_fixture_quality()) + return issues + + def validate_fixture_quality(self) -> list[ValidationIssue]: + """Reject host-generated junk and broken links from the build context.""" + issues: list[ValidationIssue] = [] + for dirpath, dirnames, filenames in os.walk(self.root): + dirnames[:] = sorted(name for name in dirnames if name not in self.SOURCE_EXCLUDE_DIRS) + directory = Path(dirpath) + + if "__MACOSX" in dirnames: + path = directory / "__MACOSX" + issues.append( + ValidationIssue( + severity="error", + message="Remove macOS archive metadata directory", + file=self.relative_path(path), + ) + ) + dirnames.remove("__MACOSX") + + for name in sorted(filenames): + path = directory / name + if name == ".DS_Store": + issues.append( + ValidationIssue( + severity="error", + message="Remove macOS Finder metadata file", + file=self.relative_path(path), + ) + ) + elif path.is_symlink() and not path.exists(): + issues.append( + ValidationIssue( + severity="error", + message="Broken symbolic link", + file=str(path.relative_to(self.root)).replace("\\", "/"), + ) + ) return issues def validate_pyproject_references(self) -> list[ValidationIssue]: diff --git a/hud/cli/utils/tests/test_registry.py b/hud/cli/utils/tests/test_registry.py index dcec53c18..0c6979028 100644 --- a/hud/cli/utils/tests/test_registry.py +++ b/hud/cli/utils/tests/test_registry.py @@ -18,13 +18,18 @@ def test_from_record_maps_registry_detail_response() -> None: env = RegistryEnvironment.from_record( - {"id": "abc123456", "name": "my-env", "latest_build": {"version": 2}} + { + "id": "abc123456", + "name": "my-env", + "latest_build": {"version": 2, "manifest": {"tasks": [{"id": "solve"}]}}, + } ) assert env.id == "abc123456" assert env.name == "my-env" assert env.short_id == "abc12345" assert env.version_label == " v2" + assert env.manifest == {"tasks": [{"id": "solve"}]} def test_resolve_accepts_uuid_without_lookup() -> None: diff --git a/hud/cli/utils/tests/test_source.py b/hud/cli/utils/tests/test_source.py index d29c7df9b..1a3841e64 100644 --- a/hud/cli/utils/tests/test_source.py +++ b/hud/cli/utils/tests/test_source.py @@ -310,6 +310,17 @@ def test_no_dockerfile_is_clean(tmp_path: Path) -> None: assert EnvironmentSource.open(tmp_path).validate_dockerfile() == [] +def test_fixture_validation_rejects_host_junk_and_broken_links(tmp_path: Path) -> None: + metadata = tmp_path / "__MACOSX" + metadata.mkdir() + _write(tmp_path / ".DS_Store", "finder") + (tmp_path / "missing-link").symlink_to(tmp_path / "missing") + + issues = EnvironmentSource.open(tmp_path).validate_fixture_quality() + + assert {issue.file for issue in issues} == {"__MACOSX", ".DS_Store", "missing-link"} + + def test_validate_environment_aggregates(tmp_path: Path) -> None: _write(tmp_path / "pyproject.toml", '[project]\nname = "x"\nlicense = {file = "LICENSE"}\n') _write(