diff --git a/docs/v6/advanced/harbor-convert.mdx b/docs/v6/advanced/harbor-convert.mdx index 5ddb9f4f9..08976e670 100644 --- a/docs/v6/advanced/harbor-convert.mdx +++ b/docs/v6/advanced/harbor-convert.mdx @@ -9,13 +9,27 @@ task dirs - is a *frontend* that loads into the same primitives (`Environment`, `Task`, `Taskset`). Integrations are **loaders, not converters**: no codegen roundtrip to run foreign tasks. The Harbor integration lives in the SDK repo at [`integrations/harbor.py`](https://github.com/hud-evals/hud-python/blob/main/integrations/harbor.py) -- a recipe built only on the public SDK surface; copy it into your project or -run it from a checkout. +- a public-surface loader that maps Harbor folders into SDK primitives. The +included `HarborRuntime` is maintained with the SDK for local Docker execution; +copy the loader into your project or run it from a checkout. ## Prerequisites - A Harbor task directory - each task has `task.toml` + `instruction.md`, and usually an `environment/` (with a `Dockerfile`) and `tests/`. +- For local execution, a running Docker Engine or Docker Desktop daemon that + your user can access, plus the Docker Compose v2 plugin (`docker compose`). + +Loading and exporting task metadata does not require Docker. `HarborRuntime` +does: it builds and starts each task through the local Docker daemon. + + +**Run only trusted Harbor tasks.** `HarborRuntime` executes task-supplied +Dockerfiles, Compose definitions, and `tests/test.sh` against your local Docker +daemon. Review them first, especially bind mounts, devices, privileged settings, +build inputs, and scripts; Docker task definitions are not a security boundary +from the host running the daemon. + ## Load Harbor tasks @@ -26,24 +40,75 @@ directly - one row per task dir (`id` = the dir name), sharing one declarative ```python from integrations.harbor import detect, load -assert detect("./terminal-bench") -taskset = load("./terminal-bench") +assert detect("./harbor_tasks") +taskset = load("./harbor_tasks") for task in taskset: print(task.env, task.id) ``` -Like every task row, the result carries no placement. Run it by supplying one - -today that means a substrate already serving the control channel -(`runtime=Runtime(url)`); a docker provider that builds and runs each task's -`environment/` image is the planned follow-up: +Like every task row, the result carries no placement. Run it by supplying one. +For local Docker-backed Harbor execution, use `HarborRuntime`; it builds the +task's `environment/` image, runs a fresh container, exposes the workspace +through HUD's normal shell capability, and grades by running `tests/test.sh`: ```python -from hud import Runtime +from integrations.harbor import HarborRuntime + +job = await taskset.run(agent, runtime=HarborRuntime("./harbor_tasks")) +``` + +The eval CLI can run local Harbor task directories and datasets when you opt +into the Harbor source format: -job = await taskset.run(agent, runtime=Runtime("tcp://127.0.0.1:8765")) +```bash +hud eval ./harbor_tasks claude --format harbor --task-ids cancel-async-tasks --max-steps 30 ``` +Each concurrent Harbor rollout starts its own Compose project and may build an +image. Set `--max-concurrent` to a value your Docker host can sustain; start +with `1` to `3` for unfamiliar tasks, then increase it after observing CPU, +memory, disk, and build-cache pressure. + +### Supported Harbor subset + +`HarborRuntime` validates `task.toml` before starting Docker and fails closed on +features it does not implement: + +| Area | Supported | Rejected | +|------|-----------|----------| +| Task shape | Linux, single-step tasks with `instruction.md` | Windows, `steps`, multi-step reward strategies, artifact collection | +| Environment | `Dockerfile`, a prebuilt `docker_image`, or Compose with a `main` service and sidecars | GPU/TPU resources, task-config healthchecks, MCP servers, `skills_dir` | +| Placement | Public/default networking, environment variables, users, absolute workdirs, CPU and memory limits | Restricted network policies, host allowlists, agent/verifier network overrides | +| Verification | Shared-container `tests/test.sh`, verifier user/env, verifier timeout, `reward.json` or `reward.txt` | Separate verifier environments and verifier collect hooks | + +`[environment].storage_mb` is accepted for Harbor schema compatibility but is +not currently enforced by the Docker/Compose runtime. `[agent].timeout_sec` +bounds only the agent phase, after provisioning and task startup. An explicit +`Taskset.run(..., rollout_timeout=...)` still imposes an earlier whole-rollout +limit when configured. + +Verification uses Harbor's shared-container mode. Hidden tests are staged only +after the agent phase, but the agent and verifier still share one container; +this is parity with Harbor's shared mode, not a security boundary against a +root agent that leaves background processes behind. + +### Choose the primary reward + +`reward.txt` supplies one numeric reward. A `reward.json` object may contain +multiple numeric rewards; `HarborRuntime` selects `reward`, or the only key when +the object has exactly one entry. If neither rule is unambiguous, select the +primary metric explicitly: + +```python +runtime = HarborRuntime("./harbor_tasks", reward_key="security") +job = await taskset.run(agent, runtime=runtime) +``` + +The configured key must exist and contain a numeric reward. An invalid or +ambiguous reward is reported as a grading error rather than silently choosing a +metric. + ## Export HUD tasks to Harbor `export(source, out_dir)` goes the other way: it turns a HUD task source (a diff --git a/docs/v6/reference/cli.mdx b/docs/v6/reference/cli.mdx index 27605ecf7..2db474eb1 100644 --- a/docs/v6/reference/cli.mdx +++ b/docs/v6/reference/cli.mdx @@ -105,6 +105,7 @@ For a platform taskset, pass its name or id directly: `hud eval "My Tasks" claud | `--config`, `-c` | Agent config `key=value` (repeatable). | | `--verbose`, `-v` | Show agent logs (step progress, tool calls) for batch runs too. | | `--very-verbose`, `-vv` | Debug-level logs. | +| `--format` | Task source format: `hud` (default) or `harbor`. | | `--runtime` | Placement: `local` (a child process per rollout, serving the env source — `SubprocessRuntime`), `hud` (HUD runtime tunnel), or `tcp://host:port`. Defaults to `local` for a tasks file; platform tasksets default to remote hosted execution. | | `--remote` | Run the whole rollout remotely on the HUD platform. | | `--yes`, `-y` | Skip confirmation prompt. | @@ -133,7 +134,9 @@ hud sync env # sync environment metadata ``` External benchmark formats (currently Harbor) load directly into the runtime -as `Taskset`s - no conversion step. See [Harbor interop](/v6/advanced/harbor-convert). +as `Taskset`s - no conversion step. For local Harbor directories, opt in with +`--format harbor` so the CLI uses the Harbor loader and Docker-backed runtime +provider. See [Harbor interop](/v6/advanced/harbor-convert). ## Inspect diff --git a/docs/v6/reference/runtime.mdx b/docs/v6/reference/runtime.mdx index 102760d83..1f2ff027a 100644 --- a/docs/v6/reference/runtime.mdx +++ b/docs/v6/reference/runtime.mdx @@ -176,11 +176,15 @@ completion. It requires a gateway agent that can serialize its identity (Claude/ ### `Runtime` ```python -Runtime(url, params=..., config=...) +Runtime(url, params=..., config=..., agent_timeout_s=None) ``` - **`url`** - control-channel address of an already-running substrate (e.g. `tcp://host:8765`). - **`params`** - connection-time data a transport may need (auth token, sandbox id). +- **`agent_timeout_s`** - optional positive agent-phase limit. Its clock starts after provisioning, + connection, and task startup. If `Task.run(..., rollout_timeout=...)` or + `Taskset.run(..., rollout_timeout=...)` supplies an earlier whole-rollout deadline, that earlier + deadline still wins. ## Run on your own infra diff --git a/hud/cli/eval.py b/hud/cli/eval.py index 0a071b53b..e6e0a2589 100644 --- a/hud/cli/eval.py +++ b/hud/cli/eval.py @@ -63,6 +63,7 @@ def _resolve_model_from_catalog(model_id: str) -> tuple[AgentType, str] | None: _CONFIG_PATH = ".hud_eval.toml" _PLACEMENT_CONFLICT_ERROR = "--runtime and --remote are mutually exclusive placement options" +_SOURCE_FORMATS = ("hud", "harbor") def _resolve_env_vars(obj: Any) -> Any: @@ -167,6 +168,7 @@ class AgentPreset: # very_verbose = true # auto_respond = true # gateway = false # Route LLM API calls through HUD Gateway +# format = "hud" # hud or harbor # runtime = "local" # local, hud, or tcp://host:port # remote = false # Run the whole rollout remotely on HUD @@ -264,6 +266,7 @@ class EvalConfig(BaseModel): "group_size", "auto_respond", "gateway", + "format", "runtime", "remote", } @@ -279,6 +282,9 @@ class EvalConfig(BaseModel): auto_respond: bool | None = None group_size: int = 1 gateway: bool = False + #: Source format. ``None``/``hud`` means normal HUD task source loading; + #: ``harbor`` opts into the Harbor integration loader/runtime. + format: str | None = None #: Placement: "local" (spawn each row's env from the source), "hud" #: (HUD runtime tunnel), or a tcp:// url of an already-served env. #: ``None`` means "infer from the source": a local file runs locally, a @@ -306,6 +312,20 @@ def _parse_agent_type(cls, v: Any) -> AgentType | None: ) from None return v + @field_validator("format", mode="before") + @classmethod + def _parse_format(cls, v: Any) -> str | None: + if v is None: + return None + if not isinstance(v, str): + return v + normalized = v.strip().lower() + if normalized in ("", "hud"): + return None + if normalized in _SOURCE_FORMATS: + return normalized + raise ValueError(f"Invalid format: {v}. Must be one of: {', '.join(_SOURCE_FORMATS)}") + def source_is_local_file(self) -> bool: """Whether ``source`` points at an on-disk taskset (vs. a platform slug/id).""" return self.source is not None and Path(self.source).exists() @@ -319,6 +339,13 @@ def resolve_runtime(self) -> EvalConfig: ``--runtime`` is always honored, except ``local`` against a platform taskset, which has no env to spawn. """ + if self.format == "harbor": + if not self.source_is_local_file(): + hud_console.error("--format harbor requires a local Harbor task directory") + raise typer.Exit(1) + if self.remote or (self.runtime is not None and self.runtime != "local"): + hud_console.error("--format harbor currently supports only local runtime placement") + raise typer.Exit(1) if self.runtime is None: if self.source_is_local_file(): return self.model_copy(update={"runtime": "local"}) @@ -502,6 +529,7 @@ def merge_cli( gateway: bool = False, config: list[str] | None = None, task_ids: str | None = None, + format: str | None = None, runtime: str | None = None, remote: bool = False, ) -> EvalConfig: @@ -517,6 +545,7 @@ def merge_cli( "max_concurrent": max_concurrent, "max_steps": max_steps, "group_size": group_size, + "format": format, "runtime": runtime, }.items() if value is not None @@ -604,6 +633,8 @@ def display(self) -> None: table.add_column("Value", style="green") table.add_row("source", str(self.source or "-")) + if self.format: + table.add_row("format", self.format) table.add_row("runtime", str(self.runtime or "-")) table.add_row("agent", self.agent_type.value if self.agent_type else "-") if self.task_ids: @@ -728,6 +759,28 @@ def _spawn_target(source: Path) -> Path: return resolved.parent +def _load_local_taskset(source_path: Path, source_format: str | None) -> Any: + from hud.eval import Taskset + + format_name = source_format or "hud" + if format_name == "hud": + taskset = Taskset.from_file(source_path) + if len(taskset) == 0: + from integrations.harbor import detect + + if detect(source_path): + hud_console.hint( + f"{source_path} looks like a Harbor task directory; " + "rerun with --format harbor to load it." + ) + return taskset + if format_name == "harbor": + from integrations.harbor import load + + return load(source_path) + raise ValueError(f"unsupported task source format: {format_name}") + + def _resolve_placement(cfg: EvalConfig, source_path: Path | None) -> Any: """Map the config's ``runtime`` onto a placement for ``Taskset.run``. @@ -744,6 +797,10 @@ def _resolve_placement(cfg: EvalConfig, source_path: Path | None) -> Any: if cfg.runtime == "local": if source_path is None: raise ValueError("local placement requires a local source path") + if cfg.format == "harbor": + from integrations.harbor import HarborRuntime + + return HarborRuntime(source_path) return SubprocessRuntime(_spawn_target(source_path)) if cfg.runtime == "hud": require_api_key("run HUD runtime tunnel evals") @@ -774,7 +831,7 @@ async def _run_evaluation(cfg: EvalConfig) -> Any: if is_local: hud_console.info(f"Loading tasks from: {cfg.source}") try: - taskset = Taskset.from_file(source_path) + taskset = _load_local_taskset(source_path, cfg.format) except Exception as e: hud_console.error(f"Failed to load tasks from {cfg.source}: {e}") raise typer.Exit(1) from e @@ -888,6 +945,11 @@ def eval_command( gateway: bool = typer.Option( False, "--gateway", "-g", help="Route LLM API calls through HUD Gateway" ), + format: str | None = typer.Option( + None, + "--format", + help="Task source format: hud (default) or harbor.", + ), runtime: str | None = typer.Option( None, "--runtime", @@ -908,6 +970,7 @@ def eval_command( hud eval "My Tasks" claude-sonnet-4-6 --full # Platform taskset, run on the platform hud eval tasks.json claude --config max_tokens=32768 hud eval tasks.json claude --gateway # Route LLM calls through HUD Gateway + hud eval ./harbor_tasks claude --format harbor # Run Harbor task dirs locally hud eval tasks.json claude-sonnet-4-6 --runtime hud # Use HUD runtime tunnel hud eval tasks.json claude-sonnet-4-6 --remote # Execute rollout remotely """ @@ -938,6 +1001,7 @@ def eval_command( group_size=group_size, config=config, gateway=gateway, + format=format, runtime=runtime, remote=remote, ) diff --git a/hud/cli/tests/test_eval_config.py b/hud/cli/tests/test_eval_config.py index 6b94f0b23..4ec6d6ff2 100644 --- a/hud/cli/tests/test_eval_config.py +++ b/hud/cli/tests/test_eval_config.py @@ -6,6 +6,7 @@ from __future__ import annotations +from types import SimpleNamespace from typing import TYPE_CHECKING import pytest @@ -20,6 +21,23 @@ _ARN = "arn:aws:bedrock:us-east-1:123456789012:inference-profile/anthropic.claude" +def _write_harbor_task(root: Path, name: str = "demo-task") -> Path: + task = root / name + (task / "environment").mkdir(parents=True) + (task / "tests").mkdir() + (task / "instruction.md").write_text("Fix the demo task.\n", encoding="utf-8") + (task / "task.toml").write_text( + 'schema_version = "1.3"\n\n[task]\nname = "demo/demo-task"\n', + encoding="utf-8", + ) + (task / "environment" / "Dockerfile").write_text("FROM python:3.12-slim\n", encoding="utf-8") + (task / "tests" / "test.sh").write_text( + "#!/usr/bin/env bash\nmkdir -p /logs/verifier\necho 1 > /logs/verifier/reward.txt\n", + encoding="utf-8", + ) + return task + + def test_is_bedrock_arn() -> None: assert _is_bedrock_arn(_ARN) is True assert _is_bedrock_arn("claude-sonnet-4-6") is False @@ -136,6 +154,121 @@ def test_resolve_placement_runtime_hud_uses_tunnel( assert isinstance(placement, HUDRuntime) +def test_load_local_taskset_uses_hud_loader_by_default(tmp_path: Path) -> None: + _write_harbor_task(tmp_path) + + taskset = eval_mod._load_local_taskset(tmp_path, None) + + assert len(taskset) == 0 + + +def test_load_local_taskset_hints_harbor_format_on_zero_task_harbor_dir( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _write_harbor_task(tmp_path) + hints: list[str] = [] + monkeypatch.setattr(eval_mod.hud_console, "hint", lambda message, **_: hints.append(message)) + + taskset = eval_mod._load_local_taskset(tmp_path, None) + + assert len(taskset) == 0 + assert any("--format harbor" in hint for hint in hints) + + +def test_load_local_taskset_rejects_unknown_format(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="unsupported task source format"): + eval_mod._load_local_taskset(tmp_path, "unknown") + + +def test_load_local_taskset_uses_harbor_loader_when_format_is_harbor(tmp_path: Path) -> None: + _write_harbor_task(tmp_path) + + taskset = eval_mod._load_local_taskset(tmp_path, "harbor") + + assert len(taskset) == 1 + assert taskset["demo-task"].id == "demo-task" + + +def test_resolve_placement_local_harbor_format_uses_harbor_runtime(tmp_path: Path) -> None: + from integrations.harbor import HarborRuntime + + _write_harbor_task(tmp_path) + + placement = eval_mod._resolve_placement( + EvalConfig(runtime="local", format="harbor"), + tmp_path, + ) + + assert isinstance(placement, HarborRuntime) + + +def test_resolve_placement_local_hud_format_uses_subprocess_runtime(tmp_path: Path) -> None: + from hud.eval import SubprocessRuntime + + _write_harbor_task(tmp_path) + + placement = eval_mod._resolve_placement(EvalConfig(runtime="local"), tmp_path) + + assert isinstance(placement, SubprocessRuntime) + + +async def test_run_evaluation_local_harbor_reaches_taskset_run( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hud.eval import Taskset + from integrations.harbor import HarborRuntime + + _write_harbor_task(tmp_path) + agent = object() + calls: list[tuple[Taskset, object, object, int, int]] = [] + + async def fake_run( + taskset: Taskset, + received_agent: object, + *, + runtime: object, + group: int, + max_concurrent: int, + ) -> SimpleNamespace: + calls.append((taskset, received_agent, runtime, group, max_concurrent)) + return SimpleNamespace(id="test-job", runs=[]) + + monkeypatch.setattr(eval_mod, "_build_agent", lambda _: agent) + monkeypatch.setattr(Taskset, "run", fake_run) + + job = await eval_mod._run_evaluation( + EvalConfig( + source=str(tmp_path), + agent_type="openai", + format="harbor", + runtime="local", + ) + ) + + assert job.id == "test-job" + assert len(calls) == 1 + taskset, received_agent, runtime, group, max_concurrent = calls[0] + assert len(taskset) == 1 + assert received_agent is agent + assert isinstance(runtime, HarborRuntime) + assert group == 1 + assert max_concurrent == 30 + + +def test_harbor_format_rejects_nonlocal_source() -> None: + with pytest.raises(typer.Exit): + EvalConfig(source="platform/taskset", format="harbor").resolve_runtime() + + +def test_harbor_format_rejects_nonlocal_runtime(tmp_path: Path) -> None: + _write_harbor_task(tmp_path) + + with pytest.raises(typer.Exit): + EvalConfig(source=str(tmp_path), format="harbor", runtime="hud").resolve_runtime() + + def test_resolve_placement_remote_uses_hosted_runtime( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/hud/eval/run.py b/hud/eval/run.py index f947efd6c..6463b7e00 100644 --- a/hud/eval/run.py +++ b/hud/eval/run.py @@ -51,6 +51,10 @@ logger = logging.getLogger("hud.eval.run") +class _RolloutDeadlineExceeded(Exception): + """The driver's wall-clock deadline expired around a bounded phase.""" + + def _prompt_message(item: Any) -> mcp_types.PromptMessage: """Coerce one wire prompt turn onto MCP's ``PromptMessage`` vocabulary. @@ -314,6 +318,12 @@ async def rollout( reward is captured. Either way the logged warning names the lifecycle phase (``provisioning``, ``starting task``, ``agent loop``, ``grading``) so callers can tell where the failure landed without reading the trace. + + A provider may set :attr:`Runtime.agent_timeout_s` for a task-specific + agent-phase budget. That clock starts immediately before ``agent(run)``; + provisioning, connection, and task startup do not consume it. When both + limits are present, the earlier of that phase deadline and the explicit + whole-rollout ``rollout_timeout`` deadline wins. """ if job_id is None: # no standalone traces: a lone rollout is a job of one job_id = uuid.uuid4().hex @@ -342,12 +352,20 @@ async def _bounded(awaitable: Any) -> Any: ``rollout_timeout`` in total. A client read-timeout is not enough on its own: a wedged upstream that trickles bytes resets the read timer forever, so a single stuck sampling call could otherwise hang the - rollout — and the batch waits on it — indefinitely. A breach surfaces - as ``TimeoutError`` (distinct from a Ctrl-C ``CancelledError``). + rollout — and the batch waits on it — indefinitely. The private + deadline exception distinguishes a real breach from a ``TimeoutError`` + raised by provider code. """ if deadline is None: return await awaitable - return await asyncio.wait_for(awaitable, max(deadline - loop.time(), 0.0)) + timeout = asyncio.timeout_at(deadline) + try: + async with timeout: + return await awaitable + except TimeoutError: + if not timeout.expired(): + raise + raise _RolloutDeadlineExceeded from None async def _drive() -> None: nonlocal run, _phase @@ -367,31 +385,53 @@ async def _drive() -> None: # File tracking (when enabled) emits setup separately, then # streams workspace diffs for the duration of the agent loop. async with file_tracking_observer(client): - try: - await _bounded(agent(run)) - except TimeoutError: - # The run is live with a partial trajectory worth - # grading, so record the truncation and fall through - # to the normal grade-on-exit path. A bare cancel - # (Ctrl-C) does not land here — it is a CancelledError, - # which this does not catch, so it keeps the non-graded - # cancel path in ``__aexit__``. - logger.warning( - "rollout agent loop timed out after %.0fs; grading partial", - rollout_timeout, - ) - run.trace.stop_reason = "timeout" - run.record( - Step( - source="system", - error=f"agent loop timed out after {rollout_timeout:.0f}s", - ) - ) + agent_deadline = deadline + timeout_source = "rollout" if deadline is not None else None + timeout_value = rollout_timeout + if addr.agent_timeout_s is not None: + runtime_deadline = loop.time() + addr.agent_timeout_s + if agent_deadline is None or runtime_deadline < agent_deadline: + agent_deadline = runtime_deadline + timeout_source = "runtime" + timeout_value = addr.agent_timeout_s + + if agent_deadline is None: + await agent(run) + else: + timeout = asyncio.timeout_at(agent_deadline) + try: + async with timeout: + await agent(run) + except TimeoutError: + # Do not misclassify a TimeoutError raised by the + # agent itself as a deadline breach. + if not timeout.expired(): + raise + assert timeout_source is not None + assert timeout_value is not None + if timeout_source == "runtime": + detail = ( + "agent loop timed out after " + f"{timeout_value:g}s (runtime agent limit)" + ) + else: + detail = ( + "agent loop reached rollout timeout after " + f"{timeout_value:g}s" + ) + # The run is live with a partial trajectory worth + # grading, so record the truncation and fall through + # to the normal grade-on-exit path. A bare cancel + # (Ctrl-C) remains a CancelledError and keeps the + # non-graded cancel path in ``__aexit__``. + logger.warning("%s; grading partial", detail) + run.trace.stop_reason = "timeout" + run.record(Step(source="system", error=detail)) _phase = "grading" try: await _drive() - except TimeoutError: + except _RolloutDeadlineExceeded: # A setup-phase deadline (provision/connect) fired — the agent-loop # timeout is handled inside _drive. Isolate it so one wedged rollout # never collapses the batch, keeping any partial trace. diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py index a05e16ab1..0e35296eb 100644 --- a/hud/eval/runtime.py +++ b/hud/eval/runtime.py @@ -33,6 +33,9 @@ import asyncio import contextlib import logging +import math +import os +import shutil import sys import uuid from collections import deque @@ -137,14 +140,27 @@ class Runtime: local process, ``tcp://sandbox-abc.hud.so:443`` for a hosted box). ``params`` carries connection-time data a transport may need (auth token, sandbox id). ``config`` is the effective runtime configuration used to - construct the runtime. Constructed directly, it is also a provider — the - borrowed, shared case: it yields itself with a no-op lifecycle, since - whoever provisioned the substrate owns its teardown. + construct the runtime. ``agent_timeout_s`` optionally bounds only the + agent phase, after provisioning and task startup. Constructed directly, + it is also a provider — the borrowed, shared case: it yields itself with a + no-op lifecycle, since whoever provisioned the substrate owns its teardown. """ url: str params: dict[str, Any] = field(default_factory=dict) config: RuntimeConfig | None = None + agent_timeout_s: float | None = None + + def __post_init__(self) -> None: + if self.agent_timeout_s is None: + return + if ( + isinstance(self.agent_timeout_s, bool) + or not isinstance(self.agent_timeout_s, int | float) + or not math.isfinite(self.agent_timeout_s) + or self.agent_timeout_s <= 0 + ): + raise ValueError("Runtime agent_timeout_s must be a positive finite number") def __call__(self, task: Task) -> AbstractAsyncContextManager[Runtime]: return nullcontext(self) @@ -788,21 +804,117 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]: await daytona.delete(sandbox) -async def _docker(*args: str, check: bool = True) -> tuple[str, str]: - """Run a docker CLI command and return decoded ``(stdout, stderr)``.""" - proc = await asyncio.create_subprocess_exec( - "docker", +async def _docker( + *args: str, + check: bool = True, + env: Mapping[str, str] | None = None, + unset_env: Sequence[str] = (), +) -> tuple[str, str]: + """Run a non-interactive Docker CLI command with bounded diagnostic output. + + ``env`` is reserved for trusted host-side overrides. ``unset_env`` removes + task-controlled names from the Docker client's process environment without + affecting executable lookup, which is resolved before the child is spawned. + """ + docker = shutil.which("docker") or "docker" + process_env: dict[str, str] | None = None + if env is not None or unset_env: + process_env = dict(os.environ) + for key in unset_env: + process_env.pop(key, None) + process_env.update(env or {}) + group = await create_process_group_exec( + docker, *args, + term_timeout=5.0, + kill_timeout=5.0, + stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + env=process_env, ) - out, err = await proc.communicate() + proc = group.process + assert proc.stdout is not None + assert proc.stderr is not None + out_task = asyncio.create_task(_read_bounded_output(proc.stdout)) + err_task = asyncio.create_task(_read_bounded_output(proc.stderr)) + try: + await proc.wait() + out, err = await asyncio.gather(out_task, err_task) + await group.terminate() + except BaseException: + with contextlib.suppress(Exception): + await group.terminate() + for reader in (out_task, err_task): + if not reader.done(): + reader.cancel() + await asyncio.gather(out_task, err_task, return_exceptions=True) + raise if check and proc.returncode != 0: detail = err.decode("utf-8", "replace").strip() or out.decode("utf-8", "replace").strip() - raise RuntimeError(f"docker {' '.join(args)} failed ({proc.returncode}): {detail}") + for sensitive_value in _sensitive_docker_values(args): + detail = detail.replace(sensitive_value, "") + command = " ".join(_redact_docker_args(args)) + raise RuntimeError(f"docker {command} failed ({proc.returncode}): {detail}") return out.decode("utf-8", "replace"), err.decode("utf-8", "replace") +_DOCKER_OUTPUT_LIMIT = 1_000_000 + + +async def _read_bounded_output(stream: asyncio.StreamReader) -> bytes: + """Drain a subprocess stream while retaining only its diagnostic tail.""" + tail = bytearray() + while chunk := await stream.read(64 * 1024): + tail.extend(chunk) + if len(tail) > _DOCKER_OUTPUT_LIMIT: + del tail[: len(tail) - _DOCKER_OUTPUT_LIMIT] + return bytes(tail) + + +def _redact_docker_args(args: Sequence[str]) -> list[str]: + """Render Docker arguments without exposing inline credentials or env values.""" + rendered: list[str] = [] + redact_next = False + sensitive = {"--env", "-e", "--env-file", "--password", "--secret"} + for arg in args: + if redact_next: + rendered.append("") + redact_next = False + elif arg in sensitive: + rendered.append(arg) + redact_next = True + elif any(arg.startswith(f"{flag}=") for flag in sensitive): + rendered.append(f"{arg.split('=', 1)[0]}=") + else: + rendered.append(arg) + return rendered + + +def _sensitive_docker_values(args: Sequence[str]) -> set[str]: + """Values to scrub if a failed CLI echoes its own sensitive arguments.""" + values: set[str] = set() + sensitive = {"--env", "-e", "--env-file", "--password", "--secret"} + redact_next = False + for arg in args: + if redact_next: + if arg: + values.add(arg) + if "=" in arg: + value = arg.split("=", 1)[1] + if len(value) >= 4: + values.add(value) + redact_next = False + elif arg in sensitive: + redact_next = True + elif any(arg.startswith(f"{flag}=") for flag in sensitive): + values.add(arg) + value = arg.split("=", 1)[1] + if len(value) >= 4: + values.add(value) + return values + + @asynccontextmanager async def _local(env: Environment, *, ready_timeout: float | None = None) -> AsyncIterator[Runtime]: """Substrate-side serving: a live env owned by *this* process, as a runtime. diff --git a/hud/eval/tests/test_docker_cli.py b/hud/eval/tests/test_docker_cli.py new file mode 100644 index 000000000..942c651f7 --- /dev/null +++ b/hud/eval/tests/test_docker_cli.py @@ -0,0 +1,269 @@ +"""Safety properties of the shared Docker CLI runner. + +These tests execute tiny Python programs in place of ``docker``. They exercise +the real subprocess boundary without requiring or mutating a Docker daemon. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +import signal +import sys +import time +from typing import TYPE_CHECKING, cast + +import pytest + +from hud.eval import runtime as runtime_module + +if TYPE_CHECKING: + from pathlib import Path + from typing import Any + + +def _write_cli(tmp_path: Path, source: str) -> Path: + script = tmp_path / "fake_docker.py" + script.write_text(source) + return script + + +@pytest.fixture +def python_docker(monkeypatch: pytest.MonkeyPatch) -> None: + """Resolve ``docker`` to Python so the first CLI argument is a test script.""" + + monkeypatch.setattr( + runtime_module.shutil, + "which", + lambda executable: sys.executable if executable == "docker" else None, + ) + + +def _wait_for_file(path: Path, *, wait_seconds: float = 3.0) -> None: + deadline = time.monotonic() + wait_seconds + while not path.exists(): + if time.monotonic() >= deadline: + raise TimeoutError(f"timed out waiting for {path}") + time.sleep(0.01) + + +def _pid_is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + +async def test_docker_cli_retains_only_bounded_output_tails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + python_docker: None, +) -> None: + limit = 32 + stdout = "discard-stdout-" + "o" * 100 + "-stdout-tail" + stderr = "discard-stderr-" + "e" * 100 + "-stderr-tail" + script = _write_cli( + tmp_path, + f"import sys\nsys.stdout.write({stdout!r})\nsys.stderr.write({stderr!r})\n", + ) + monkeypatch.setattr(runtime_module, "_DOCKER_OUTPUT_LIMIT", limit) + + out, err = await runtime_module._docker(str(script)) + + assert out == stdout[-limit:] + assert err == stderr[-limit:] + + +async def test_docker_cli_uses_devnull_stdin( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + python_docker: None, +) -> None: + script = _write_cli( + tmp_path, + "import sys\nassert sys.stdin.buffer.read() == b''\nprint('stdin-eof')\n", + ) + real_create_subprocess_exec = asyncio.create_subprocess_exec + seen: dict[str, object] = {} + + async def recording_create_subprocess_exec( + *args: str, **kwargs: object + ) -> asyncio.subprocess.Process: + seen["stdin"] = kwargs.get("stdin") + return await cast("Any", real_create_subprocess_exec)(*args, **kwargs) + + monkeypatch.setattr( + runtime_module.asyncio, + "create_subprocess_exec", + recording_create_subprocess_exec, + ) + + out, _ = await runtime_module._docker(str(script)) + + assert out.strip() == "stdin-eof" + assert seen["stdin"] is asyncio.subprocess.DEVNULL + + +async def test_docker_cli_unsets_task_environment_before_trusted_overrides( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + python_docker: None, +) -> None: + task_only = "HUD_TEST_DOCKER_TASK_ONLY" + overridden = "HUD_TEST_DOCKER_OVERRIDE" + trusted_only = "HUD_TEST_DOCKER_TRUSTED_ONLY" + script = _write_cli( + tmp_path, + "import json, os, sys\n" + "print(json.dumps({key: os.environ.get(key) for key in sys.argv[1:]}))\n", + ) + monkeypatch.setenv(task_only, "task-value") + monkeypatch.setenv(overridden, "task-value") + monkeypatch.delenv(trusted_only, raising=False) + + out, _ = await runtime_module._docker( + str(script), + task_only, + overridden, + trusted_only, + "PATH", + env={overridden: "trusted-value", trusted_only: "trusted-value"}, + unset_env=(task_only, overridden, "PATH"), + ) + + values = json.loads(out) + assert values == { + task_only: None, + overridden: "trusted-value", + trusted_only: "trusted-value", + "PATH": None, + } + + +async def test_docker_cli_redacts_sensitive_arguments_on_failure( + tmp_path: Path, + python_docker: None, +) -> None: + script = _write_cli( + tmp_path, + "import sys\n" + "sys.stderr.write('controlled failure: ' + ' '.join(sys.argv[1:]))\n" + "raise SystemExit(19)\n", + ) + + with pytest.raises(RuntimeError) as error: + await runtime_module._docker( + str(script), + "--env", + "API_TOKEN=visible-env-secret", + "--env-file=/tmp/visible-env-file-secret", + "-e=INLINE=visible-inline-secret", + "--password", + "visible-password-secret", + "--secret=id=token,src=/tmp/visible-secret-file", + ) + + message = str(error.value) + assert "controlled failure" in message + assert "--env " in message + assert "--env-file=" in message + assert "-e=" in message + assert "--password " in message + assert "--secret=" in message + assert "visible-" not in message + + +@pytest.mark.skipif(not hasattr(os, "killpg"), reason="process groups require POSIX") +async def test_docker_cli_cancellation_terminates_group_and_reaps_leader( + tmp_path: Path, + python_docker: None, +) -> None: + parent_pid_path = tmp_path / "parent.pid" + child_pid_path = tmp_path / "child.pid" + parent_term_path = tmp_path / "parent.terminated" + child_term_path = tmp_path / "child.terminated" + script = _write_cli( + tmp_path, + """\ +import os +import pathlib +import signal +import subprocess +import sys +import time + +parent_pid, child_pid, parent_term, child_term = map(pathlib.Path, sys.argv[1:]) + +def terminate_parent(signum, frame): + parent_term.write_text(str(signum)) + raise SystemExit(0) + +signal.signal(signal.SIGTERM, terminate_parent) +parent_pid.write_text(str(os.getpid())) +child_source = '''\ +import os +import pathlib +import signal +import sys +import time + +pid_path, term_path = map(pathlib.Path, sys.argv[1:]) + +def terminate_child(signum, frame): + term_path.write_text(str(signum)) + raise SystemExit(0) + +signal.signal(signal.SIGTERM, terminate_child) +pid_path.write_text(str(os.getpid())) +while True: + time.sleep(1) +''' +subprocess.Popen( + [sys.executable, "-c", child_source, str(child_pid), str(child_term)], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, +) +while True: + time.sleep(1) +""", + ) + process = asyncio.create_task( + runtime_module._docker( + str(script), + str(parent_pid_path), + str(child_pid_path), + str(parent_term_path), + str(child_term_path), + ) + ) + + try: + await asyncio.gather( + asyncio.to_thread(_wait_for_file, parent_pid_path), + asyncio.to_thread(_wait_for_file, child_pid_path), + ) + parent_pid = int(parent_pid_path.read_text()) + + process.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(process, timeout=3.0) + + await asyncio.gather( + asyncio.to_thread(_wait_for_file, parent_term_path), + asyncio.to_thread(_wait_for_file, child_term_path), + ) + assert not _pid_is_alive(parent_pid) + finally: + process.cancel() + with contextlib.suppress(asyncio.CancelledError, TimeoutError): + await asyncio.wait_for(process, timeout=1.0) + for pid_path in (parent_pid_path, child_pid_path): + if pid_path.exists(): + pid = int(pid_path.read_text()) + if _pid_is_alive(pid): + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) diff --git a/hud/eval/tests/test_rollout.py b/hud/eval/tests/test_rollout.py index 4995a9505..cf3e77730 100644 --- a/hud/eval/tests/test_rollout.py +++ b/hud/eval/tests/test_rollout.py @@ -31,7 +31,7 @@ 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 import Job, Runtime, SubprocessRuntime, Task, Taskset from hud.eval.run import Run, rollout from hud.eval.runtime import _local @@ -39,7 +39,6 @@ from collections.abc import AsyncIterator from pathlib import Path - from hud.eval.runtime import Runtime from hud.eval.task import Task as TaskRow _SUMS_ENV = """\ @@ -120,6 +119,12 @@ def _solve_add(prompt: str) -> str: return str(int(a) + int(b)) +@pytest.mark.parametrize("timeout", [0.0, -1.0, float("inf"), float("nan"), True]) +def test_runtime_rejects_invalid_agent_timeout(timeout: float) -> None: + with pytest.raises(ValueError, match="positive finite"): + Runtime("tcp://127.0.0.1:1", agent_timeout_s=timeout) + + def _pid_status(pid: int) -> str | None: result = subprocess.run( ["ps", "-o", "stat=", "-p", str(pid)], @@ -313,6 +318,11 @@ async def __call__(self, run: Any) -> None: await asyncio.sleep(30) +class _AgentRaisedTimeout(Agent): + async def __call__(self, run: Any) -> None: + raise TimeoutError("agent's internal timeout") + + async def test_agent_loop_timeout_grades_the_partial_trajectory() -> None: # In-process env (no subprocess spawn) so only the agent loop, not setup, # races the short deadline. @@ -338,6 +348,87 @@ async def add(a: int, b: int): assert run.trace_id is not None +async def test_runtime_agent_timeout_only_bounds_the_agent_phase() -> None: + env = Environment("sums") + + @env.template() + async def add(a: int, b: int): + answer = yield f"add:{a}:{b}" + yield 1.0 if answer == str(a + b) else 0.0 + + @asynccontextmanager + async def delayed_provider(task: TaskRow) -> AsyncIterator[Runtime]: + await asyncio.sleep(0.05) + async with _local(env) as runtime: + yield Runtime(runtime.url, runtime.params, agent_timeout_s=0.01) + + run = await rollout(_add_task(2, 3), _FnAgent(_solve_add), runtime=delayed_provider) + + assert run.reward == 1.0 + assert run.trace.stop_reason is None + + +async def test_runtime_agent_timeout_grades_the_partial_trajectory() -> None: + env = Environment("sums") + + @env.template() + async def add(a: int, b: int): + answer = yield f"add:{a}:{b}" + yield 1.0 if answer == str(a + b) else 0.0 + + @asynccontextmanager + async def provider(task: TaskRow) -> AsyncIterator[Runtime]: + async with _local(env) as runtime: + yield Runtime(runtime.url, runtime.params, agent_timeout_s=0.05) + + run = await rollout(_add_task(2, 3), _SlowAgent(_solve_add), runtime=provider) + + assert run.reward == 1.0 + assert run.trace.status == "completed" + assert run.trace.stop_reason == "timeout" + assert any(step.error and "runtime agent limit" in step.error for step in run.trace.steps) + + +async def test_rollout_timeout_remains_the_dominant_agent_deadline() -> None: + env = Environment("sums") + + @env.template() + async def add(a: int, b: int): + answer = yield f"add:{a}:{b}" + yield 1.0 if answer == str(a + b) else 0.0 + + @asynccontextmanager + async def provider(task: TaskRow) -> AsyncIterator[Runtime]: + async with _local(env) as runtime: + yield Runtime(runtime.url, runtime.params, agent_timeout_s=30.0) + + run = await rollout( + _add_task(2, 3), + _SlowAgent(_solve_add), + runtime=provider, + rollout_timeout=0.2, + ) + + assert run.reward == 1.0 + assert run.trace.stop_reason == "timeout" + assert any(step.error and "rollout timeout" in step.error for step in run.trace.steps) + + +async def test_agent_raised_timeout_is_not_reported_as_a_deadline(env_file: Path) -> None: + base = SubprocessRuntime(env_file) + + @asynccontextmanager + async def provider(task: TaskRow) -> AsyncIterator[Runtime]: + async with base(task) as runtime: + yield Runtime(runtime.url, runtime.params, agent_timeout_s=30.0) + + run = await rollout(_add_task(2, 3), _AgentRaisedTimeout(), runtime=provider) + + assert run.trace.is_error + assert run.trace.stop_reason is None + assert "agent's internal timeout" in (run.trace.error or "") + + async def test_pre_launch_failure_yields_a_synthesized_failed_run() -> None: @asynccontextmanager async def broken_provider(task: TaskRow) -> AsyncIterator[Runtime]: diff --git a/integrations/__init__.py b/integrations/__init__.py index c8549e0fe..baa460f4d 100644 --- a/integrations/__init__.py +++ b/integrations/__init__.py @@ -5,11 +5,12 @@ primitives. Integrations are **loaders, not converters**: no codegen roundtrip to run foreign tasks. -This package lives outside ``hud`` on purpose: each module is a recipe built -**only on the public SDK surface** (``Environment``, ``Task``, -``Taskset``, ``Runtime``) — that constraint is the proof the core is -flexible. Copy a module into your project or run it from a checkout; nothing -in the SDK or CLI imports it. +This package lives outside ``hud`` on purpose: loaders are recipes built on the +public SDK surface (``Environment``, ``Task``, ``Taskset``, ``Runtime``). Copy a +loader into your project or run it from a checkout. The CLI may call selected +integrations explicitly for polished interop paths. A repo-maintained +integration may also expose a local provider for that explicit CLI path; that +provider is SDK implementation code, not the portable loader contract. The contract: an integration module exposes ``detect(path) -> bool`` and ``load(path) -> Taskset``. Placement stays an execution-time concern — loaders diff --git a/integrations/harbor.py b/integrations/harbor.py index 35b903f73..e36a5ed94 100644 --- a/integrations/harbor.py +++ b/integrations/harbor.py @@ -11,11 +11,9 @@ :func:`load` parses a task dir (or a dataset of them) into rows sharing one env name per distinct ``environment/`` build context — no codegen, no -roundtrip. Like every row, the result is runnable -once a placement is supplied (``runtime=Runtime(url)`` against a served substrate -today). Providers receive the row being placed, so a docker provider that -builds and runs each row's ``environment/`` image is the named follow-up — -expressible without engine changes. +roundtrip. Like every row, the result is runnable once a placement is supplied. +Use :class:`HarborRuntime` for local Docker-backed execution of Harbor tasks, or +``runtime=Runtime(url)`` to attach to a substrate served elsewhere. :func:`export` is the reverse direction: turn a HUD task source into self-contained Harbor task folders (``task.toml`` + ``instruction.md`` + @@ -40,12 +38,10 @@ from __future__ import annotations -import hashlib import json import logging -import re +import shlex import shutil -import tomllib from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any @@ -53,6 +49,14 @@ from hud.environment import Environment from hud.environment.server import TaskRunner from hud.eval import Task, Taskset +from integrations.harbor_common import ( + _HarborTask, + _is_harbor_task, + _parse_task, + _slugify, + _task_dirs, +) +from integrations.harbor_runtime import HarborRuntime if TYPE_CHECKING: from collections.abc import Callable @@ -74,18 +78,12 @@ "__pycache__", "*.pyc", ".git", ".venv", "venv", "*.egg-info", ".pytest_cache" ) - # ─── load: Harbor dirs -> Taskset ────────────────────────────────────── def detect(path: str | Path) -> bool: """True when *path* is a Harbor task dir or a dataset of them.""" - root = Path(path) - if _is_harbor_task(root): - return True - if root.is_dir(): - return any(_is_harbor_task(d) for d in root.iterdir() if d.is_dir()) - return False + return bool(_task_dirs(path)) def load(path: str | Path) -> Taskset: @@ -96,12 +94,8 @@ def load(path: str | Path) -> Taskset: context (content-hashed), derived from the dataset name. """ root = Path(path).resolve() - if _is_harbor_task(root): - task_dirs = [root] - dataset_name = root.parent.name - else: - task_dirs = sorted(d for d in root.iterdir() if d.is_dir() and _is_harbor_task(d)) - dataset_name = root.name + task_dirs = _task_dirs(root) + dataset_name = root.parent.name if _is_harbor_task(root) else root.name if not task_dirs: raise ValueError(f"no Harbor tasks found in {path}") @@ -122,59 +116,31 @@ def load(path: str | Path) -> Taskset: tasks: list[Task] = [] for idx, group in enumerate(sorted_groups, start=1): env_name = base_name if len(sorted_groups) == 1 else f"{base_name}-g{idx}" - tasks.extend(Task(env=env_name, id=harbor_task.task_id) for harbor_task in group) + tasks.extend( + Task( + env=env_name, + id=harbor_task.task_id, + columns=( + metadata + if isinstance((metadata := harbor_task.config.get("metadata")), dict) + else None + ), + ) + for harbor_task in group + ) return Taskset(base_name, tasks) -def _slugify(name: str) -> str: - """A valid env name (lowercase ``[a-z0-9-]``) from a dataset dir name.""" - normalized = re.sub(r"[^a-z0-9-]", "", name.strip().lower().replace(" ", "-").replace("_", "-")) - return re.sub(r"-+", "-", normalized).strip("-") or "harbor" - - -def _is_harbor_task(path: Path) -> bool: - return path.is_dir() and (path / "task.toml").exists() and (path / "instruction.md").exists() - - -def _hash_directory(path: Path) -> str: - """Content-hash a directory for grouping tasks by identical environments.""" - hasher = hashlib.sha256() - if not path.exists(): - return "empty" - for file_path in sorted(path.rglob("*")): - if file_path.is_file(): - hasher.update(str(file_path.relative_to(path)).encode()) - hasher.update(file_path.read_bytes()) - return hasher.hexdigest()[:16] - - -@dataclass(frozen=True, slots=True) -class _HarborTask: - """One parsed Harbor task dir.""" - - task_id: str - config: dict[str, Any] - env_hash: str - +# ─── export: HUD tasks -> Harbor task folders ─────────────────────────── -def _parse_task(task_dir: Path) -> _HarborTask | None: - if not (task_dir / "instruction.md").is_file(): - LOGGER.warning("failed to read instruction.md in %s", task_dir) - return None - try: - config: dict[str, Any] = tomllib.loads((task_dir / "task.toml").read_text("utf-8")) - except (OSError, tomllib.TOMLDecodeError): - LOGGER.warning("failed to parse task.toml in %s", task_dir) - config = {} - env_dir = task_dir / "environment" - return _HarborTask( - task_id=task_dir.name, - config=config, - env_hash=_hash_directory(env_dir) if env_dir.exists() else "no-env", - ) +@dataclass(frozen=True) +class _AuthoredEnvironment: + """An authored env together with the source pointer Harbor must serve.""" -# ─── export: HUD tasks -> Harbor task folders ─────────────────────────── + env: Environment + module_path: Path + symbol: str def _write_text(path: Path, text: str) -> None: @@ -205,20 +171,32 @@ async def _materialize_prompt(env: Environment, task: str, args: dict[str, Any]) return prompt if isinstance(prompt, str) else json.dumps(prompt, indent=2, default=str) -def _resolve_env(task: Task, authored: dict[str, Environment]) -> Environment: +def _resolve_env( + task: Task, + authored: dict[str, list[_AuthoredEnvironment]], +) -> _AuthoredEnvironment: """Resolve a task row's env name to a local, authored env defining the task. Rows reference envs by name; export materializes prompts in-process, so the authored ``Environment`` must be defined in (or next to) the task source. A row whose name matches nothing exportable fails loudly. """ - env = authored.get(task.env) - if env is None or task.id not in env.tasks: + matches = [ref for ref in authored.get(task.env, ()) if task.id in ref.env.tasks] + # One Environment may be re-exported under several symbols. Preserve the + # first concrete pointer, but reject genuinely distinct same-named envs. + unique = {id(ref.env): ref for ref in reversed(matches)} + if not unique: raise TypeError( f"harbor export needs a local env defining task {task.id!r} " - f"(an env.py named {task.env!r} next to the tasks); none was found.", + f"(an Environment named {task.env!r} in the source or an adjacent " + "Python module); none was found.", + ) + if len(unique) > 1: + raise TypeError( + f"harbor export found multiple local envs named {task.env!r} " + f"defining task {task.id!r}; make the env name unique.", ) - return env + return next(iter(unique.values())) # ─── generated files ─────────────────────────────────────────────────── @@ -231,40 +209,83 @@ def _resolve_env(task: Task, authored: dict[str, Environment]) -> Environment: # run via ENTRYPOINT; `exec "$@"` keeps the channel alive alongside it. The # agent and the verifier both run in this same container, so the verifier # reaches the parked run on 127.0.0.1:{port} to grade. -set -u - -hud serve env:env --port {port} & +set -eu + +HUD_ENV_TARGET={env_target} +HUD_TASK={task} +HUD_ARGS={args_json} +HUD_READY_FILE=/tmp/.hud-harbor-ready +server_pid= + +cleanup() {{ + status=$? + trap - EXIT HUP INT TERM + if [ -n "$server_pid" ]; then + kill "$server_pid" 2>/dev/null || true + wait "$server_pid" 2>/dev/null || true + fi + exit "$status" +}} + +trap cleanup EXIT +trap 'exit 129' HUP +trap 'exit 130' INT +trap 'exit 143' TERM + +rm -f "$HUD_READY_FILE" + +hud serve "$HUD_ENV_TARGET" --port {port} & +server_pid=$! # Wait for the control channel to accept connections (python is always present). -python3 -c 'import socket, sys, time +python3 -c 'import os, socket, sys, time port = int(sys.argv[1]) +pid = int(sys.argv[2]) for _ in range(120): + try: + os.kill(pid, 0) + except ProcessLookupError: + raise SystemExit("hud serve exited before becoming ready") from None try: socket.create_connection(("127.0.0.1", port), 0.5).close() break except OSError: - time.sleep(0.5)' {port} || true + time.sleep(0.5) +else: + raise SystemExit("hud serve did not become ready in 60 seconds")' {port} "$server_pid" # Run the task setup phase and park the run for grading. -hud task start '{task}' --args '{args_json}' --url tcp://127.0.0.1:{port} >/dev/null 2>&1 || true +hud task start --args "$HUD_ARGS" --url tcp://127.0.0.1:{port} -- "$HUD_TASK" >/dev/null +: > "$HUD_READY_FILE" +trap - EXIT HUP INT TERM exec "$@" """ _TEST_SH = """\ #!/bin/sh # Grade the parked HUD run against the agent's work, writing the Harbor reward. -set -u +set -eu mkdir -p /logs/verifier -ANSWER_FILE='{answer_file}' +HUD_TASK={task} +HUD_ARGS={args_json} +ANSWER_FILE={answer_file} [ -f "$ANSWER_FILE" ] || : > "$ANSWER_FILE" - -if hud task grade '{task}' --args '{args_json}' --answer-file "$ANSWER_FILE" \\ - --url tcp://127.0.0.1:{port} > /logs/verifier/reward.txt 2> /logs/verifier/grade.err; then - : +REWARD_FILE=/logs/verifier/reward.txt +REWARD_TMP=/logs/verifier/reward.txt.tmp +GRADE_ERR=/logs/verifier/grade.err +rm -f "$REWARD_FILE" "$REWARD_TMP" + +if hud task grade --args "$HUD_ARGS" --answer-file "$ANSWER_FILE" \\ + --url tcp://127.0.0.1:{port} -- "$HUD_TASK" \ + > "$REWARD_TMP" 2> "$GRADE_ERR"; then + mv "$REWARD_TMP" "$REWARD_FILE" else - echo 0 > /logs/verifier/reward.txt + status=$? + [ ! -s "$GRADE_ERR" ] || cat "$GRADE_ERR" >&2 + rm -f "$REWARD_TMP" "$REWARD_FILE" + exit "$status" fi """ @@ -276,18 +297,13 @@ def _resolve_env(task: Task, authored: dict[str, Environment]) -> Environment: def _adapt_env_dockerfile(content: str) -> str: - """Neutralize the env's own CMD/ENTRYPOINT and bake the boot ENTRYPOINT. + """Append the HUD boot process as the image's final startup configuration. - ENTRYPOINT (not CMD) because Harbor overrides the container command with - ``sleep infinity``; our entrypoint runs setup then ``exec "$@"`` into it. + Docker uses the final CMD/ENTRYPOINT, so preserving earlier instructions + verbatim also preserves valid multiline JSON-form instructions. ENTRYPOINT + runs setup before handing off to Harbor's command; the healthcheck prevents + Compose ``--wait`` from racing ahead of that setup. """ - lines: list[str] = [] - for line in content.splitlines(): - stripped = line.strip().upper() - if stripped.startswith(("CMD ", "CMD[", "ENTRYPOINT ", "ENTRYPOINT[")): - lines.append(f"# [hud original] {line}") - else: - lines.append(line) boot_layer = ( "\n# ─── HUD → Harbor boot entrypoint ───\n" "COPY hud_entrypoint.sh /hud_entrypoint.sh\n" @@ -295,17 +311,26 @@ def _adapt_env_dockerfile(content: str) -> str: 'ENTRYPOINT ["/hud_entrypoint.sh"]\n' "# Default command for standalone `docker run`; Harbor injects its own.\n" 'CMD ["sh", "-c", "sleep infinity"]\n' + "HEALTHCHECK --interval=1s --timeout=1s --start-period=1s --retries=120 " + 'CMD ["test", "-f", "/tmp/.hud-harbor-ready"]\n' ) - return "\n".join(lines) + "\n" + boot_layer + return content.rstrip() + "\n" + boot_layer def _harbor_task_toml(name: str, task: str, args: dict[str, Any], timeout: float) -> str: - """A Harbor-native ``task.toml`` (``name``/``version`` required by the registry).""" + """Return a current, registry-publishable Harbor ``task.toml``. + + Harbor's package identity lives under ``[task]`` and must use ``org/name`` + form. A root-level ``name`` is merely ignored by Harbor's ``TaskConfig``; + the publisher then rejects the task because no package was declared. + """ + package_name = f"hud/{_slugify(name)}" return ( - 'version = "1.0"\n' - f'name = "{name}"\n' + 'schema_version = "1.4"\n' + "\n[task]\n" + f"name = {json.dumps(package_name)}\n" "\n[metadata]\n" - f'hud_task = "{task}"\n' + f"hud_task = {json.dumps(task)}\n" f"hud_args = {json.dumps(json.dumps(args))}\n" "\n[agent]\n" f"timeout_sec = {timeout}\n" @@ -321,15 +346,23 @@ def _find_dockerfile(source_dir: Path) -> Path | None: ) -def _make_ignore(out_root: Path) -> Callable[[str, list[str]], set[str]]: +def _make_ignore( + out_root: Path, + excluded_files: tuple[Path, ...] = (), +) -> Callable[[str, list[str]], set[str]]: """Ignore the standard caches plus the export output dir (which may live under the source dir, e.g. ``./harbor_tasks`` next to ``env.py``).""" out_resolved = out_root.resolve() + excluded_resolved = {path.resolve() for path in excluded_files} def _ignore(dirpath: str, names: list[str]) -> set[str]: ignored = set(_BUILD_CONTEXT_IGNORE(dirpath, names)) base = Path(dirpath) - ignored.update(n for n in names if (base / n).resolve() == out_resolved) + ignored.update( + n + for n in names + if (candidate := (base / n).resolve()) == out_resolved or candidate in excluded_resolved + ) return ignored return _ignore @@ -341,17 +374,19 @@ def _write_environment( dockerfile: Path, task: str, args: dict[str, Any], + env_target: str, out_root: Path, + taskset_source: Path | None, ) -> None: """Copy the env build context into ``environment/`` and bake the boot entrypoint.""" env_out = task_dir / "environment" if env_out.exists(): shutil.rmtree(env_out) - shutil.copytree(source_dir, env_out, ignore=_make_ignore(out_root)) + excluded = (taskset_source,) if taskset_source is not None else () + shutil.copytree(source_dir, env_out, ignore=_make_ignore(out_root, excluded)) - # Drop any copied taskset files and the source Dockerfile name we don't use. - for stale in env_out.glob("*.json*"): - stale.unlink() + # The exact JSON/JSONL taskset source was excluded during copying; unrelated + # JSON build inputs remain part of the environment context. for name in ("Dockerfile.hud", "dockerfile"): leftover = env_out / name if leftover.exists() and leftover.name != "Dockerfile": @@ -360,7 +395,12 @@ def _write_environment( _write_text(env_out / "Dockerfile", _adapt_env_dockerfile(dockerfile.read_text("utf-8"))) _write_text( env_out / "hud_entrypoint.sh", - _ENTRYPOINT_SH.format(port=CONTROL_PORT, task=task, args_json=json.dumps(args)), + _ENTRYPOINT_SH.format( + port=CONTROL_PORT, + env_target=shlex.quote(env_target), + task=shlex.quote(task), + args_json=shlex.quote(json.dumps(args)), + ), ) @@ -390,12 +430,16 @@ async def export( # Rows reference envs by name; collect the authored envs (defined in the # source, or next to a tasks file) to materialize prompts in-process. scan = source_dir if src.suffix in (".json", ".jsonl") else src - authored = { - env.name: env - for module in iter_modules(scan) - for env in vars(module).values() - if isinstance(env, Environment) - } + authored: dict[str, list[_AuthoredEnvironment]] = {} + for module in iter_modules(scan): + module_file = getattr(module, "__file__", None) + if module_file is None: + continue + for symbol, env in vars(module).items(): + if isinstance(env, Environment): + authored.setdefault(env.name, []).append( + _AuthoredEnvironment(env, Path(module_file).resolve(), symbol), + ) dockerfile = _find_dockerfile(source_dir) if dockerfile is None: @@ -404,12 +448,22 @@ async def export( "build context to rebuild the image under Harbor.", ) + normalized_tasks = [(_slugify(task.slug or task.default_slug()), task) for task in tasks] + seen_slugs: set[str] = set() + for slug, _task in normalized_tasks: + if slug in seen_slugs: + raise ValueError( + f"multiple HUD task slugs normalize to Harbor folder {slug!r}; " + "give each task a distinct slug.", + ) + seen_slugs.add(slug) + created: list[Path] = [] - for task in tasks: - env = _resolve_env(task, authored) + for slug, task in normalized_tasks: + authored_env = _resolve_env(task, authored) + env = authored_env.env _check_capabilities(env) - slug = task.slug or task.default_slug() task_dir = out / slug (task_dir / "tests").mkdir(parents=True, exist_ok=True) @@ -422,15 +476,27 @@ async def export( _harbor_task_toml(slug, task.id, task.args, timeout_sec), ) - _write_environment(task_dir, source_dir, dockerfile, task.id, task.args, out) + env_module = authored_env.module_path.relative_to(source_dir).as_posix() + env_target = f"{env_module}:{authored_env.symbol}" + taskset_source = src if src.suffix in (".json", ".jsonl") else None + _write_environment( + task_dir, + source_dir, + dockerfile, + task.id, + task.args, + env_target, + out, + taskset_source, + ) _write_text( task_dir / "tests" / "test.sh", _TEST_SH.format( port=CONTROL_PORT, - task=task.id, - args_json=json.dumps(task.args), - answer_file=answer_file, + task=shlex.quote(task.id), + args_json=shlex.quote(json.dumps(task.args)), + answer_file=shlex.quote(answer_file), ), ) @@ -443,6 +509,7 @@ async def export( "ALLOWED_PROTOCOLS", "CONTROL_PORT", "DEFAULT_ANSWER_FILE", + "HarborRuntime", "detect", "export", "load", diff --git a/integrations/harbor_common.py b/integrations/harbor_common.py new file mode 100644 index 000000000..53294e091 --- /dev/null +++ b/integrations/harbor_common.py @@ -0,0 +1,70 @@ +"""Shared helpers for Harbor task integration.""" + +from __future__ import annotations + +import hashlib +import logging +import re +import tomllib +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +LOGGER = logging.getLogger(__name__) + + +def _slugify(name: str) -> str: + """A valid env name (lowercase ``[a-z0-9-]``) from a dataset dir name.""" + normalized = re.sub(r"[^a-z0-9-]", "", name.strip().lower().replace(" ", "-").replace("_", "-")) + return re.sub(r"-+", "-", normalized).strip("-") or "harbor" + + +def _is_harbor_task(path: Path) -> bool: + return path.is_dir() and (path / "task.toml").exists() and (path / "instruction.md").exists() + + +def _task_dirs(path: str | Path) -> list[Path]: + root = Path(path) + if _is_harbor_task(root): + return [root] + if root.is_dir(): + return sorted(d for d in root.iterdir() if d.is_dir() and _is_harbor_task(d)) + return [] + + +def _hash_directory(path: Path) -> str: + """Content-hash a directory for grouping tasks by identical environments.""" + hasher = hashlib.sha256() + if not path.exists(): + return "empty" + for file_path in sorted(path.rglob("*")): + if file_path.is_file(): + hasher.update(str(file_path.relative_to(path)).encode()) + hasher.update(file_path.read_bytes()) + return hasher.hexdigest()[:16] + + +@dataclass(frozen=True, slots=True) +class _HarborTask: + """One parsed Harbor task dir.""" + + task_id: str + config: dict[str, Any] + env_hash: str + + +def _parse_task(task_dir: Path) -> _HarborTask | None: + if not (task_dir / "instruction.md").is_file(): + LOGGER.warning("failed to read instruction.md in %s", task_dir) + return None + try: + config: dict[str, Any] = tomllib.loads((task_dir / "task.toml").read_text("utf-8")) + except (OSError, tomllib.TOMLDecodeError): + LOGGER.warning("failed to parse task.toml in %s", task_dir) + config = {} + env_dir = task_dir / "environment" + return _HarborTask( + task_id=task_dir.name, + config=config, + env_hash=_hash_directory(env_dir) if env_dir.exists() else "no-env", + ) diff --git a/integrations/harbor_runtime.py b/integrations/harbor_runtime.py new file mode 100644 index 000000000..addcb872b --- /dev/null +++ b/integrations/harbor_runtime.py @@ -0,0 +1,984 @@ +"""Local Docker-backed runtime for Harbor task directories.""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import math +import os +import re +import shutil +import stat +import tempfile +import tomllib +import uuid +from collections.abc import AsyncGenerator # noqa: TC003 - env.template resolves this at runtime. +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any, Literal + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + ValidationError, + field_validator, + model_validator, +) + +from hud.environment import Environment +from hud.environment.workspace import Workspace +from integrations.harbor_common import _hash_directory, _slugify, _task_dirs + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Coroutine, Mapping + + from hud.eval import Task + from hud.eval.runtime import Runtime + + +class _PhaseConfig(BaseModel): + model_config = ConfigDict(extra="forbid") + + timeout_sec: float | None = Field(default=None, gt=0) + user: str | int | None = None + network_mode: Literal["public", "no-network", "allowlist"] | None = None + allowed_hosts: list[str] | None = None + + +class _EnvironmentConfig(BaseModel): + model_config = ConfigDict(extra="forbid") + + build_timeout_sec: float = Field(default=600.0, gt=0) + docker_image: str | None = Field(default=None, min_length=1) + os: Literal["linux", "windows"] = "linux" + cpus: int | None = Field(default=None, gt=0) + memory_mb: int | None = Field(default=None, gt=0) + storage_mb: int | None = Field(default=None, gt=0) + gpus: int | None = Field(default=None, ge=0) + gpu_types: list[str] | None = None + tpu: dict[str, Any] | None = None + mcp_servers: list[dict[str, Any]] = Field(default_factory=list) + env: dict[str, str] = Field(default_factory=dict) + skills_dir: str | None = None + healthcheck: dict[str, Any] | None = None + workdir: str | None = None + network_mode: Literal["public", "no-network", "allowlist"] = "public" + allowed_hosts: list[str] | None = None + allow_internet: bool | None = None + + @field_validator("os", mode="before") + @classmethod + def _normalize_os(cls, value: Any) -> Any: + return value.lower() if isinstance(value, str) else value + + @model_validator(mode="before") + @classmethod + def _accept_legacy_sizes(cls, value: Any) -> Any: + if not isinstance(value, dict): + return value + migrated = dict(value) + for legacy, current in (("memory", "memory_mb"), ("storage", "storage_mb")): + if legacy not in migrated: + continue + parsed = _parse_size_mb(migrated.pop(legacy), field=legacy) + if current in migrated and migrated[current] != parsed: + raise ValueError(f"conflicting Harbor environment values: {legacy} and {current}") + migrated.setdefault(current, parsed) + return migrated + + +class _VerifierConfig(_PhaseConfig): + timeout_sec: float | None = Field(default=600.0, gt=0) + env: dict[str, str] = Field(default_factory=dict) + environment_mode: Literal["shared", "separate"] | None = None + environment: dict[str, Any] | None = None + collect: list[dict[str, Any]] = Field(default_factory=list) + + +class _RuntimeTaskConfig(BaseModel): + model_config = ConfigDict(extra="forbid") + + schema_version: str = Field(default="1.4", min_length=1) + task: dict[str, Any] | None = None + metadata: dict[str, Any] = Field(default_factory=dict) + verifier: _VerifierConfig = Field(default_factory=_VerifierConfig) + agent: _PhaseConfig = Field(default_factory=_PhaseConfig) + environment: _EnvironmentConfig = Field(default_factory=_EnvironmentConfig) + solution: dict[str, Any] = Field(default_factory=dict) + source: str | None = None + multi_step_reward_strategy: str | None = None + steps: list[dict[str, Any]] | None = None + artifacts: list[str | dict[str, Any]] = Field(default_factory=list) + + @model_validator(mode="before") + @classmethod + def _accept_legacy_version_key(cls, value: Any) -> Any: + if isinstance(value, dict) and "version" in value: + value = dict(value) + value.setdefault("schema_version", value.pop("version")) + value.pop("version", None) + return value + + +def _load_runtime_config(task_dir: Path) -> _RuntimeTaskConfig: + config_path = task_dir / "task.toml" + try: + raw = tomllib.loads(config_path.read_text(encoding="utf-8")) + config = _RuntimeTaskConfig.model_validate(raw) + except (OSError, tomllib.TOMLDecodeError, ValidationError) as exc: + raise ValueError(f"invalid Harbor task config {config_path}: {exc}") from exc + + unsupported: list[str] = [] + if config.steps: + unsupported.append("multi-step tasks") + if config.multi_step_reward_strategy is not None: + unsupported.append("multi_step_reward_strategy") + if config.artifacts: + unsupported.append("artifact collection") + if config.environment.os != "linux": + unsupported.append("Windows containers") + if config.environment.network_mode != "public" or config.environment.allowed_hosts: + unsupported.append("restricted environment network policy") + if ( + config.environment.allow_internet is False + and "network_mode" not in config.environment.model_fields_set + and "allowed_hosts" not in config.environment.model_fields_set + ): + unsupported.append("environment.allow_internet=false") + if config.environment.tpu is not None: + unsupported.append("TPU resources") + if config.environment.mcp_servers: + unsupported.append("environment MCP servers") + if config.environment.skills_dir is not None: + unsupported.append("environment skills_dir") + if config.environment.healthcheck is not None: + unsupported.append("task-config healthcheck") + if config.environment.gpu_types: + unsupported.append("GPU type selection") + if config.environment.gpus: + unsupported.append("GPU resources") + if config.environment.workdir is not None: + workdir = PurePosixPath(config.environment.workdir) + if not workdir.is_absolute() or ".." in workdir.parts: + raise ValueError( + f"invalid Harbor workdir {config.environment.workdir!r}: " + "expected an absolute container path without '..'" + ) + for label, phase in (("agent", config.agent), ("verifier", config.verifier)): + if phase.network_mode not in (None, "public") or phase.allowed_hosts: + unsupported.append(f"{label} network override") + if config.verifier.environment_mode == "separate" or config.verifier.environment is not None: + unsupported.append("separate verifier environment") + if config.verifier.collect: + unsupported.append("verifier collect hooks") + if unsupported: + rendered = ", ".join(dict.fromkeys(unsupported)) + raise ValueError( + f"HarborRuntime does not support {rendered} in {config_path}; " + "run this task through Harbor proper or remove the unsupported declaration" + ) + + config.environment.env = _resolve_env_vars( + config.environment.env, + label=f"[environment].env in {config_path}", + ) + config.verifier.env = _resolve_env_vars( + config.verifier.env, + label=f"[verifier].env in {config_path}", + ) + _validate_env_values(config.environment.env) + _validate_env_values(config.verifier.env) + return config + + +class HarborRuntime: + """Run Harbor task directories through HUD's local rollout engine. + + The provider builds the Harbor task's ``environment/`` Docker context and + starts a fresh container for agent work. If the task ships a Compose file, + its ``main`` service and sidecars use the task-authored startup semantics. HUD's SSH + capability executes every shell and file operation through ``docker exec``, + so edits land directly in the real container workdir without shadowing image + contents with a host bind mount. Hidden tests are mounted from an initially + empty staging directory and copied into it only after the agent finishes. + Grading then runs ``tests/test.sh`` in the shared main container, bounded by + ``[verifier].timeout_sec``, and reads ``reward.json`` or ``reward.txt``. + """ + + def __init__( + self, + path: str | Path, + *, + ready_timeout: float = 120.0, + reward_key: str | None = None, + ) -> None: + if reward_key is not None and not reward_key: + raise ValueError("HarborRuntime reward_key must be non-empty") + self.root = Path(path).resolve() + self.ready_timeout = ready_timeout + self.reward_key = reward_key + self._task_dirs = {task_dir.name: task_dir for task_dir in _task_dirs(self.root)} + self._image_names: dict[Path, str] = {} + self._build_locks: dict[str, asyncio.Lock] = {} + if not self._task_dirs: + raise ValueError(f"no Harbor tasks found in {path}") + + @contextlib.asynccontextmanager + async def __call__(self, task: Task) -> AsyncIterator[Runtime]: + from hud.eval.runtime import LocalRuntime, Runtime + + task_dir = self._task_dirs.get(task.id) + if task_dir is None: + raise KeyError(f"HarborRuntime has no task directory for {task.id!r}") + env_dir = task_dir / "environment" + tests_dir = task_dir / "tests" + config = _load_runtime_config(task_dir) + compose_file = _compose_file(env_dir) + if ( + config.environment.docker_image is None + and not (env_dir / "Dockerfile").is_file() + and compose_file is None + ): + raise FileNotFoundError( + f"Harbor task {task.id!r} needs environment/Dockerfile, " + "environment/docker-compose.yaml, or [environment].docker_image" + ) + if not (tests_dir / "test.sh").is_file(): + raise FileNotFoundError(f"Harbor task {task.id!r} has no tests/test.sh") + + tmp_path = Path(tempfile.mkdtemp(prefix=f"hud-harbor-{_slugify(task.id)}-")) + try: + control_root = tmp_path / "control" + staged_tests = tmp_path / "tests" + logs = tmp_path / "logs" + control_root.mkdir() + staged_tests.mkdir() + _prepare_host_logs(logs) + acquire = self._compose_container( + task, + env_dir, + compose_file, + staged_tests, + logs, + config, + ) + async with acquire as (container, provider, workdir): + env = self._environment_for( + task, + task_dir, + control_root, + workdir, + tests_dir, + staged_tests, + logs, + container, + config, + ) + local = LocalRuntime(lambda _task: env, ready_timeout=self.ready_timeout) + async with local(task) as runtime: + yield Runtime( + runtime.url, + params={ + **runtime.params, + "provider": provider, + "container": container, + "ready_timeout": self.ready_timeout, + }, + config=runtime.config, + agent_timeout_s=config.agent.timeout_sec, + ) + finally: + await _shielded_cleanup(asyncio.to_thread(_remove_tree, tmp_path)) + + @contextlib.asynccontextmanager + async def _compose_container( + self, + task: Task, + env_dir: Path, + compose_file: Path | None, + staged_tests: Path, + logs: Path, + config: _RuntimeTaskConfig, + ) -> AsyncIterator[tuple[str, str, str]]: + from hud.eval.runtime import _docker + + project = f"hud-harbor-{_slugify(task.id)}-{uuid.uuid4().hex[:8]}" + control_dir = staged_tests.parent + base = control_dir / "compose.base.json" + overlay = control_dir / "compose.hud.json" + compose_env_file = control_dir / "compose.env" + compose_env: dict[str, str] = {} + compose_args: list[str] = [] + container = "" + try: + try: + async with asyncio.timeout(config.environment.build_timeout_sec): + main_image = await self._main_image_name(env_dir) + compose_env = { + **config.environment.env, + **_compose_infra_env( + env_dir=env_dir, + logs=logs, + main_image=main_image, + docker_image=config.environment.docker_image, + environment=config.environment, + ), + } + await asyncio.to_thread( + _write_compose_control_files, + base=base, + overlay=overlay, + compose_env_file=compose_env_file, + staged_tests=staged_tests, + logs=logs, + environment=config.environment, + compose_env=compose_env, + ) + compose_files = [ + base, + *([compose_file] if compose_file is not None else []), + overlay, + ] + compose_args = [ + "compose", + "--env-file", + str(compose_env_file), + "--project-directory", + str(env_dir), + "-p", + project, + ] + for path in compose_files: + compose_args.extend(["-f", str(path)]) + if config.environment.docker_image is None: + lock = self._build_locks.setdefault(main_image, asyncio.Lock()) + async with lock: + await _docker( + *compose_args, + "build", + unset_env=tuple(compose_env), + ) + await _docker( + *compose_args, + "up", + "--detach", + "--wait", + unset_env=tuple(compose_env), + ) + out, _ = await _docker( + *compose_args, + "ps", + "-q", + "main", + unset_env=tuple(compose_env), + ) + container = out.strip() + if not container: + raise RuntimeError( + f"docker compose project {project} did not create a main service" + ) + workdir = config.environment.workdir or await _container_workdir(container) + await _prepare_container_logs(container) + if _should_upload_environment_dir( + env_dir=env_dir, + docker_image=config.environment.docker_image, + ): + await _docker("cp", f"{env_dir}/.", f"{container}:{workdir}") + except TimeoutError as exc: + raise TimeoutError( + "Harbor environment build/start timed out after " + f"{config.environment.build_timeout_sec:.0f}s" + ) from exc + yield container, "harbor-compose", workdir + finally: + if compose_args: + + async def cleanup() -> None: + if container: + with contextlib.suppress(Exception): + await _release_log_permissions(container) + await _docker( + *compose_args, + "down", + "--volumes", + "--remove-orphans", + "--rmi", + "local", + check=False, + unset_env=tuple(compose_env), + ) + + await _shielded_cleanup(cleanup()) + + async def _main_image_name(self, env_dir: Path) -> str: + cached = self._image_names.get(env_dir) + if cached is not None: + return cached + digest = await asyncio.to_thread(_hash_directory, env_dir) + image = f"hud-harbor:{digest}" + return self._image_names.setdefault(env_dir, image) + + def _environment_for( + self, + task: Task, + task_dir: Path, + control_root: Path, + workdir: str, + tests_dir: Path, + staged_tests: Path, + logs: Path, + container: str, + config: _RuntimeTaskConfig, + ) -> Environment: + env = Environment(task.env) + workspace_daemon = _DockerWorkspace( + control_root, + container=container, + guest_path=workdir, + exec_user=config.agent.user, + ) + verifier_timeout = config.verifier.timeout_sec or _DEFAULT_VERIFIER_TIMEOUT + + @env.initialize + async def _up() -> None: + await workspace_daemon.start() + env.add_capability(workspace_daemon.capability("shell")) + + @env.shutdown + async def _down() -> None: + await workspace_daemon.stop() + + @env.template(id=task.id, description=f"Harbor task {task.id}") + async def _run_harbor_task() -> AsyncGenerator[Any, Any]: + answer = yield (task_dir / "instruction.md").read_text(encoding="utf-8") + yield await self._grade( + container, + workdir, + tests_dir, + staged_tests, + logs, + answer, + verifier_timeout=verifier_timeout, + verifier_user=config.verifier.user, + verifier_env={**config.environment.env, **config.verifier.env}, + ) + + return env + + async def _grade( + self, + container: str, + workdir: str, + tests_dir: Path, + staged_tests: Path, + logs: Path, + answer: Any, + *, + verifier_timeout: float, + verifier_user: str | int | None, + verifier_env: dict[str, str], + ) -> dict[str, Any]: + answer_file = staged_tests.parent / "agent_answer.txt" + await asyncio.to_thread( + answer_file.write_text, + "" if answer is None else str(answer), + encoding="utf-8", + ) + await _release_log_permissions(container) + await asyncio.to_thread( + _prepare_verifier_staging, + tests_dir, + staged_tests, + logs / "verifier", + ) + verifier_env_file = staged_tests.parent / "verifier.env" + _write_docker_env_file(verifier_env_file, verifier_env) + exec_args = ["exec", "--workdir", workdir] + if verifier_user is not None: + exec_args.extend(["--user", str(verifier_user)]) + if verifier_env: + exec_args.extend(["--env-file", str(verifier_env_file)]) + exec_args.extend([container, "/tests/test.sh"]) + from hud.eval.runtime import _docker + + try: + out, err = await asyncio.wait_for( + _docker(*exec_args, check=False), timeout=verifier_timeout + ) + except TimeoutError: + return { + "score": 0.0, + "isError": True, + "content": f"Harbor verifier timed out after {verifier_timeout:.0f}s", + "info": {"verifier_timeout_sec": verifier_timeout}, + } + await _release_log_permissions(container) + reward, info = await asyncio.to_thread( + _read_harbor_reward, + logs / "verifier", + reward_key=self.reward_key, + ) + info.update( + { + "stdout": out[-4000:], + "stderr": err[-4000:], + } + ) + if reward is None: + parse_error = info.get("reward_parse_error") + return { + "score": 0.0, + "isError": True, + "content": ( + f"Harbor verifier reward is invalid: {parse_error}" + if parse_error + else "Harbor verifier did not write reward.json or reward.txt" + ), + "info": info, + } + return {"score": reward, "info": info} + + +class _DockerWorkspace(Workspace): + """Workspace SSH whose commands and file streams run through ``docker exec``.""" + + def __init__( + self, + *args: Any, + container: str, + exec_user: str | int | None = None, + **kwargs: Any, + ) -> None: + super().__init__(*args, **kwargs) + self._container = container + self._exec_user = exec_user + self._docker = shutil.which("docker") or "docker" + + def shell_argv( + self, + command: str | None = None, + *, + cwd: str | None = None, + env: Mapping[str, str] | None = None, + ) -> list[str]: + argv = [self._docker, "exec", "-i", "--workdir", cwd or self._guest_path] + if self._exec_user is not None: + argv.extend(["--user", str(self._exec_user)]) + for key, value in (env or {}).items(): + argv.extend(["--env", f"{key}={value}"]) + argv.extend([self._container, "bash", "-lc", command or "bash -l"]) + return argv + + +_DEFAULT_VERIFIER_TIMEOUT = 600.0 +_ENV_TEMPLATE_RE = re.compile(r"\$\{([^}:]+)(?::-(.*))?\}") +_ENV_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +_MAX_REWARD_BYTES = 1_000_000 +_CLEANUP_TIMEOUT = 60.0 + + +def _parse_size_mb(value: Any, *, field: str) -> int: + if not isinstance(value, str): + raise ValueError(f"legacy Harbor environment.{field} must be a size string") + normalized = value.strip().upper() + multipliers = {"G": 1024.0, "M": 1.0, "K": 1.0 / 1024.0} + suffix = normalized[-1:] if normalized else "" + if suffix not in multipliers: + raise ValueError(f"invalid Harbor environment.{field} size {value!r}") + try: + return int(float(normalized[:-1]) * multipliers[suffix]) + except (OverflowError, ValueError) as exc: + raise ValueError(f"invalid Harbor environment.{field} size {value!r}") from exc + + +def _resolve_env_vars(values: dict[str, str], *, label: str) -> dict[str, str]: + """Resolve Harbor's ``${VAR}`` / ``${VAR:-default}`` task env syntax.""" + + def resolve(value: str) -> str: + match = _ENV_TEMPLATE_RE.fullmatch(value) + if match is None: + return value + name, default = match.group(1), match.group(2) + if name in os.environ: + return os.environ[name] + if default is not None: + return default + raise ValueError(f"{label} references unset environment variable {name!r}") + + return {key: resolve(value) for key, value in values.items()} + + +async def _container_workdir(container: str) -> str: + """The running container's configured ``WORKDIR``, or Docker's root default.""" + from hud.eval.runtime import _docker + + out, _ = await _docker("inspect", "--format", "{{.Config.WorkingDir}}", container) + return out.strip() or "/" + + +def _clear_directory(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True) + for child in path.iterdir(): + if child.is_dir() and not child.is_symlink(): + shutil.rmtree(child) + else: + child.unlink() + + +def _prepare_host_logs(logs: Path) -> None: + """Create bind-mounted Harbor log paths writable before container startup.""" + logs.mkdir(parents=True, exist_ok=True) + logs.chmod(0o777) + for name in ("agent", "verifier", "artifacts"): + child = logs / name + child.mkdir(exist_ok=True) + child.chmod(0o777) + + +def _prepare_verifier_staging(tests_dir: Path, staged_tests: Path, verifier_logs: Path) -> None: + """Expose hidden tests only after agent work and discard any pre-seeded reward.""" + _clear_directory(verifier_logs) + _clear_directory(staged_tests) + shutil.copytree(tests_dir, staged_tests, dirs_exist_ok=True) + test_script = staged_tests / "test.sh" + test_script.chmod(test_script.stat().st_mode | 0o111) + + +def _read_harbor_reward( + verifier_logs: Path, + *, + reward_key: str | None = None, +) -> tuple[float | None, dict[str, Any]]: + reward_json = verifier_logs / "reward.json" + try: + reward_json_text = _read_regular_text(reward_json) + except ValueError as exc: + return None, { + "reward_file": "reward.json", + "reward_parse_error": str(exc), + } + if reward_json_text is not None: + try: + data = json.loads(reward_json_text) + except json.JSONDecodeError as exc: + return None, { + "reward_file": "reward.json", + "reward_parse_error": f"invalid reward.json: {exc}", + } + if isinstance(data, dict): + converted = {str(key): _finite_number(value) for key, value in data.items()} + invalid_keys = [key for key, value in converted.items() if value is None] + if invalid_keys: + return None, { + "reward_file": "reward.json", + "reward_parse_error": ( + "reward.json contains non-numeric reward(s): " + ", ".join(invalid_keys) + ), + } + rewards = {key: value for key, value in converted.items() if value is not None} + info: dict[str, Any] = { + "reward_file": "reward.json", + "harbor_rewards": rewards, + } + if reward_key is not None: + if reward_key not in rewards: + info["reward_parse_error"] = ( + f"configured reward key {reward_key!r} is missing or non-numeric" + ) + return None, info + info["primary_reward_key"] = reward_key + return rewards[reward_key], info + if "reward" in rewards: + info["primary_reward_key"] = "reward" + return rewards["reward"], info + if len(rewards) == 1: + key, value = next(iter(rewards.items())) + info["primary_reward_key"] = key + return value, info + info["reward_parse_error"] = ( + "reward.json has no unambiguous primary reward; set HarborRuntime(reward_key=...)" + if rewards + else "reward.json has no numeric rewards" + ) + return None, info + return None, { + "reward_file": "reward.json", + "reward_parse_error": "reward.json must be an object of numeric rewards", + } + + reward_txt = verifier_logs / "reward.txt" + try: + reward_text = _read_regular_text(reward_txt) + except ValueError as exc: + return None, { + "reward_file": "reward.txt", + "reward_parse_error": str(exc), + } + if reward_text is not None: + text = reward_text.strip() + try: + reward = float(text) + except (OverflowError, ValueError): + reward = math.nan + if math.isfinite(reward): + return reward, {"reward_file": "reward.txt"} + return None, {"reward_file": "reward.txt", "reward_parse_error": text} + + return None, {} + + +def _finite_number(value: Any) -> float | None: + if not isinstance(value, int | float) or isinstance(value, bool): + return None + try: + converted = float(value) + except (OverflowError, ValueError): + return None + return converted if math.isfinite(converted) else None + + +def _read_regular_text(path: Path) -> str | None: + """Read a small regular file without following an untrusted verifier symlink.""" + try: + metadata = path.lstat() + except FileNotFoundError: + return None + if not stat.S_ISREG(metadata.st_mode): + raise ValueError(f"{path.name} is not a regular file") + if metadata.st_size > _MAX_REWARD_BYTES: + raise ValueError(f"{path.name} exceeds {_MAX_REWARD_BYTES} bytes") + + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise ValueError(f"could not safely open {path.name}: {exc}") from exc + with os.fdopen(descriptor, "rb") as file: + opened = os.fstat(file.fileno()) + if not stat.S_ISREG(opened.st_mode): + raise ValueError(f"{path.name} is not a regular file") + payload = file.read(_MAX_REWARD_BYTES + 1) + if len(payload) > _MAX_REWARD_BYTES: + raise ValueError(f"{path.name} exceeds {_MAX_REWARD_BYTES} bytes") + try: + return payload.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"{path.name} is not valid UTF-8") from exc + + +async def _release_log_permissions(container: str) -> None: + """Let the host user remove verifier logs written by container root.""" + from hud.eval.runtime import _docker + + await _docker( + "exec", + "--user", + "0", + container, + "chmod", + "-R", + "a+rwX", + "/logs", + check=False, + ) + + +async def _prepare_container_logs(container: str) -> None: + """Create Harbor log directories writable by non-root agent/verifier users.""" + from hud.eval.runtime import _docker + + await _docker( + "exec", + "--user", + "0", + container, + "sh", + "-c", + "mkdir -p /logs/agent /logs/verifier /logs/artifacts && chmod -R a+rwX /logs", + ) + + +def _compose_file(env_dir: Path) -> Path | None: + for name in ("docker-compose.yaml", "docker-compose.yml", "compose.yaml", "compose.yml"): + path = env_dir / name + if path.is_file(): + return path + return None + + +def _should_upload_environment_dir(*, env_dir: Path, docker_image: str | None) -> bool: + """Match Harbor's prebuilt-image upload rule for environment-only files.""" + return bool( + docker_image + and env_dir.is_dir() + and not (env_dir / "Dockerfile").exists() + and _compose_file(env_dir) is None + and any(env_dir.iterdir()) + ) + + +def _compose_base(*, docker_image: str | None) -> str: + """Harbor's base ``main`` service: image/build plus its keepalive command.""" + main: dict[str, Any] + if docker_image is not None: + main = {"image": "${PREBUILT_IMAGE_NAME}"} + else: + main = { + "build": {"context": "${CONTEXT_DIR}"}, + "image": "${MAIN_IMAGE_NAME}", + "pull_policy": "build", + } + main["command"] = ["sh", "-c", "sleep infinity"] + return json.dumps({"services": {"main": main}}, indent=2) + "\n" + + +def _compose_overlay( + *, + tests_dir: Path, + logs: Path, + environment: _EnvironmentConfig, +) -> str: + """Final Compose override for staged tests, logs, env, workdir, and limits.""" + main: dict[str, Any] = { + "volumes": [ + f"{tests_dir}:/tests:ro", + f"{logs}:/logs", + ], + } + if environment.env: + # Keep task values literal here. Harbor infrastructure variables win for + # Compose-file interpolation, but a same-named task variable must still + # reach the main container with the value the task declared. + main["environment"] = environment.env + if environment.cpus is not None: + main["cpus"] = float(environment.cpus) + if environment.memory_mb is not None: + main["mem_limit"] = f"{environment.memory_mb}m" + return json.dumps({"services": {"main": main}}, indent=2) + "\n" + + +def _compose_infra_env( + *, + env_dir: Path, + logs: Path, + main_image: str, + docker_image: str | None, + environment: _EnvironmentConfig, +) -> dict[str, str]: + """Protected Harbor variables used by task-authored Compose templates.""" + values = { + "CONTEXT_DIR": str(env_dir.resolve()), + "MAIN_IMAGE_NAME": main_image, + "HOST_AGENT_LOGS_PATH": str((logs / "agent").resolve()), + "ENV_AGENT_LOGS_PATH": "/logs/agent", + "HOST_VERIFIER_LOGS_PATH": str((logs / "verifier").resolve()), + "ENV_VERIFIER_LOGS_PATH": "/logs/verifier", + "HOST_ARTIFACTS_LOGS_PATH": str((logs / "artifacts").resolve()), + "ENV_ARTIFACTS_LOGS_PATH": "/logs/artifacts", + } + if docker_image is not None: + values["PREBUILT_IMAGE_NAME"] = docker_image + if environment.cpus is not None: + values["CPUS"] = str(environment.cpus) + if environment.memory_mb is not None: + values["MEMORY"] = f"{environment.memory_mb}M" + return values + + +def _validate_env_values(values: Mapping[str, str]) -> None: + for key, value in values.items(): + if _ENV_NAME_RE.fullmatch(key) is None: + raise ValueError(f"invalid Harbor environment variable name {key!r}") + if "\x00" in value or "\n" in value or "\r" in value: + raise ValueError( + f"Harbor environment variable {key!r} contains an unsupported newline or NUL" + ) + + +def _write_compose_env_file(path: Path, values: Mapping[str, str]) -> None: + """Write a mode-0600 Compose interpolation file with literal values.""" + _validate_env_values(values) + lines = [] + for key, value in values.items(): + escaped = value.replace("'", "\\'") + lines.append(f"{key}='{escaped}'") + path.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n") + path.chmod(0o600) + + +def _write_compose_control_files( + *, + base: Path, + overlay: Path, + compose_env_file: Path, + staged_tests: Path, + logs: Path, + environment: _EnvironmentConfig, + compose_env: Mapping[str, str], +) -> None: + """Materialize one acquisition's generated Compose inputs off the event loop.""" + base.write_text( + _compose_base(docker_image=environment.docker_image), + encoding="utf-8", + newline="\n", + ) + overlay.write_text( + _compose_overlay( + tests_dir=staged_tests, + logs=logs, + environment=environment, + ), + encoding="utf-8", + newline="\n", + ) + overlay.chmod(0o600) + _write_compose_env_file(compose_env_file, compose_env) + + +def _write_docker_env_file(path: Path, values: Mapping[str, str]) -> None: + """Write a mode-0600 Docker ``--env-file`` for verifier overrides.""" + _validate_env_values(values) + path.write_text( + "".join(f"{key}={value}\n" for key, value in values.items()), + encoding="utf-8", + newline="\n", + ) + path.chmod(0o600) + + +async def _shielded_cleanup( + cleanup: Coroutine[Any, Any, Any], *, timeout_s: float = _CLEANUP_TIMEOUT +) -> None: + """Finish bounded Docker cleanup even when the rollout is cancelled.""" + cleanup_task = asyncio.create_task(cleanup) + + async def wait() -> None: + async with asyncio.timeout(timeout_s): + await asyncio.shield(cleanup_task) + + try: + await wait() + except asyncio.CancelledError: + try: + await wait() + except BaseException: + cleanup_task.cancel() + await asyncio.gather(cleanup_task, return_exceptions=True) + raise + except Exception: + cleanup_task.cancel() + await asyncio.gather(cleanup_task, return_exceptions=True) + + +def _remove_tree(path: Path) -> None: + """Best-effort removal of one runtime-owned temporary directory.""" + if not path.exists(): + return + + def make_writable(function: Any, target: str, _error: Any) -> None: + os.chmod(target, 0o700) + function(target) + + with contextlib.suppress(OSError): + shutil.rmtree(path, onerror=make_writable) diff --git a/integrations/tests/test_harbor.py b/integrations/tests/test_harbor.py index e25930925..625035e81 100644 --- a/integrations/tests/test_harbor.py +++ b/integrations/tests/test_harbor.py @@ -2,12 +2,34 @@ from __future__ import annotations +import asyncio +import json +import subprocess +import sys import textwrap +import tomllib from typing import TYPE_CHECKING import pytest +from typer.testing import CliRunner -from integrations.harbor import detect, export, load +from hud.cli import app +from hud.environment.workspace import Workspace +from hud.eval import Task +from integrations.harbor import HarborRuntime, detect, export, load +from integrations.harbor_runtime import ( + _compose_base, + _compose_file, + _compose_overlay, + _container_workdir, + _DockerWorkspace, + _EnvironmentConfig, + _load_runtime_config, + _prepare_verifier_staging, + _read_harbor_reward, + _shielded_cleanup, + _should_upload_environment_dir, +) from .conftest import make_harbor_task @@ -34,6 +56,11 @@ def test_load_single_task_dir_maps_rows(single_task: Path) -> None: assert row.id == "cancel-async-tasks" assert row.args == {} assert row.env == taskset.name + assert row.columns == { + "category": "systems", + "difficulty": "medium", + "tags": ["bash", "linux"], + } def test_load_dataset_shares_one_env_per_build_context(dataset_same_env: Path) -> None: @@ -74,6 +101,675 @@ def test_load_skips_unparseable_toml_but_keeps_the_rest(tmp_path: Path) -> None: assert {task.id for task in taskset} == {"good", "broken"} +def test_harbor_runtime_accepts_dataset_dirs(single_task: Path) -> None: + runtime = HarborRuntime(single_task.parent) + + assert single_task.name in runtime._task_dirs + + +async def test_shielded_cleanup_finishes_before_propagating_cancellation() -> None: + started = asyncio.Event() + finished = asyncio.Event() + + async def cleanup() -> None: + started.set() + await asyncio.sleep(0.01) + finished.set() + + task = asyncio.create_task(_shielded_cleanup(cleanup())) + await started.wait() + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + assert finished.is_set() + + +def test_runtime_config_rejects_invalid_toml(single_task: Path) -> None: + (single_task / "task.toml").write_text("not [valid toml", encoding="utf-8") + + with pytest.raises(ValueError, match="invalid Harbor task config"): + _load_runtime_config(single_task) + + +def test_runtime_config_rejects_unknown_fields(single_task: Path) -> None: + (single_task / "task.toml").write_text( + "[environment]\nunknown_runtime_knob = true\n", encoding="utf-8" + ) + + with pytest.raises(ValueError, match="invalid Harbor task config"): + _load_runtime_config(single_task) + + +@pytest.mark.parametrize( + ("task_toml", "feature"), + [ + ('[environment]\nos = "windows"\n', "Windows containers"), + ("steps = [{}]\n", "multi-step tasks"), + ('[verifier]\nenvironment_mode = "separate"\n', "separate verifier environment"), + ('[environment]\nnetwork_mode = "no-network"\n', "restricted environment network policy"), + ], +) +def test_runtime_config_fails_closed_for_unsupported_features( + tmp_path: Path, + task_toml: str, + feature: str, +) -> None: + task_dir = make_harbor_task(tmp_path, "unsupported", task_toml=task_toml) + + with pytest.raises(ValueError, match=feature): + _load_runtime_config(task_dir) + + +def test_runtime_config_accepts_legacy_sizes_and_uppercase_linux(tmp_path: Path) -> None: + task_dir = make_harbor_task( + tmp_path, + "legacy-runtime", + task_toml=textwrap.dedent("""\ + version = "1.3" + + [environment] + os = "LINUX" + memory = "4G" + storage = "10240M" + """), + ) + + config = _load_runtime_config(task_dir) + + assert config.schema_version == "1.3" + assert config.environment.os == "linux" + assert config.environment.memory_mb == 4096 + assert config.environment.storage_mb == 10240 + + +def test_runtime_config_resolves_environment_templates( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("HARBOR_REQUIRED", "from-host") + task_dir = make_harbor_task( + tmp_path, + "environment-templates", + task_toml=textwrap.dedent("""\ + [environment.env] + REQUIRED = "${HARBOR_REQUIRED}" + DEFAULTED = "${HARBOR_UNSET:-fallback}" + LITERAL = "plain" + + [verifier.env] + VERIFIER_DEFAULT = "${HARBOR_VERIFIER_UNSET:-verifier-fallback}" + """), + ) + + config = _load_runtime_config(task_dir) + + assert config.environment.env == { + "REQUIRED": "from-host", + "DEFAULTED": "fallback", + "LITERAL": "plain", + } + assert config.verifier.env == {"VERIFIER_DEFAULT": "verifier-fallback"} + + +async def test_compose_container_isolates_task_env_and_cleans_up_after_failed_start( + single_task: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + secret = "secret-value-never-in-docker-argv" + (single_task / "task.toml").write_text( + textwrap.dedent("""\ + schema_version = "1.3" + + [environment] + build_timeout_sec = 12 + cpus = 3 + memory_mb = 2048 + workdir = "/srv/app" + + [environment.env] + HARBOR_SENTINEL = "visible" + DOCKER_HOST = "task-controlled-docker-host" + MAIN_IMAGE_NAME = "task-must-not-override-infrastructure" + """), + encoding="utf-8", + ) + with (single_task / "task.toml").open("a", encoding="utf-8") as config_file: + config_file.write(f'SECRET = "{secret}"\n') + compose = single_task / "environment" / "docker-compose.yaml" + compose.write_text("services:\n main:\n build: .\n", encoding="utf-8") + calls: list[tuple[tuple[str, ...], bool, dict[str, str] | None, tuple[str, ...]]] = [] + + async def fake_docker( + *args: str, + check: bool = True, + env: dict[str, str] | None = None, + unset_env: tuple[str, ...] = (), + ) -> tuple[str, str]: + calls.append((args, check, env, unset_env)) + if args[-3:] == ("up", "--detach", "--wait"): + raise RuntimeError("compose failed") + return "", "" + + monkeypatch.setattr("hud.eval.runtime._docker", fake_docker) + runtime = HarborRuntime(single_task.parent) + config = _load_runtime_config(single_task) + staged_tests = tmp_path / "staged-tests" + staged_tests.mkdir() + + with pytest.raises(RuntimeError, match="compose failed"): + async with runtime._compose_container( + Task(env="bench", id=single_task.name), + single_task / "environment", + compose, + staged_tests, + tmp_path / "logs", + config, + ): + raise AssertionError("compose acquisition should not yield") + + up_args, _, up_env, up_unset_env = next(call for call in calls if "up" in call[0]) + assert up_args[up_args.index("--project-directory") + 1] == str(single_task / "environment") + assert up_args.count("-f") == 3 + assert up_env is None + assert { + "HARBOR_SENTINEL", + "DOCKER_HOST", + "SECRET", + "CONTEXT_DIR", + "MAIN_IMAGE_NAME", + "HOST_AGENT_LOGS_PATH", + "ENV_AGENT_LOGS_PATH", + "HOST_VERIFIER_LOGS_PATH", + "ENV_VERIFIER_LOGS_PATH", + "HOST_ARTIFACTS_LOGS_PATH", + "ENV_ARTIFACTS_LOGS_PATH", + "CPUS", + "MEMORY", + } <= set(up_unset_env) + assert secret not in " ".join(up_args) + + compose_env_path = tmp_path / "compose.env" + compose_env = compose_env_path.read_text("utf-8") + assert f"SECRET='{secret}'" in compose_env + assert "DOCKER_HOST='task-controlled-docker-host'" in compose_env + assert f"CONTEXT_DIR='{(single_task / 'environment').resolve()}'" in compose_env + assert "MAIN_IMAGE_NAME='hud-harbor:" in compose_env + assert "task-must-not-override-infrastructure" not in compose_env + assert f"HOST_AGENT_LOGS_PATH='{(tmp_path / 'logs' / 'agent').resolve()}'" in compose_env + assert "ENV_AGENT_LOGS_PATH='/logs/agent'" in compose_env + assert f"HOST_VERIFIER_LOGS_PATH='{(tmp_path / 'logs' / 'verifier').resolve()}'" in compose_env + assert "ENV_VERIFIER_LOGS_PATH='/logs/verifier'" in compose_env + artifacts_logs = (tmp_path / "logs" / "artifacts").resolve() + assert f"HOST_ARTIFACTS_LOGS_PATH='{artifacts_logs}'" in compose_env + assert "ENV_ARTIFACTS_LOGS_PATH='/logs/artifacts'" in compose_env + assert "CPUS='3'" in compose_env + assert "MEMORY='2048M'" in compose_env + assert compose_env_path.stat().st_mode & 0o777 == 0o600 + + base = json.loads((tmp_path / "compose.base.json").read_text("utf-8"))["services"]["main"] + assert base["build"] == {"context": "${CONTEXT_DIR}"} + assert base["image"] == "${MAIN_IMAGE_NAME}" + overlay = json.loads((tmp_path / "compose.hud.json").read_text("utf-8"))["services"]["main"] + # ``environment.workdir`` controls agent/verifier exec cwd only; Compose + # must retain the task-authored main service working_dir. + assert "working_dir" not in overlay + assert overlay["cpus"] == 3.0 + assert overlay["mem_limit"] == "2048m" + assert overlay["environment"] == { + "HARBOR_SENTINEL": "visible", + "DOCKER_HOST": "task-controlled-docker-host", + "MAIN_IMAGE_NAME": "task-must-not-override-infrastructure", + "SECRET": secret, + } + assert (tmp_path / "compose.hud.json").stat().st_mode & 0o777 == 0o600 + assert any( + args[-5:] == ("down", "--volumes", "--remove-orphans", "--rmi", "local") and check is False + for args, check, _, _ in calls + ) + + +async def test_plain_dockerfile_uses_compose_wait_readiness_and_cleanup( + single_task: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[tuple[str, ...], bool, tuple[str, ...]]] = [] + + async def fake_docker( + *args: str, + check: bool = True, + env: dict[str, str] | None = None, + unset_env: tuple[str, ...] = (), + ) -> tuple[str, str]: + assert env is None + calls.append((args, check, unset_env)) + if args[-3:] == ("ps", "-q", "main"): + return "container-id\n", "" + if args[:3] == ("inspect", "--format", "{{.Config.WorkingDir}}"): + return "/workspace\n", "" + return "", "" + + monkeypatch.setattr("hud.eval.runtime._docker", fake_docker) + runtime = HarborRuntime(single_task.parent) + config = _load_runtime_config(single_task) + staged_tests = tmp_path / "staged-tests" + staged_tests.mkdir() + + async with runtime._compose_container( + Task(env="bench", id=single_task.name), + single_task / "environment", + None, + staged_tests, + tmp_path / "logs", + config, + ) as (container, provider, workdir): + assert (container, provider, workdir) == ( + "container-id", + "harbor-compose", + "/workspace", + ) + + command_tails = [args[-3:] for args, _, _ in calls] + assert any(args[-1:] == ("build",) for args, _, _ in calls) + assert ("up", "--detach", "--wait") in command_tails + up_args = next(args for args, _, _ in calls if args[-3:] == ("up", "--detach", "--wait")) + assert up_args.count("-f") == 2 + assert any( + args[-5:] == ("down", "--volumes", "--remove-orphans", "--rmi", "local") and check is False + for args, check, _ in calls + ) + + +def test_compose_file_detection_prefers_harbor_names(tmp_path: Path) -> None: + env = tmp_path / "environment" + env.mkdir() + compose = env / "docker-compose.yaml" + compose.write_text("services: {}\n", encoding="utf-8") + + assert _compose_file(env) == compose + + +def test_compose_base_builds_or_uses_prebuilt_main() -> None: + built = json.loads(_compose_base(docker_image=None))["services"]["main"] + prebuilt = json.loads(_compose_base(docker_image="registry/task:v1"))["services"]["main"] + + assert built["build"] == {"context": "${CONTEXT_DIR}"} + assert built["pull_policy"] == "build" + assert built["image"] == "${MAIN_IMAGE_NAME}" + assert prebuilt["image"] == "${PREBUILT_IMAGE_NAME}" + assert "build" not in prebuilt + assert built["command"] == prebuilt["command"] == ["sh", "-c", "sleep infinity"] + + +def test_compose_overlay_mounts_staged_tests_logs_and_runtime_config(tmp_path: Path) -> None: + overlay = json.loads( + _compose_overlay( + tests_dir=tmp_path / "tests", + logs=tmp_path / "logs", + environment=_EnvironmentConfig( + workdir="/srv/app", + cpus=4, + memory_mb=8192, + env={"SERVICE_TOKEN": "resolved"}, + ), + ) + )["services"]["main"] + + assert "working_dir" not in overlay + assert overlay["environment"] == {"SERVICE_TOKEN": "resolved"} + assert overlay["cpus"] == 4.0 + assert overlay["mem_limit"] == "8192m" + assert overlay["volumes"] == [ + f"{tmp_path / 'tests'}:/tests:ro", + f"{tmp_path / 'logs'}:/logs", + ] + + +async def test_container_workdir_reads_running_container_config( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def fake_docker(*args: str, check: bool = True) -> tuple[str, str]: + assert args == ("inspect", "--format", "{{.Config.WorkingDir}}", "container-id") + return "/srv/app\n", "" + + monkeypatch.setattr("hud.eval.runtime._docker", fake_docker) + + assert await _container_workdir("container-id") == "/srv/app" + + +async def test_container_workdir_defaults_to_docker_root_when_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def fake_docker(*args: str, check: bool = True) -> tuple[str, str]: + return "\n", "" + + monkeypatch.setattr("hud.eval.runtime._docker", fake_docker) + + assert await _container_workdir("container-id") == "/" + + +def test_prebuilt_environment_upload_requires_environment_only_files(tmp_path: Path) -> None: + env_dir = tmp_path / "environment" + env_dir.mkdir() + assert not _should_upload_environment_dir(env_dir=env_dir, docker_image="task:v1") + + payload = env_dir / "fixture.txt" + payload.write_text("fixture", encoding="utf-8") + assert _should_upload_environment_dir(env_dir=env_dir, docker_image="task:v1") + assert not _should_upload_environment_dir(env_dir=env_dir, docker_image=None) + + dockerfile = env_dir / "Dockerfile" + dockerfile.write_text("FROM task:v1\n", encoding="utf-8") + assert not _should_upload_environment_dir(env_dir=env_dir, docker_image="task:v1") + dockerfile.unlink() + (env_dir / "compose.yaml").write_text("services: {}\n", encoding="utf-8") + assert not _should_upload_environment_dir(env_dir=env_dir, docker_image="task:v1") + + +def test_docker_workspace_shell_argv_preserves_stdin_user_and_per_call_env(tmp_path: Path) -> None: + workspace = _DockerWorkspace( + tmp_path / "control", + container="container-id", + guest_path="/workspace", + exec_user=1001, + ) + + argv = workspace.shell_argv( + "cat > result.txt", + cwd="/srv/app", + env={"OVERRIDE": "new", "PER_CALL": "yes"}, + ) + + assert argv[0].endswith("docker") + assert argv[1:5] == ["exec", "-i", "--workdir", "/srv/app"] + assert argv[5:7] == ["--user", "1001"] + assert "OVERRIDE=new" in argv + assert "PER_CALL=yes" in argv + assert argv[-4:] == ["container-id", "bash", "-lc", "cat > result.txt"] + # The current Workspace implementation owns the stdin relay. Keeping this + # inherited prevents Docker-backed file writes from silently becoming empty. + assert "_handle_process" not in _DockerWorkspace.__dict__ + assert _DockerWorkspace._handle_process is Workspace._handle_process + + +def test_read_harbor_reward_prefers_reward_key_and_preserves_named_metrics(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text( + json.dumps({"reward": 0.5, "passed": 3, "total": 5}), "utf-8" + ) + + reward, info = _read_harbor_reward(verifier) + + assert reward == 0.5 + assert info["reward_file"] == "reward.json" + assert info["primary_reward_key"] == "reward" + assert info["harbor_rewards"] == {"reward": 0.5, "passed": 3.0, "total": 5.0} + + +def test_read_harbor_reward_selects_explicit_named_metric(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text(json.dumps({"tests_passed": 0.8, "style": 0.25}), "utf-8") + + reward, info = _read_harbor_reward(verifier, reward_key="tests_passed") + + assert reward == 0.8 + assert info["primary_reward_key"] == "tests_passed" + assert info["harbor_rewards"] == {"tests_passed": 0.8, "style": 0.25} + + +def test_read_harbor_reward_selects_a_single_named_metric(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text(json.dumps({"quality": 0.75}), "utf-8") + + reward, info = _read_harbor_reward(verifier) + + assert reward == 0.75 + assert info["primary_reward_key"] == "quality" + + +def test_read_harbor_reward_rejects_ambiguous_named_metrics(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text(json.dumps({"passed": 3, "total": 5}), "utf-8") + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info["harbor_rewards"] == {"passed": 3.0, "total": 5.0} + assert "no unambiguous primary reward" in info["reward_parse_error"] + + +def test_read_harbor_reward_does_not_special_case_score(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text( + json.dumps({"score": 1.0, "style": 0.5}), encoding="utf-8" + ) + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info["reward_file"] == "reward.json" + assert "no unambiguous primary reward" in info["reward_parse_error"] + + +@pytest.mark.parametrize( + "invalid_reward", + [float("nan"), float("inf"), float("-inf"), 10**1000], + ids=["nan", "positive-infinity", "negative-infinity", "huge-integer"], +) +def test_read_harbor_reward_rejects_non_finite_numbers( + tmp_path: Path, + invalid_reward: float | int, +) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text(json.dumps({"reward": invalid_reward}), encoding="utf-8") + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info["reward_file"] == "reward.json" + assert "non-numeric reward(s): reward" in info["reward_parse_error"] + + +def test_read_harbor_reward_rejects_oversized_reward_file(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text(" " * 1_000_001, encoding="utf-8") + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info == { + "reward_file": "reward.json", + "reward_parse_error": "reward.json exceeds 1000000 bytes", + } + + +@pytest.mark.parametrize("invalid_reward", ["NaN", "Infinity", "-Infinity", "1e1000000"]) +def test_read_harbor_reward_txt_rejects_non_finite_values( + tmp_path: Path, + invalid_reward: str, +) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.txt").write_text(invalid_reward, encoding="utf-8") + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info["reward_file"] == "reward.txt" + assert info["reward_parse_error"] == invalid_reward + + +def test_read_harbor_reward_reports_invalid_json(tmp_path: Path) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + (verifier / "reward.json").write_text('{"reward":', "utf-8") + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info["reward_parse_error"].startswith("invalid reward.json:") + + +@pytest.mark.parametrize("reward_name", ["reward.json", "reward.txt"]) +def test_read_harbor_reward_rejects_symlinks(tmp_path: Path, reward_name: str) -> None: + verifier = tmp_path / "verifier" + verifier.mkdir() + target = tmp_path / "outside-reward" + target.write_text('{"reward": 1.0}' if reward_name.endswith("json") else "1.0", "utf-8") + (verifier / reward_name).symlink_to(target) + + reward, info = _read_harbor_reward(verifier) + + assert reward is None + assert info["reward_file"] == reward_name + assert info["reward_parse_error"] == f"{reward_name} is not a regular file" + + +async def test_grade_stages_hidden_tests_clears_forged_reward_and_reads_fresh_result( + single_task: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + tests_dir = single_task / "tests" + (tests_dir / "hidden.txt").write_text("secret verifier fixture", encoding="utf-8") + staged_tests = tmp_path / "staged-tests" + staged_tests.mkdir() + assert list(staged_tests.iterdir()) == [] + + logs = tmp_path / "logs" + verifier_logs = logs / "verifier" + verifier_logs.mkdir(parents=True) + (verifier_logs / "reward.txt").write_text("1.0\n", encoding="utf-8") + (verifier_logs / "forged-artifact.txt").write_text("agent supplied", encoding="utf-8") + secret = "verifier-secret-never-in-argv" + events: list[str] = [] + + async def fake_docker( + *args: str, + check: bool = True, + env: dict[str, str] | None = None, + unset_env: tuple[str, ...] = (), + ) -> tuple[str, str]: + events.append("docker") + assert check is False + assert env is None + assert unset_env == () + assert args[:4] == ("exec", "--workdir", "/app", "--user") + assert args[4] == "1002" + assert "--env-file" in args + env_file = staged_tests.parent / "verifier.env" + assert args[args.index("--env-file") + 1] == str(env_file) + assert env_file.read_text("utf-8") == (f"SHARED=environment\nVERIFIER_ONLY={secret}\n") + assert env_file.stat().st_mode & 0o777 == 0o600 + assert secret not in " ".join(args) + assert args[-2:] == ("container", "/tests/test.sh") + assert (staged_tests / "hidden.txt").read_text("utf-8") == "secret verifier fixture" + assert not (verifier_logs / "reward.txt").exists() + assert not (verifier_logs / "forged-artifact.txt").exists() + (verifier_logs / "reward.json").write_text(json.dumps({"checks": 0.25}), encoding="utf-8") + return "verifier out", "" + + async def fake_release_log_permissions(container: str) -> None: + assert container == "container" + events.append("release") + + def fake_prepare_verifier_staging( + source_tests: Path, + destination_tests: Path, + destination_logs: Path, + ) -> None: + assert events == ["release"] + events.append("stage") + _prepare_verifier_staging(source_tests, destination_tests, destination_logs) + + monkeypatch.setattr("hud.eval.runtime._docker", fake_docker) + monkeypatch.setattr( + "integrations.harbor_runtime._release_log_permissions", + fake_release_log_permissions, + ) + monkeypatch.setattr( + "integrations.harbor_runtime._prepare_verifier_staging", + fake_prepare_verifier_staging, + ) + runtime = HarborRuntime(single_task.parent) + + result = await runtime._grade( + "container", + "/app", + tests_dir, + staged_tests, + logs, + "done", + verifier_timeout=120.0, + verifier_user=1002, + verifier_env={"SHARED": "environment", "VERIFIER_ONLY": secret}, + ) + + assert result["score"] == 0.25 + assert result["info"]["stdout"] == "verifier out" + assert result["info"]["reward_file"] == "reward.json" + assert events == ["release", "stage", "docker", "release"] + assert (staged_tests.parent / "agent_answer.txt").read_text("utf-8") == "done" + + +async def test_grade_times_out_when_verifier_hangs( + single_task: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + cancelled = asyncio.Event() + + async def fake_docker( + *args: str, + check: bool = True, + env: dict[str, str] | None = None, + unset_env: tuple[str, ...] = (), + ) -> tuple[str, str]: + try: + await asyncio.sleep(3600) + raise AssertionError("unreachable") + except asyncio.CancelledError: + cancelled.set() + raise + + async def fake_release_log_permissions(container: str) -> None: + assert container == "container" + + monkeypatch.setattr("hud.eval.runtime._docker", fake_docker) + monkeypatch.setattr( + "integrations.harbor_runtime._release_log_permissions", + fake_release_log_permissions, + ) + runtime = HarborRuntime(single_task.parent) + staged_tests = tmp_path / "staged-tests" + staged_tests.mkdir() + + result = await runtime._grade( + "container", + "/app", + single_task / "tests", + staged_tests, + tmp_path / "logs", + None, + verifier_timeout=0.05, + verifier_user=None, + verifier_env={}, + ) + + assert result["isError"] is True + assert "timed out" in result["content"] + assert cancelled.is_set() + + # ─── export: HUD tasks -> Harbor task folders ─────────────────────────── _ENV_PY = """\ @@ -140,12 +836,14 @@ async def test_task_toml_is_harbor_native(tmp_path: Path) -> None: _write_env(tmp_path) created = await export(str(tmp_path / "env.py"), tmp_path / "out") toml = (created[0] / "task.toml").read_text(encoding="utf-8") - assert 'version = "1.0"' in toml - assert "name = " in toml - assert "[verifier]" in toml and "[agent]" in toml - assert "timeout_sec" in toml - # HUD task/args preserved as metadata for the record. - assert "hud_task" in toml and "hud_args" in toml + assert tomllib.loads(toml) == { + "schema_version": "1.4", + "task": {"name": "hud/solve-99dd84a6"}, + # HUD task/args are metadata, not invalid root-level config fields. + "metadata": {"hud_task": "solve", "hud_args": '{"n": 2}'}, + "agent": {"timeout_sec": 600.0}, + "verifier": {"timeout_sec": 600.0}, + } async def test_scripts_drive_hud_task_lifecycle(tmp_path: Path) -> None: @@ -155,21 +853,38 @@ async def test_scripts_drive_hud_task_lifecycle(tmp_path: Path) -> None: test_sh = (created[0] / "tests" / "test.sh").read_text(encoding="utf-8") # Boot serves the channel, parks the run via setup, then hands off. - assert "hud serve env:env" in boot - assert "hud task start 'solve'" in boot + assert "HUD_ENV_TARGET=env.py:env" in boot + assert 'hud serve "$HUD_ENV_TARGET"' in boot + assert '-- "$HUD_TASK"' in boot + assert boot.index('hud task start --args "$HUD_ARGS"') < boot.index(': > "$HUD_READY_FILE"') assert 'exec "$@"' in boot # Verifier grades the parked run and writes the Harbor reward. - assert "hud task grade 'solve'" in test_sh + assert 'hud task grade --args "$HUD_ARGS"' in test_sh + assert '-- "$HUD_TASK"' in test_sh assert "--answer-file" in test_sh assert "/logs/verifier/reward.txt" in test_sh + assert "echo 0" not in test_sh + assert 'exit "$status"' in test_sh + + await asyncio.to_thread( + subprocess.run, + ["sh", "-n", str(created[0] / "environment" / "hud_entrypoint.sh")], + check=True, + ) + await asyncio.to_thread( + subprocess.run, + ["sh", "-n", str(created[0] / "tests" / "test.sh")], + check=True, + ) -async def test_dockerfile_neutralizes_cmd_and_bakes_boot(tmp_path: Path) -> None: +async def test_dockerfile_preserves_original_startup_and_appends_boot(tmp_path: Path) -> None: _write_env(tmp_path) created = await export(str(tmp_path / "env.py"), tmp_path / "out") dockerfile = (created[0] / "environment" / "Dockerfile").read_text(encoding="utf-8") - assert "# [hud original]" in dockerfile # original CMD neutralized + assert 'CMD ["hud", "serve", "env:env"]' in dockerfile assert 'ENTRYPOINT ["/hud_entrypoint.sh"]' in dockerfile + assert 'CMD ["test", "-f", "/tmp/.hud-harbor-ready"]' in dockerfile # The env build context is copied so the image can be rebuilt under Harbor. assert (created[0] / "environment" / "env.py").exists() @@ -179,3 +894,339 @@ async def test_custom_answer_file(tmp_path: Path) -> None: created = await export(str(tmp_path / "env.py"), tmp_path / "out", answer_file="/app/out.txt") assert "/app/out.txt" in (created[0] / "instruction.md").read_text(encoding="utf-8") assert "/app/out.txt" in (created[0] / "tests" / "test.sh").read_text(encoding="utf-8") + + +async def test_export_preserves_nondefault_module_and_symbol(tmp_path: Path) -> None: + source = tmp_path / "custom_runtime.py" + source.write_text( + textwrap.dedent(_ENV_PY) + .replace("env = Environment", "project_runtime = Environment") + .replace("@env.template", "@project_runtime.template"), + encoding="utf-8", + ) + (tmp_path / "Dockerfile").write_text( + 'FROM python:3.11-slim\nCOPY custom_runtime.py ./\nCMD ["true"]\n', + encoding="utf-8", + ) + + created = await export(str(source), tmp_path / "out") + + boot = (created[0] / "environment" / "hud_entrypoint.sh").read_text(encoding="utf-8") + assert "HUD_ENV_TARGET=custom_runtime.py:project_runtime" in boot + + +async def test_json_taskset_exclusion_preserves_other_json_build_inputs(tmp_path: Path) -> None: + _write_env(tmp_path) + taskset = tmp_path / "tasks.json" + taskset.write_text( + json.dumps([{"env": "demo", "id": "solve", "args": {"n": 2}}]), + encoding="utf-8", + ) + preserved = { + "package.json": '{"scripts":{"start":"node index.js"}}\n', + "tsconfig.json": '{"compilerOptions":{"strict":true}}\n', + "fixtures.jsonl": '{"input":1}\n', + } + for name, content in preserved.items(): + (tmp_path / name).write_text(content, encoding="utf-8") + + created = await export(str(taskset), tmp_path / "out") + environment = created[0] / "environment" + + assert not (environment / taskset.name).exists() + for name, content in preserved.items(): + assert (environment / name).read_text(encoding="utf-8") == content + + +async def test_export_normalizes_traversal_slug_and_rejects_collisions(tmp_path: Path) -> None: + _write_env(tmp_path) + taskset = tmp_path / "traversal.json" + taskset.write_text( + json.dumps([{"env": "demo", "id": "solve", "args": {"n": 2}, "slug": "../../escape"}]), + encoding="utf-8", + ) + out = tmp_path / "out" + + created = await export(str(taskset), out) + + assert created == [out.resolve() / "escape"] + assert not (tmp_path / "escape").exists() + + taskset.write_text( + json.dumps( + [ + {"env": "demo", "id": "solve", "args": {"n": 1}, "slug": "a/b"}, + {"env": "demo", "id": "solve", "args": {"n": 2}, "slug": "ab"}, + ] + ), + encoding="utf-8", + ) + collision_out = tmp_path / "collision-out" + with pytest.raises(ValueError, match="normalize to Harbor folder 'ab'"): + await export(str(taskset), collision_out) + assert not (collision_out / "ab").exists() + + +async def test_generated_scripts_preserve_hostile_values_as_single_argv(tmp_path: Path) -> None: + sentinel = tmp_path / "injected" + task_id = "--help'$(touch injected)\nnext" + arg_value = "quote' ; touch injected; $(touch injected)\nline" + source = tmp_path / "custom_runtime.py" + source.write_text( + textwrap.dedent( + f"""\ + from hud import Environment + from hud.eval import Task + + project_runtime = Environment("demo") + + @project_runtime.template(id={task_id!r}) + async def solve(payload: str): + yield "hostile" + yield 1.0 + + tasks = [Task( + env="demo", + id={task_id!r}, + args={{"payload": {arg_value!r}}}, + slug="hostile-safe", + )] + """ + ), + encoding="utf-8", + ) + (tmp_path / "Dockerfile").write_text("FROM python:3.11-slim\n", encoding="utf-8") + answer_file = str(tmp_path / "answer'$(touch injected)\nfile") + created = await export(str(source), tmp_path / "out", answer_file=answer_file) + + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_hud = fake_bin / "hud" + fake_hud.write_text( + textwrap.dedent("""\ + #!/bin/sh + if [ "$1" = "serve" ]; then + exit 0 + fi + printf '%s\\037' "$@" > "$HUD_CAPTURE" + printf '0\\n' + """), + encoding="utf-8", + ) + fake_hud.chmod(0o755) + fake_python = fake_bin / "python3" + fake_python.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + fake_python.chmod(0o755) + command_env = {"PATH": f"{fake_bin}:/usr/bin:/bin"} + + boot_capture = tmp_path / "boot-argv" + boot = (created[0] / "environment" / "hud_entrypoint.sh").read_text(encoding="utf-8") + boot_runner = tmp_path / "boot-runner.sh" + boot_runner.write_text( + boot.replace("/tmp/.hud-harbor-ready", str(tmp_path / "ready")), + encoding="utf-8", + ) + await asyncio.to_thread( + subprocess.run, + ["sh", str(boot_runner), "true"], + check=True, + cwd=tmp_path, + env={**command_env, "HUD_CAPTURE": str(boot_capture)}, + ) + boot_argv = boot_capture.read_bytes().split(b"\x1f")[:-1] + assert [part.decode() for part in boot_argv] == [ + "task", + "start", + "--args", + json.dumps({"payload": arg_value}), + "--url", + "tcp://127.0.0.1:8765", + "--", + task_id, + ] + + grade_capture = tmp_path / "grade-argv" + verifier_logs = tmp_path / "verifier-logs" + verifier = (created[0] / "tests" / "test.sh").read_text(encoding="utf-8") + verifier_runner = tmp_path / "verifier-runner.sh" + verifier_runner.write_text( + verifier.replace("/logs/verifier", str(verifier_logs)), + encoding="utf-8", + ) + await asyncio.to_thread( + subprocess.run, + ["sh", str(verifier_runner)], + check=True, + cwd=tmp_path, + env={**command_env, "HUD_CAPTURE": str(grade_capture)}, + ) + grade_argv = grade_capture.read_bytes().split(b"\x1f")[:-1] + assert [part.decode() for part in grade_argv] == [ + "task", + "grade", + "--args", + json.dumps({"payload": arg_value}), + "--answer-file", + answer_file, + "--url", + "tcp://127.0.0.1:8765", + "--", + task_id, + ] + assert not sentinel.exists() + + +def test_task_cli_option_terminator_keeps_help_like_id_positional( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: list[str] = [] + + def stop_after_parse(task: str, *_args: object) -> None: + captured.append(task) + raise RuntimeError("parsed") + + monkeypatch.setattr("hud.cli.task._resolve", stop_after_parse) + + result = CliRunner().invoke( + app, + ["task", "start", "--url", "tcp://127.0.0.1:1", "--", "--help"], + ) + + assert captured == ["--help"] + assert isinstance(result.exception, RuntimeError) + + +async def test_entrypoint_fails_closed_when_server_dies(tmp_path: Path) -> None: + src = _write_env(tmp_path) + created = await export(str(src), tmp_path / "out") + fake_bin = tmp_path / "dead-server-bin" + fake_bin.mkdir() + start_seen = tmp_path / "start-seen" + fake_hud = fake_bin / "hud" + fake_hud.write_text( + textwrap.dedent( + f"""\ + #!/bin/sh + if [ "$1" = "serve" ]; then + exit 7 + fi + : > {str(start_seen)!r} + exit 0 + """ + ), + encoding="utf-8", + ) + fake_hud.chmod(0o755) + final_seen = tmp_path / "final-seen" + boot = (created[0] / "environment" / "hud_entrypoint.sh").read_text(encoding="utf-8") + boot_runner = tmp_path / "dead-server-entrypoint.sh" + boot_runner.write_text( + boot.replace("/tmp/.hud-harbor-ready", str(tmp_path / "ready")), + encoding="utf-8", + ) + + result = await asyncio.to_thread( + subprocess.run, + ["sh", str(boot_runner), "sh", "-c", f": > {str(final_seen)!r}"], + check=False, + timeout=5, + env={"PATH": f"{fake_bin}:{sys.executable.rpartition('/')[0]}:/usr/bin:/bin"}, + capture_output=True, + text=True, + ) + + assert result.returncode != 0 + assert not start_seen.exists() + assert not final_seen.exists() + + +async def test_entrypoint_cleans_up_and_fails_closed_when_setup_fails(tmp_path: Path) -> None: + src = _write_env(tmp_path) + created = await export(str(src), tmp_path / "out") + fake_bin = tmp_path / "setup-failure-bin" + fake_bin.mkdir() + server_pid_file = tmp_path / "server-pid" + fake_hud = fake_bin / "hud" + fake_hud.write_text( + textwrap.dedent("""\ + #!/bin/sh + if [ "$1" = "serve" ]; then + printf '%s' "$$" > "$SERVER_PID_FILE" + exec sleep 30 + fi + echo 'setup failed' >&2 + exit 9 + """), + encoding="utf-8", + ) + fake_hud.chmod(0o755) + fake_python = fake_bin / "python3" + fake_python.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + fake_python.chmod(0o755) + final_seen = tmp_path / "final-seen" + ready = tmp_path / "ready" + boot = (created[0] / "environment" / "hud_entrypoint.sh").read_text(encoding="utf-8") + boot_runner = tmp_path / "setup-failure-entrypoint.sh" + boot_runner.write_text(boot.replace("/tmp/.hud-harbor-ready", str(ready)), encoding="utf-8") + + result = await asyncio.to_thread( + subprocess.run, + ["sh", str(boot_runner), "sh", "-c", f": > {str(final_seen)!r}"], + check=False, + timeout=5, + env={ + "PATH": f"{fake_bin}:/usr/bin:/bin", + "SERVER_PID_FILE": str(server_pid_file), + }, + capture_output=True, + text=True, + ) + + assert result.returncode == 9 + assert "setup failed" in result.stderr + assert not ready.exists() + assert not final_seen.exists() + server_pid = server_pid_file.read_text(encoding="utf-8") + process_check = await asyncio.to_thread( + subprocess.run, + ["kill", "-0", server_pid], + check=False, + ) + assert process_check.returncode != 0 + + +async def test_verifier_failure_leaves_no_reward_file(tmp_path: Path) -> None: + src = _write_env(tmp_path) + created = await export( + str(src), + tmp_path / "out", + answer_file=str(tmp_path / "answer.txt"), + ) + fake_bin = tmp_path / "grade-failure-bin" + fake_bin.mkdir() + fake_hud = fake_bin / "hud" + fake_hud.write_text( + "#!/bin/sh\nprintf 'partial reward\\n'\necho 'grade infrastructure failed' >&2\nexit 7\n", + encoding="utf-8", + ) + fake_hud.chmod(0o755) + verifier_logs = tmp_path / "verifier-logs" + verifier = (created[0] / "tests" / "test.sh").read_text(encoding="utf-8") + verifier_runner = tmp_path / "grade-failure-verifier.sh" + verifier_runner.write_text( + verifier.replace("/logs/verifier", str(verifier_logs)), + encoding="utf-8", + ) + + result = await asyncio.to_thread( + subprocess.run, + ["sh", str(verifier_runner)], + check=False, + env={"PATH": f"{fake_bin}:/usr/bin:/bin"}, + capture_output=True, + text=True, + ) + + assert result.returncode == 7 + assert "grade infrastructure failed" in result.stderr + assert not (verifier_logs / "reward.txt").exists() diff --git a/pyproject.toml b/pyproject.toml index 9419597c7..426fee6bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,7 +57,6 @@ build-backend = "hatchling.build" exclude = [ "docs/", "cookbooks/", - "integrations/", "hud-python/", "**/checkpoints/", "**/*.safetensors", @@ -85,6 +84,7 @@ allow-direct-references = true [tool.hatch.build.targets.sdist] include = [ "hud/**", + "integrations/**", "/README.md", "/LICENSE", "/pyproject.toml" @@ -102,7 +102,7 @@ exclude = [ ] [tool.hatch.build.targets.wheel] -packages = ["hud"] +packages = ["hud", "integrations"] # Ensure py.typed is included in the package [tool.hatch.build.targets.wheel.force-include]