diff --git a/hud/cli/__init__.py b/hud/cli/__init__.py index 7775d21d1..dc3b37125 100644 --- a/hud/cli/__init__.py +++ b/hud/cli/__init__.py @@ -31,6 +31,7 @@ # --------------------------------------------------------------------------- from .cancel import cancel_command # noqa: E402 +from .check import check_command # noqa: E402 from .client import client_app # noqa: E402 from .deploy import deploy_command # noqa: E402 from .eval import eval_command # noqa: E402 @@ -49,6 +50,7 @@ app.command(name="eval")(eval_command) app.command(name="init")(init_command) app.command(name="cancel")(cancel_command) +app.command(name="check")(check_command) app.add_typer(models_app, name="models") app.add_typer(jobs_app, name="jobs") app.add_typer(trace_app, name="trace") diff --git a/hud/cli/check.py b/hud/cli/check.py new file mode 100644 index 000000000..4d186b8d2 --- /dev/null +++ b/hud/cli/check.py @@ -0,0 +1,659 @@ +"""Check a HUD Task through the existing lifecycle primitives.""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import math +import time +from dataclasses import dataclass, field +from pathlib import Path # noqa: TC003 - Typer resolves annotations at runtime +from typing import Any, Literal + +import typer +from pydantic import BaseModel, ConfigDict, Field + +from hud.cli.task_runtime import ( + TaskResolutionError, + attached_task, + find_local_env_url, + normalize_control_url, + parse_task_args, + select_local_task, + spawn_target, +) + +CheckStatus = Literal["passed", "failed", "error", "skipped"] +CheckOutcome = Literal["passed", "failed", "error"] +CheckMode = Literal["oracle", "agent", "start-only"] +ErrorKind = Literal["input", "execution"] + +_CRITERIA = ( + "resolution", + "environment_startup", + "task_startup", + "grader_execution", + "oracle_or_agent_reward", +) +_SECRET_MARKERS = ("api_key", "apikey", "authorization", "cookie", "password", "secret", "token") + + +class CheckCriterion(BaseModel): + """One stable Task lifecycle criterion.""" + + model_config = ConfigDict(extra="forbid") + + name: str + status: CheckStatus + detail: str + evidence: dict[str, Any] | None = None + + +class TaskCheckReport(BaseModel): + """Versioned output contract for ``hud check``.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: Literal["hud.task-check.v1"] + outcome: CheckOutcome + mode: CheckMode + task_id: str + runtime: str + reward: float | None = None + min_reward: float + trace_id: str | None = None + criteria: list[CheckCriterion] + error: str | None = None + error_kind: ErrorKind | None = None + duration_seconds: float = Field(ge=0) + + +@dataclass(slots=True) +class CheckRequest: + """Validated command inputs passed into the asynchronous checker.""" + + task: str + source: str | None = None + args_json: str = "{}" + url: str | None = None + runtime: str | None = None + remote: bool = False + answer: str | None = None + agent: str | None = None + model: str | None = None + start_only: bool = False + min_reward: float = 1.0 + timeout: float = 3600.0 + startup_timeout: float = 120.0 + max_steps: int = 10 + gateway: bool = False + config: list[str] = field(default_factory=list) + + @property + def mode(self) -> CheckMode: + if self.start_only: + return "start-only" + return "agent" if self.agent is not None else "oracle" + + +def _criteria_template() -> dict[str, CheckCriterion]: + return { + name: CheckCriterion(name=name, status="skipped", detail="not reached") + for name in _CRITERIA + } + + +def _safe_value(value: Any, *, string_limit: int = 2_000) -> Any: + if isinstance(value, dict): + return { + str(key): ( + "[REDACTED]" + if any(marker in str(key).lower() for marker in _SECRET_MARKERS) + else _safe_value(item, string_limit=string_limit) + ) + for key, item in value.items() + } + if isinstance(value, (list, tuple)): + return [_safe_value(item, string_limit=string_limit) for item in value[:50]] + if isinstance(value, str) and len(value) > string_limit: + return f"{value[:string_limit]}…[truncated]" + if value is None or isinstance(value, (bool, int, float, str)): + return value + return str(value) + + +def _redact_evidence(value: Any, *, max_chars: int = 4_000) -> dict[str, Any]: + """Return bounded JSON evidence with common credential fields removed.""" + safe = _safe_value(value) + if not isinstance(safe, dict): + safe = {"value": safe} + encoded = json.dumps(safe, sort_keys=True, default=str) + if len(encoded) <= max_chars: + return safe + preview_limit = max(0, max_chars - 50) + return {"truncated": True, "preview": encoded[:preview_limit]} + + +def _runtime_label(request: CheckRequest, attached_url: str | None) -> str: + if request.remote: + return "hosted" + if request.runtime == "hud": + return "hud" + if attached_url is not None: + return attached_url + return "local" + + +def _resolve(request: CheckRequest) -> tuple[Any, Any, str]: + """Resolve one Task row and its existing runtime provider.""" + from hud.eval import HostedRuntime, HUDRuntime, Runtime, SubprocessRuntime + from hud.settings import settings + + if (request.remote or request.runtime == "hud" or request.gateway) and not settings.api_key: + raise TaskResolutionError( + "HUD_API_KEY is required for HUD runtime, hosted, or gateway checks", + ) + + args = parse_task_args(request.args_json) + attached_url = request.url + if ( + attached_url is None + and request.source is None + and request.runtime is None + and not request.remote + ): + attached_url = find_local_env_url() + + if request.source is not None: + task, source_path = select_local_task(request.task, request.source, args) + elif attached_url is not None: + task, source_path = attached_task(request.task, args), None + else: + task, source_path = select_local_task(request.task, ".", args) + + if request.remote: + provider = HostedRuntime(run_timeout=request.timeout) + elif request.runtime == "hud": + provider = HUDRuntime(run_timeout=request.timeout) + elif attached_url is not None: + attached_url = normalize_control_url(attached_url) + provider = Runtime(attached_url) + else: + assert source_path is not None + provider = SubprocessRuntime( + spawn_target(source_path), + ready_timeout=request.startup_timeout, + ) + return task, provider, _runtime_label(request, attached_url) + + +def _agent(request: CheckRequest) -> Any: + from hud.cli.eval import EvalConfig, _build_agent + + try: + config = EvalConfig().merge_cli( + source=request.source, + agent=request.agent, + model=request.model, + max_steps=request.max_steps, + gateway=request.gateway, + config=request.config, + runtime=request.runtime, + remote=request.remote, + ) + config.validate_api_keys() + return _build_agent(config) + except typer.Exit: + raise TaskResolutionError("agent credentials or model configuration are invalid") from None + except (RuntimeError, ValueError) as exc: + raise TaskResolutionError(str(exc)) from None + + +async def _run_direct( + request: CheckRequest, + task: Any, + provider: Any, + criteria: dict[str, CheckCriterion], +) -> tuple[float | None, str | None]: + """Run start/grade directly against the control channel.""" + from hud.clients import connect + + async with contextlib.AsyncExitStack() as stack: + try: + async with asyncio.timeout(request.startup_timeout): + runtime = await stack.enter_async_context(provider(task)) + client = await stack.enter_async_context(connect(runtime)) + except TimeoutError as exc: + detail = f"environment did not become ready within {request.startup_timeout:g}s" + criteria["environment_startup"] = CheckCriterion( + name="environment_startup", + status="error", + detail=detail, + ) + raise TimeoutError(detail) from exc + except Exception as exc: + criteria["environment_startup"] = CheckCriterion( + name="environment_startup", + status="error", + detail=f"environment did not become ready: {exc}", + ) + raise + criteria["environment_startup"] = CheckCriterion( + name="environment_startup", + status="passed", + detail="environment control channel is ready", + evidence=_redact_evidence({"runtime_url": runtime.url}), + ) + + session_active = False + phase = "task_startup" + try: + started = await client.start_task(task.id, task.args) + session_active = True + criteria["task_startup"] = CheckCriterion( + name="task_startup", + status="passed", + detail="task started successfully", + evidence=_redact_evidence(started), + ) + + if request.start_only: + criteria["grader_execution"] = CheckCriterion( + name="grader_execution", + status="skipped", + detail="explicit --start-only check", + ) + criteria["oracle_or_agent_reward"] = CheckCriterion( + name="oracle_or_agent_reward", + status="skipped", + detail="explicit --start-only check", + ) + return None, None + + assert request.answer is not None + phase = "grader_execution" + graded = await client.grade({"answer": request.answer}) + session_active = False + raw_score = graded["score"] + if isinstance(raw_score, bool) or not isinstance(raw_score, (int, float)): + raise TypeError("grade score is not numeric") + reward = float(raw_score) + if not math.isfinite(reward): + raise ValueError("grade score is not finite") + criteria["grader_execution"] = CheckCriterion( + name="grader_execution", + status="passed", + detail="grader returned a reward", + evidence=_redact_evidence(graded), + ) + return reward, None + except BaseException as exc: + criteria[phase] = CheckCriterion( + name=phase, + status="error", + detail=( + f"task did not start: {exc}" + if phase == "task_startup" + else f"grader did not return a numeric reward: {exc}" + ), + ) + raise + finally: + if session_active: + with contextlib.suppress(Exception): + await asyncio.wait_for(client.cancel(), timeout=2.0) + + +def _run_error(run: Any) -> str | None: + if run.trace.status == "cancelled": + return run.trace.error or "rollout was cancelled" + if run.trace.status != "error" and not run.grade.is_error: + return None + return run.trace.error or run.grade.content or "rollout returned an error" + + +def _error_phase(detail: str) -> str: + lowered = detail.lower() + timeout_phase = lowered.rsplit(" during ", 1)[-1] if " during " in lowered else None + if ( + "[provisioning]" in lowered + or "[connecting]" in lowered + or timeout_phase in {"provisioning", "connecting"} + ): + return "environment_startup" + if "[starting task]" in lowered or timeout_phase == "starting task": + return "task_startup" + if ( + "[grading]" in lowered + or "[verifying]" in lowered + or timeout_phase + in { + "grading", + "verifying", + "provisioning verifier", + "cleanup", + } + ): + return "grader_execution" + return "oracle_or_agent_reward" + + +async def _run_agent( + request: CheckRequest, + task: Any, + provider: Any, + agent: Any, + criteria: dict[str, CheckCriterion], +) -> tuple[float | None, str | None]: + """Run a full agent rollout locally, through HUD, or hosted.""" + from hud.eval import HostedRuntime + from hud.eval.run import rollout + + if isinstance(provider, HostedRuntime): + job = await task.run( + agent, + runtime=provider, + rollout_timeout=request.timeout, + ) + if not job.runs: + raise RuntimeError("hosted rollout returned no run") + run = job.runs[0] + else: + run = await rollout( + task, + agent, + runtime=provider, + rollout_timeout=request.timeout, + ) + + detail = _run_error(run) + if detail is not None: + phase = _error_phase(detail) + reached = True + for name in _CRITERIA[1:4]: + if name == phase: + criteria[name] = CheckCriterion(name=name, status="error", detail=detail) + reached = False + elif reached: + criteria[name] = CheckCriterion( + name=name, + status="passed", + detail=f"{name.replace('_', ' ')} completed", + ) + if phase == "oracle_or_agent_reward": + criteria[phase] = CheckCriterion(name=phase, status="error", detail=detail) + return None, run.trace.trace_id + + for name in _CRITERIA[1:4]: + criteria[name] = CheckCriterion( + name=name, + status="passed", + detail=f"{name.replace('_', ' ')} completed", + ) + return float(run.reward), run.trace.trace_id + + +def _input_error_report( + request: CheckRequest, + *, + started_at: float, + criteria: dict[str, CheckCriterion], + detail: str, +) -> TaskCheckReport: + criteria["resolution"] = CheckCriterion( + name="resolution", + status="error", + detail=detail, + ) + return TaskCheckReport( + schema_version="hud.task-check.v1", + outcome="error", + mode=request.mode, + task_id=request.task, + runtime=_runtime_label(request, request.url), + min_reward=request.min_reward, + criteria=list(criteria.values()), + error=detail, + error_kind="input", + duration_seconds=time.monotonic() - started_at, + ) + + +async def _run_check(request: CheckRequest) -> TaskCheckReport: + started_at = time.monotonic() + criteria = _criteria_template() + agent_instance: Any = None + try: + task, provider, runtime_label = _resolve(request) + if request.mode == "agent": + agent_instance = _agent(request) + except TaskResolutionError as exc: + return _input_error_report( + request, + started_at=started_at, + criteria=criteria, + detail=str(exc), + ) + + criteria["resolution"] = CheckCriterion( + name="resolution", + status="passed", + detail="task and placement resolved", + evidence=_redact_evidence({"task_id": task.id, "environment": task.env}), + ) + reward: float | None = None + trace_id: str | None = None + error: str | None = None + try: + if request.mode == "agent": + reward, trace_id = await _run_agent( + request, + task, + provider, + agent_instance, + criteria, + ) + error = next( + ( + criterion.detail + for criterion in criteria.values() + if criterion.status == "error" + ), + None, + ) + else: + async with asyncio.timeout(request.timeout): + reward, trace_id = await _run_direct(request, task, provider, criteria) + except asyncio.CancelledError: + error = "check cancelled" + except TimeoutError as exc: + error = str(exc) or f"check did not complete within {request.timeout:g}s" + except Exception as exc: + error = str(exc) or type(exc).__name__ + + if error is not None: + if not any(criterion.status == "error" for criterion in criteria.values()): + failed_name = next( + (name for name in _CRITERIA[1:] if criteria[name].status == "skipped"), + "oracle_or_agent_reward", + ) + criteria[failed_name] = CheckCriterion( + name=failed_name, + status="error", + detail=error, + ) + outcome: CheckOutcome = "error" + error_kind: ErrorKind | None = "execution" + elif request.start_only: + outcome, error_kind = "passed", None + else: + assert reward is not None + passed = reward >= request.min_reward + criteria["oracle_or_agent_reward"] = CheckCriterion( + name="oracle_or_agent_reward", + status="passed" if passed else "failed", + detail=( + f"reward {reward:g} meets minimum {request.min_reward:g}" + if passed + else f"reward {reward:g} is below minimum {request.min_reward:g}" + ), + evidence={"reward": reward, "min_reward": request.min_reward}, + ) + outcome, error_kind = ("passed", None) if passed else ("failed", None) + + return TaskCheckReport( + schema_version="hud.task-check.v1", + outcome=outcome, + mode=request.mode, + task_id=task.id, + runtime=runtime_label, + reward=reward, + min_reward=request.min_reward, + trace_id=trace_id, + criteria=list(criteria.values()), + error=error, + error_kind=error_kind, + duration_seconds=time.monotonic() - started_at, + ) + + +def _print_report(report: TaskCheckReport, *, as_json: bool) -> None: + if as_json: + typer.echo(report.model_dump_json(indent=2)) + return + typer.echo(f"HUD Task Check: {report.outcome.upper()}") + typer.echo(f"Task: {report.task_id}") + typer.echo(f"Mode: {report.mode} Runtime: {report.runtime}") + for criterion in report.criteria: + marker = {"passed": "PASS", "failed": "FAIL", "error": "ERROR", "skipped": "SKIP"}[ + criterion.status + ] + typer.echo(f"[{marker}] {criterion.name}: {criterion.detail}") + if report.reward is not None: + typer.echo(f"Reward: {report.reward:g} (minimum {report.min_reward:g})") + if report.trace_id is not None: + typer.echo(f"Trace: {report.trace_id}") + if report.error is not None: + typer.echo(f"Error: {report.error}", err=True) + + +def _exit_code(report: TaskCheckReport) -> int: + if report.error_kind == "input": + return 2 + return {"passed": 0, "failed": 1, "error": 3}[report.outcome] + + +def check_command( + task: str = typer.Argument(..., help="Task id or local Task slug."), + source: str | None = typer.Option(None, "--source", "-s", help="Task/env source path."), + args: str = typer.Option("{}", "--args", help="Task arguments as a JSON object."), + url: str | None = typer.Option(None, "--url", help="Attach to a tcp:// control channel."), + runtime: str | None = typer.Option(None, "--runtime", help="Placement: local or hud."), + remote: bool = typer.Option(False, "--remote", help="Run the agent rollout on HUD."), + answer: str | None = typer.Option(None, "--answer", help="Direct oracle answer."), + answer_file: Path | None = typer.Option( # noqa: B008 + None, + "--answer-file", + exists=False, + dir_okay=False, + help="Read the direct oracle answer from a file.", + ), + agent: str | None = typer.Option(None, "--agent", help="Gateway agent type."), + model: str | None = typer.Option(None, "--model", help="Agent model override."), + start_only: bool = typer.Option(False, "--start-only", help="Only verify Task startup."), + min_reward: float = typer.Option(1.0, "--min-reward", min=0.0), + timeout: float = typer.Option(3600.0, "--timeout", min=0.1), + startup_timeout: float = typer.Option(120.0, "--startup-timeout", min=0.1), + max_steps: int = typer.Option(10, "--max-steps", min=1), + gateway: bool = typer.Option(False, "--gateway", help="Route local agent calls through HUD."), + config: list[str] | None = typer.Option( # noqa: B008 + None, + "--config", + help="Agent KEY=VALUE override.", + ), + json_output: bool = typer.Option(False, "--json", help="Emit hud.task-check.v1 JSON."), +) -> None: + """Check Task resolution, startup, grading, and reward.""" + strategies = int(answer is not None) + int(answer_file is not None) + int(agent is not None) + strategies += int(start_only) + if strategies != 1: + typer.echo( + "Error: choose exactly one proof strategy: --answer/--answer-file, " + "--agent, or --start-only.", + err=True, + ) + raise typer.Exit(2) + if remote and agent is None: + typer.echo("Error: --remote requires --agent.", err=True) + raise typer.Exit(2) + if sum((url is not None, runtime is not None, remote)) > 1: + typer.echo("Error: choose only one placement: --url, --runtime, or --remote.", err=True) + raise typer.Exit(2) + if runtime not in (None, "local", "hud"): + typer.echo("Error: --runtime must be local or hud.", err=True) + raise typer.Exit(2) + agent_only_options = model is not None or gateway or bool(config) or max_steps != 10 + if agent is None and agent_only_options: + typer.echo( + "Error: --model, --gateway, --config, and --max-steps require --agent.", + err=True, + ) + raise typer.Exit(2) + + if answer_file is not None: + try: + answer = answer_file.read_text(encoding="utf-8") + except OSError as exc: + typer.echo(f"Error: cannot read --answer-file: {exc}", err=True) + raise typer.Exit(2) from None + + request = CheckRequest( + task=task, + source=source, + args_json=args, + url=url, + runtime=runtime, + remote=remote, + answer=answer, + agent=agent, + model=model, + start_only=start_only, + min_reward=min_reward, + timeout=timeout, + startup_timeout=startup_timeout, + max_steps=max_steps, + gateway=gateway, + config=config or [], + ) + try: + report = asyncio.run(_run_check(request)) + except KeyboardInterrupt: + criteria = _criteria_template() + criteria["environment_startup"] = CheckCriterion( + name="environment_startup", + status="error", + detail="check cancelled", + ) + report = TaskCheckReport( + schema_version="hud.task-check.v1", + outcome="error", + mode=request.mode, + task_id=request.task, + runtime=_runtime_label(request, request.url), + min_reward=request.min_reward, + criteria=list(criteria.values()), + error="check cancelled", + error_kind="execution", + duration_seconds=0, + ) + _print_report(report, as_json=json_output) + raise typer.Exit(_exit_code(report)) + + +__all__ = [ + "CheckCriterion", + "CheckRequest", + "TaskCheckReport", + "check_command", +] diff --git a/hud/cli/task.py b/hud/cli/task.py index 3e15470b2..84a5765c5 100644 --- a/hud/cli/task.py +++ b/hud/cli/task.py @@ -13,13 +13,20 @@ import asyncio import json -import socket -from pathlib import Path +from pathlib import Path # noqa: TC003 - Typer resolves annotations at runtime from typing import TYPE_CHECKING, Any -from urllib.parse import urlsplit 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: @@ -37,23 +44,17 @@ 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}") + return parse_task_args(args) + except TaskResolutionError as exc: + hud_console.error(str(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 def _collect(source: str) -> Any: """Collect a Taskset from a source (``.py``/dir or JSON/JSONL), like ``hud eval``.""" - from hud.eval import Taskset - try: - return Taskset.from_file(source) - except FileNotFoundError as exc: + return collect_taskset(source) + except TaskResolutionError as exc: hud_console.error(str(exc)) raise typer.Exit(1) from None @@ -61,19 +62,12 @@ def _collect(source: str) -> Any: 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 + return find_local_env_url(port) 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 + return spawn_target(source) def _resolve( @@ -100,26 +94,20 @@ def _resolve( if attach is None and source is None: attach = _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}" + try: + endpoint = normalize_control_url(attach) + except TaskResolutionError as exc: + hud_console.error(str(exc)) + raise typer.Exit(1) from None return 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] + try: + selected, _ = select_local_task(task, source or ".", args) + except TaskResolutionError as exc: + hud_console.error(str(exc)) + raise typer.Exit(1) from None placement = SubprocessRuntime(_spawn_target(source or "."))(selected) - return selected.id, args or selected.args, placement + return selected.id, selected.args, placement def _emit(result: dict[str, Any], headline: str, out: Path | None) -> None: diff --git a/hud/cli/task_runtime.py b/hud/cli/task_runtime.py new file mode 100644 index 000000000..77a5c781a --- /dev/null +++ b/hud/cli/task_runtime.py @@ -0,0 +1,124 @@ +"""Shared Task resolution and placement for lifecycle CLIs.""" + +from __future__ import annotations + +import ast +import json +import socket +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + + +class TaskResolutionError(ValueError): + """The requested Task or local 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) -> Any: + 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 + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + callee = node.func + name = ( + callee.id + if isinstance(callee, ast.Name) + else callee.attr + if isinstance(callee, ast.Attribute) + else None + ) + if name == "Environment": + return True + return False + + +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], +) -> tuple[Any, Path]: + 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, Path(source) + + +def attached_task(task: str, args: dict[str, Any]) -> Any: + from hud.eval import Task + + env_name = task.split(":", 1)[0] if ":" in task else "attached" + return Task(env=env_name, id=task, args=args) + + +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") + return f"tcp://{parts.hostname}:{parts.port or 8765}" + + +__all__ = [ + "TaskResolutionError", + "attached_task", + "collect_taskset", + "find_local_env_url", + "normalize_control_url", + "parse_task_args", + "select_local_task", + "spawn_target", +] diff --git a/hud/cli/tests/test_check.py b/hud/cli/tests/test_check.py new file mode 100644 index 000000000..1337d75ba --- /dev/null +++ b/hud/cli/tests/test_check.py @@ -0,0 +1,668 @@ +"""Production contract tests for ``hud check``.""" + +from __future__ import annotations + +import asyncio +import json +import textwrap +from contextlib import asynccontextmanager +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, patch + +import pytest +from typer.testing import CliRunner + +from hud.cli import app +from hud.cli.check import ( + CheckCriterion, + CheckOutcome, + CheckRequest, + TaskCheckReport, + _criteria_template, + _redact_evidence, + _resolve, + _run_agent, + _run_check, + _run_direct, +) +from hud.eval import Runtime, SubprocessRuntime, Task + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + from pathlib import Path + +runner = CliRunner() + + +def _report( + *, + outcome: CheckOutcome = "passed", + reward: float | None = 1.0, + error: str | None = None, +) -> TaskCheckReport: + status = "passed" if outcome == "passed" else "failed" if outcome == "failed" else "error" + return TaskCheckReport( + schema_version="hud.task-check.v1", + outcome=outcome, + mode="oracle", + task_id="env:task", + runtime="local", + reward=reward, + min_reward=1.0, + trace_id="00000000-0000-4000-a000-000000000001", + criteria=[ + CheckCriterion( + name="oracle_or_agent_reward", + status=status, + detail=error or "reward evaluated", + ), + ], + error=error, + duration_seconds=0.1, + ) + + +@pytest.mark.parametrize( + "args", + [ + ["check", "task"], + ["check", "task", "--answer", "x", "--start-only"], + ["check", "task", "--answer", "x", "--agent", "claude"], + ["check", "task", "--answer", "x", "--answer-file", "answer.txt"], + ], +) +def test_check_requires_exactly_one_proof_strategy(args: list[str]) -> None: + with patch("hud.cli.check._run_check", AsyncMock()) as execute: + result = runner.invoke(app, args) + + assert result.exit_code == 2 + assert "exactly one proof strategy" in result.output.lower() + execute.assert_not_awaited() + + +def test_check_rejects_remote_direct_oracle() -> None: + with patch("hud.cli.check._run_check", AsyncMock()) as execute: + result = runner.invoke(app, ["check", "task", "--answer", "x", "--remote"]) + + assert result.exit_code == 2 + assert "--remote requires --agent" in result.output + execute.assert_not_awaited() + + +def test_check_rejects_conflicting_placement() -> None: + with patch("hud.cli.check._run_check", AsyncMock()) as execute: + result = runner.invoke( + app, + [ + "check", + "task", + "--agent", + "claude", + "--url", + "tcp://localhost:8765", + "--runtime", + "hud", + ], + ) + + assert result.exit_code == 2 + assert "choose only one placement" in result.output.lower() + execute.assert_not_awaited() + + +def test_check_rejects_agent_only_options_for_direct_proof() -> None: + with patch("hud.cli.check._run_check", AsyncMock()) as execute: + result = runner.invoke( + app, + ["check", "task", "--answer", "x", "--model", "claude-sonnet"], + ) + + assert result.exit_code == 2 + assert "require --agent" in result.output + execute.assert_not_awaited() + + +@pytest.mark.parametrize( + ("report", "expected_code"), + [ + (_report(), 0), + (_report(outcome="failed", reward=0.2), 1), + (_report(outcome="error", reward=None, error="grader unavailable"), 3), + ], +) +def test_check_exit_codes_follow_report_outcome( + report: TaskCheckReport, + expected_code: int, +) -> None: + with patch("hud.cli.check._run_check", AsyncMock(return_value=report)): + result = runner.invoke(app, ["check", "task", "--answer", "solution"]) + + assert result.exit_code == expected_code + assert "oracle_or_agent_reward" in result.output + + +def test_check_json_emits_stable_versioned_contract() -> None: + report = _report() + with patch("hud.cli.check._run_check", AsyncMock(return_value=report)): + result = runner.invoke( + app, + ["check", "task", "--answer", "solution", "--json"], + ) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["schema_version"] == "hud.task-check.v1" + assert payload["outcome"] == "passed" + assert payload["criteria"][0]["name"] == "oracle_or_agent_reward" + + +def test_check_reads_answer_file_and_forwards_runtime_options(tmp_path: Path) -> None: + answer = tmp_path / "answer.txt" + answer.write_text("file answer", encoding="utf-8") + execute = AsyncMock(return_value=_report()) + + with patch("hud.cli.check._run_check", execute): + result = runner.invoke( + app, + [ + "check", + "task", + "--source", + "env.py", + "--answer-file", + str(answer), + "--runtime", + "hud", + "--min-reward", + "0.75", + "--timeout", + "30", + "--startup-timeout", + "5", + ], + ) + + assert result.exit_code == 0 + assert execute.await_args is not None + request = execute.await_args.args[0] + assert request.answer == "file answer" + assert request.runtime == "hud" + assert request.min_reward == 0.75 + assert request.timeout == 30 + assert request.startup_timeout == 5 + + +def test_check_remote_agent_uses_hosted_strategy() -> None: + execute = AsyncMock(return_value=_report()) + with patch("hud.cli.check._run_check", execute): + result = runner.invoke( + app, + [ + "check", + "task", + "--source", + "env.py", + "--agent", + "claude", + "--model", + "claude-sonnet", + "--remote", + ], + ) + + assert result.exit_code == 0 + assert execute.await_args is not None + request = execute.await_args.args[0] + assert request.agent == "claude" + assert request.model == "claude-sonnet" + assert request.remote is True + + +def test_remote_check_does_not_auto_attach_to_a_local_environment() -> None: + from hud.settings import settings + + task = Task(env="demo", id="demo:solve") + with ( + patch.object(settings, "api_key", "api-key"), + patch("hud.cli.check.find_local_env_url", return_value="tcp://127.0.0.1:8765") as find, + patch("hud.cli.check.attached_task") as attached, + patch("hud.cli.check.select_local_task", return_value=(task, "env.py")) as select, + ): + resolved, _, runtime = _resolve( + CheckRequest(task=task.id, agent="claude", remote=True), + ) + + assert resolved is task + assert runtime == "hosted" + find.assert_not_called() + attached.assert_not_called() + select.assert_called_once_with(task.id, ".", {}) + + +def test_evidence_is_redacted_and_bounded() -> None: + evidence = { + "api_key": "secret", + "nested": {"token": "secret", "safe": "x" * 20_000}, + } + + safe = _redact_evidence(evidence, max_chars=120) + serialized = json.dumps(safe) + + assert "secret" not in serialized + assert len(serialized) <= 180 + assert "truncated" in serialized + + +@pytest.mark.parametrize(("answer", "expected_code"), [("3", 0), ("wrong", 1)]) +def test_check_runs_real_local_task_lifecycle( + tmp_path: Path, + answer: str, + expected_code: int, +) -> None: + source = tmp_path / "env.py" + source.write_text( + textwrap.dedent( + """ + from hud import Environment + + env = Environment("sums") + + @env.template(id="add") + async def add(a: int, b: int): + answer = yield f"add:{a}:{b}" + yield 1.0 if answer == str(a + b) else 0.0 + + task = add(a=1, b=2) + """ + ), + encoding="utf-8", + ) + + result = runner.invoke( + app, + [ + "check", + "add", + "--source", + str(source), + "--answer", + answer, + "--json", + "--timeout", + "15", + ], + ) + + assert result.exit_code == expected_code, result.output + payload = json.loads(result.output) + assert payload["schema_version"] == "hud.task-check.v1" + assert payload["reward"] == (1.0 if answer == "3" else 0.0) + assert payload["criteria"][1]["status"] == "passed" + assert payload["criteria"][2]["status"] == "passed" + assert payload["criteria"][3]["status"] == "passed" + + +@pytest.mark.asyncio +async def test_check_attaches_to_a_served_environment(tmp_path: Path) -> None: + source = tmp_path / "env.py" + source.write_text( + textwrap.dedent( + """ + from hud import Environment + + env = Environment("sums") + + @env.template(id="add") + async def add(a: int, b: int): + answer = yield f"add:{a}:{b}" + yield 1.0 if answer == str(a + b) else 0.0 + """ + ), + encoding="utf-8", + ) + task = Task(env="sums", id="add", args={"a": 2, "b": 3}) + + async with SubprocessRuntime(source)(task) as runtime: + result = await asyncio.to_thread( + runner.invoke, + app, + [ + "check", + "add", + "--url", + runtime.url, + "--args", + '{"a": 2, "b": 3}', + "--answer", + "5", + "--json", + "--timeout", + "15", + ], + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["runtime"] == runtime.url + assert payload["reward"] == 1.0 + + +class _Client: + def __init__( + self, + *, + reward: Any = 1.0, + start_error: Exception | None = None, + grade_error: Exception | None = None, + ) -> None: + self.reward = reward + self.start_error = start_error + self.grade_error = grade_error + self.started: list[tuple[str, dict[str, Any]]] = [] + self.graded: list[dict[str, Any]] = [] + self.cancelled = 0 + + async def start_task(self, task_id: str, args: dict[str, Any]) -> dict[str, Any]: + if self.start_error is not None: + raise self.start_error + self.started.append((task_id, args)) + return {"prompt": "safe prompt", "api_key": "must-redact"} + + async def grade(self, answer: dict[str, Any]) -> dict[str, Any]: + self.graded.append(answer) + if self.grade_error is not None: + raise self.grade_error + return {"score": self.reward} + + async def cancel(self) -> None: + self.cancelled += 1 + + +@asynccontextmanager +async def _provided(value: Any) -> AsyncIterator[Any]: + yield value + + +@pytest.mark.asyncio +async def test_startup_timeout_is_reported_separately_from_the_overall_timeout() -> None: + task = Task(env="demo", id="demo:solve") + + @asynccontextmanager + async def stalled_provider(_task: Task) -> AsyncIterator[Runtime]: + await asyncio.Event().wait() + yield Runtime("tcp://127.0.0.1:8765") + + request = CheckRequest( + task=task.id, + answer="42", + startup_timeout=0.01, + timeout=10, + ) + with patch( + "hud.cli.check._resolve", + return_value=(task, stalled_provider, "local"), + ): + report = await _run_check(request) + + assert report.outcome == "error" + assert report.error == "environment did not become ready within 0.01s" + assert report.criteria[1].detail == report.error + + +@pytest.mark.asyncio +async def test_agent_check_relies_on_rollout_timeout_without_an_outer_deadline() -> None: + task = Task(env="demo", id="demo:solve") + + async def completed_agent_run( + _request: CheckRequest, + _task: Task, + _provider: object, + _agent: object, + criteria: dict[str, CheckCriterion], + ) -> tuple[float, str]: + for name in ("environment_startup", "task_startup", "grader_execution"): + criteria[name] = CheckCriterion( + name=name, + status="passed", + detail=f"{name} completed", + ) + return 1.0, "trace-id" + + with ( + patch("hud.cli.check._resolve", return_value=(task, object(), "local")), + patch("hud.cli.check._agent", return_value=object()), + patch("hud.cli.check._run_agent", side_effect=completed_agent_run), + patch( + "hud.cli.check.asyncio.timeout", + side_effect=AssertionError("agent checks must use rollout_timeout"), + ), + ): + report = await _run_check( + CheckRequest(task=task.id, agent="claude", timeout=0.01), + ) + + assert report.outcome == "passed" + assert report.trace_id == "trace-id" + + +@pytest.mark.asyncio +async def test_direct_oracle_uses_start_and_grade_lifecycle() -> None: + client = _Client(reward=0.75) + task = Task(env="demo", id="demo:solve", args={"seed": 3}) + runtime = Runtime("tcp://127.0.0.1:8765") + provider = lambda _task: _provided(runtime) + criteria = _criteria_template() + + with patch("hud.clients.connect", lambda _runtime: _provided(client)): + reward, trace_id = await _run_direct( + CheckRequest(task=task.id, answer="42"), + task, + provider, + criteria, + ) + + assert reward == 0.75 + assert trace_id is None + assert client.started == [(task.id, {"seed": 3})] + assert client.graded == [{"answer": "42"}] + assert criteria["environment_startup"].status == "passed" + assert criteria["task_startup"].status == "passed" + assert criteria["grader_execution"].status == "passed" + assert "must-redact" not in json.dumps(criteria["task_startup"].evidence) + + +@pytest.mark.asyncio +async def test_start_only_never_invokes_grader() -> None: + client = _Client() + task = Task(env="demo", id="demo:solve") + runtime = Runtime("tcp://127.0.0.1:8765") + criteria = _criteria_template() + + with patch("hud.clients.connect", lambda _runtime: _provided(client)): + reward, _ = await _run_direct( + CheckRequest(task=task.id, start_only=True), + task, + lambda _task: _provided(runtime), + criteria, + ) + + assert reward is None + assert client.graded == [] + assert client.cancelled == 1 + assert criteria["grader_execution"].status == "skipped" + assert criteria["oracle_or_agent_reward"].status == "skipped" + + +@pytest.mark.asyncio +async def test_direct_grading_failure_cancels_the_live_task_session() -> None: + client = _Client(grade_error=RuntimeError("grader unavailable")) + task = Task(env="demo", id="demo:solve") + + with ( + patch("hud.clients.connect", lambda _runtime: _provided(client)), + pytest.raises(RuntimeError, match="grader unavailable"), + ): + await _run_direct( + CheckRequest(task=task.id, answer="42"), + task, + lambda _task: _provided(Runtime("tcp://127.0.0.1:8765")), + _criteria_template(), + ) + + assert client.cancelled == 1 + + +@pytest.mark.asyncio +async def test_direct_grader_requires_numeric_reward() -> None: + client = _Client(reward="not-a-number") + task = Task(env="demo", id="demo:solve") + criteria = _criteria_template() + + with ( + patch("hud.clients.connect", lambda _runtime: _provided(client)), + pytest.raises((TypeError, ValueError)), + ): + await _run_direct( + CheckRequest(task=task.id, answer="42"), + task, + lambda _task: _provided(Runtime("tcp://127.0.0.1:8765")), + criteria, + ) + + assert criteria["grader_execution"].status == "error" + + +@pytest.mark.asyncio +async def test_agent_rollout_preserves_trace_and_attributes_grader_failure() -> None: + run = SimpleNamespace( + trace=SimpleNamespace( + status="error", + error="[grading] connection reset", + trace_id="00000000-0000-4000-a000-000000000010", + ), + grade=SimpleNamespace(is_error=True, content="connection reset"), + reward=0.0, + ) + criteria = _criteria_template() + + with patch("hud.eval.run.rollout", AsyncMock(return_value=run)): + reward, trace_id = await _run_agent( + CheckRequest(task="demo:solve", agent="claude"), + Task(env="demo", id="demo:solve"), + Runtime("tcp://127.0.0.1:8765"), + object(), + criteria, + ) + + assert reward is None + assert trace_id == run.trace.trace_id + assert criteria["environment_startup"].status == "passed" + assert criteria["task_startup"].status == "passed" + assert criteria["grader_execution"].status == "error" + assert criteria["oracle_or_agent_reward"].status == "skipped" + + +@pytest.mark.parametrize( + ("phase", "expected_criterion"), + [ + ("provisioning", "environment_startup"), + ("connecting", "environment_startup"), + ("starting task", "task_startup"), + ("grading", "grader_execution"), + ("verifying", "grader_execution"), + ("provisioning verifier", "grader_execution"), + ("cleanup", "grader_execution"), + ("agent loop", "oracle_or_agent_reward"), + ], +) +@pytest.mark.asyncio +async def test_agent_rollout_timeout_is_attributed_to_its_lifecycle_phase( + phase: str, + expected_criterion: str, +) -> None: + detail = f"rollout timed out after 30s during {phase}" + run = SimpleNamespace( + trace=SimpleNamespace(status="error", error=detail, trace_id="timeout-trace"), + grade=SimpleNamespace(is_error=False, content=None), + reward=0.0, + ) + criteria = _criteria_template() + + with patch("hud.eval.run.rollout", AsyncMock(return_value=run)): + reward, _ = await _run_agent( + CheckRequest(task="demo:solve", agent="claude"), + Task(env="demo", id="demo:solve"), + Runtime("tcp://127.0.0.1:8765"), + object(), + criteria, + ) + + assert reward is None + assert criteria[expected_criterion].status == "error" + assert criteria[expected_criterion].detail == detail + + +@pytest.mark.asyncio +async def test_cancelled_agent_rollout_is_an_execution_error() -> None: + run = SimpleNamespace( + trace=SimpleNamespace(status="cancelled", error=None, trace_id="cancelled-trace"), + grade=SimpleNamespace(is_error=False, content=None), + reward=0.0, + ) + criteria = _criteria_template() + + with patch("hud.eval.run.rollout", AsyncMock(return_value=run)): + reward, _ = await _run_agent( + CheckRequest(task="demo:solve", agent="claude"), + Task(env="demo", id="demo:solve"), + Runtime("tcp://127.0.0.1:8765"), + object(), + criteria, + ) + + assert reward is None + assert criteria["oracle_or_agent_reward"].status == "error" + assert "cancelled" in criteria["oracle_or_agent_reward"].detail + + +@pytest.mark.asyncio +async def test_agent_error_that_mentions_grading_stays_in_agent_criterion() -> None: + run = SimpleNamespace( + trace=SimpleNamespace( + status="error", + error="[agent loop] model refused the grading instruction", + trace_id="agent-error", + ), + grade=SimpleNamespace(is_error=False, content=None), + reward=0.0, + ) + criteria = _criteria_template() + + with patch("hud.eval.run.rollout", AsyncMock(return_value=run)): + await _run_agent( + CheckRequest(task="demo:solve", agent="claude"), + Task(env="demo", id="demo:solve"), + Runtime("tcp://127.0.0.1:8765"), + object(), + criteria, + ) + + assert criteria["grader_execution"].status == "passed" + assert criteria["oracle_or_agent_reward"].status == "error" + + +@pytest.mark.asyncio +async def test_hosted_check_without_platform_key_is_invalid_configuration() -> None: + from hud.settings import settings + + with patch.object(settings, "api_key", None): + report = await _run_check( + CheckRequest(task="demo:solve", agent="claude", remote=True), + ) + + assert report.outcome == "error" + assert report.error_kind == "input" + assert report.criteria[0].name == "resolution" + assert report.criteria[0].status == "error" + assert "HUD_API_KEY" in report.criteria[0].detail diff --git a/hud/eval/run.py b/hud/eval/run.py index 22376ef90..1efdd37e0 100644 --- a/hud/eval/run.py +++ b/hud/eval/run.py @@ -429,9 +429,10 @@ async def rollout( async def _drive() -> None: nonlocal client, run, _phase async with runtime(task) as addr: - _phase = "starting task" + _phase = "connecting" async with connect(addr) as actor_client: client = actor_client + _phase = "starting task" live = Run( actor_client, task.id, diff --git a/hud/eval/tests/test_rollout.py b/hud/eval/tests/test_rollout.py index d9e60e98f..049afd2c0 100644 --- a/hud/eval/tests/test_rollout.py +++ b/hud/eval/tests/test_rollout.py @@ -27,19 +27,20 @@ import mcp.types as mcp_types import pytest +import hud.eval.run as run_module from hud.agents.base import Agent from hud.agents.openai_compatible import OpenAIChatAgent from hud.agents.types import OpenAIChatConfig from hud.environment import Environment from hud.eval import Job, SubprocessRuntime, Task, Taskset from hud.eval.run import Run, rollout -from hud.eval.runtime import _local +from hud.eval.runtime import Runtime, _local if TYPE_CHECKING: from collections.abc import AsyncIterator from pathlib import Path - from hud.eval.runtime import Runtime + from hud.clients.client import HudClient from hud.eval.task import Task as TaskRow _SUMS_ENV = """\ @@ -819,6 +820,26 @@ async def broken_provider(task: TaskRow) -> AsyncIterator[Runtime]: assert "No module named 'bugs'" in (run.trace.error or "") +async def test_connection_failure_has_its_own_lifecycle_phase( + monkeypatch: pytest.MonkeyPatch, +) -> None: + @asynccontextmanager + async def broken_connect(_runtime: Runtime) -> AsyncIterator[HudClient]: + raise ConnectionError("handshake unavailable") + yield # pragma: no cover + + monkeypatch.setattr(run_module, "connect", broken_connect) + + run = await rollout( + _add_task(1, 1), + _FnAgent(_solve_add), + runtime=Runtime("tcp://127.0.0.1:8765"), + ) + + assert "[connecting] ConnectionError: handshake unavailable" in (run.trace.error or "") + assert run.prompt is None + + async def test_provider_is_called_with_the_task_row_being_placed(env_file: Path) -> None: placed: list[str] = []