feat(agents): local integration_test authoring agent

hud/agents/__init__.py

@@ -109,10 +109,16 @@ def create_agent(model: str, **kwargs: Any) -> GatewayAgent:
             )
             raise ValueError(f"Model {requested_model!r} not found in {source}.{hint}")
 
+    if agent_type is AgentType.INTEGRATION_TEST:
+        raise ValueError(
+            "integration_test is the local authoring agent, not a gateway model: "
+            "it makes no LLM calls. Run it with `hud eval <taskset> integration_test`."
+        )
+
     kwargs.setdefault("model", model_id)
     # cls/config_cls are matched unions; the pairing is correct by construction.
     config = agent_type.config_cls(**kwargs)
-    return agent_type.cls(cast("Any", config))
+    return cast("GatewayAgent", agent_type.cls(cast("Any", config)))
 
 
 _LAZY_EXPORTS = {

hud/agents/integration_test.py

@@ -0,0 +1,196 @@
+"""The local authoring agent: ``integration_test``.
+
+The 01-coding-template documents ``hud eval <env> integration_test`` as the
+shipping check for a task: pre-stage the golden solution (``Task.validation``),
+let the environment's scenario graders run, and require Reward 1.0. This is
+the *local* implementation of that agent — no LLM, no platform. It replays
+every validation tool call through the task's own MCP capabilities (the bash
+capability for the coding template), then ends the trace with an empty answer
+so the environment grades the staged workspace.
+
+The Reward-1.0 gate lives in the CLI (``hud.cli.eval``): grading happens after
+the agent finishes, so the agent itself cannot see the reward.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Any, cast
+
+from hud.agents.base import Agent
+from hud.agents.types import IntegrationTestConfig, ToolStep
+from hud.capabilities import MCPClient, SSHClient
+from hud.types import MCPToolCall, MCPToolResult
+from hud.utils.time import now_iso
+
+if TYPE_CHECKING:
+    from hud.eval.run import Run
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_VALIDATION_TIMEOUT_SECONDS = 120.0
+
+
+class IntegrationTestAgent(Agent):
+    """Pre-stage ``Task.validation``, then yield to the scenario graders.
+
+    Stateless per run; one instance can drive concurrent rollouts.
+    """
+
+    def __init__(self, config: IntegrationTestConfig) -> None:
+        self.config = config
+
+    async def __call__(self, run: Run) -> None:
+        validation = list(getattr(run, "validation", None) or [])
+        if not validation:
+            logger.warning("integration_test: task has no Task.validation — nothing staged")
+            return
+
+        connections: dict[str, MCPClient | SSHClient] = {}
+        manifest = run.client.manifest
+        if manifest is not None:
+            for cap in manifest.bindings:
+                if cap.protocol not in (MCPClient.protocol, SSHClient.protocol):
+                    continue
+                opened = await run.client.open(cap.name)
+                # open() resolves through the capability registry, so the
+                # client type matches the protocol we filtered on above.
+                connections[cap.name] = cast("MCPClient | SSHClient", opened)
+
+        if not connections:
+            logger.warning(
+                "integration_test: no MCP capabilities to stage the golden solution through"
+            )
+
+        timeout = self.config.timeout_seconds or DEFAULT_VALIDATION_TIMEOUT_SECONDS
+        deadline = asyncio.timeout(timeout)
+        try:
+            async with deadline:
+                for entry in validation:
+                    call = self._coerce_call(entry)
+                    if call is None:
+                        continue
+                    result = await self._dispatch(connections, call)
+                    run.record(ToolStep(call=call, result=result, started_at=now_iso()))
+        except TimeoutError:
+            run.trace.status = "error"
+            run.trace.stop_reason = "timeout"
+            logger.warning("integration_test: validation staging timed out after %gs", timeout)
+
+    @staticmethod
+    def _coerce_call(entry: Any) -> MCPToolCall | None:
+        if isinstance(entry, MCPToolCall):
+            return entry
+        if isinstance(entry, dict):
+            try:
+                return MCPToolCall.model_validate(entry)
+            except Exception as exc:  # surface as a warning, not a crash
+                logger.warning("integration_test: skipping invalid validation step: %s", exc)
+                return None
+        logger.warning("integration_test: skipping unsupported validation step %r", entry)
+        return None
+
+    async def _dispatch(
+        self,
+        connections: dict[str, MCPClient | SSHClient],
+        call: MCPToolCall,
+    ) -> MCPToolResult:
+        """Run one validation tool call against whichever capability serves it.
+
+        Most tasks declare a single MCP capability (``bash``); try each
+        connected client in turn and stop at the first non-"unknown tool"
+        result so a task with several capabilities still routes correctly.
+        SSH-published workspaces (protocol ``ssh/2``) run the same golden
+        ``bash`` steps via ``bash -lc`` over the SSH connection.
+        """
+        from mcp.types import TextContent
+
+        raw_args = call.arguments or {}
+        if not isinstance(raw_args, dict):
+            from mcp.types import TextContent
+
+            return MCPToolResult(
+                content=[
+                    TextContent(
+                        type="text",
+                        text="the validation step's arguments arrived as a string "
+                        "and were not executed",
+                    )
+                ],
+                isError=True,
+            )
+        args: dict[str, Any] = raw_args
+        last: MCPToolResult | None = None
+        for name, client in connections.items():
+            try:
+                if isinstance(client, SSHClient):
+                    result = await self._run_over_ssh(client, call)
+                else:
+                    result = await client.call_tool(call.name, args)
+            except Exception as exc:
+                logger.warning(
+                    "integration_test: capability %r failed for %r: %s",
+                    name,
+                    call.name,
+                    exc,
+                )
+                last = MCPToolResult(
+                    content=[TextContent(type="text", text=f"tool error: {exc}")],
+                    isError=True,
+                )
+                continue
+            text = getattr(result, "content", None)
+            unknown = any(
+                getattr(item, "type", None) == "text"
+                and str(getattr(item, "text", "")).startswith("unknown tool:")
+                for item in (text or [])
+            )
+            if not unknown:
+                return result
+            last = result
+
+        if last is not None:
+            return last
+        return MCPToolResult(
+            content=[
+                TextContent(
+                    type="text",
+                    text=(
+                        f"unknown tool: {call.name!r} — no connected MCP "
+                        "or SSH capability serves it"
+                    ),
+                )
+            ],
+            isError=True,
+        )
+
+    @staticmethod
+    async def _run_over_ssh(client: SSHClient, call: MCPToolCall) -> MCPToolResult:
+        from mcp.types import TextContent
+
+        raw_args = call.arguments or {}
+        command = raw_args.get("command") if isinstance(raw_args, dict) else None
+        if not isinstance(command, str):
+            return MCPToolResult(
+                content=[
+                    TextContent(
+                        type="text",
+                        text=f"validation step {call.name!r} has no bash command to run",
+                    )
+                ],
+                isError=True,
+            )
+        # Single remote command string: asyncssh shlex-splits it, preserving
+        # the command's own quoting, and ships `command` to `bash -lc` as one
+        # argument (the codebase idiom; avoids multi-arg space-join hazards).
+        completed = await client.run(f"bash -lc {command}")
+        output_parts = [p for p in (completed.stdout, completed.stderr) if p]
+        output = "".join(
+            p.decode("utf-8", errors="replace") if isinstance(p, bytes) else str(p)
+            for p in output_parts
+        )
+        return MCPToolResult(
+            content=[TextContent(type="text", text=output.strip() or "(no output)")],
+            isError=completed.returncode != 0,
+        )

hud/agents/tests/test_base.py

@@ -83,6 +83,12 @@ def gateway_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr("hud.agents.settings.api_key", "test-key")
 
 
+def test_create_agent_rejects_integration_test() -> None:
+    # The authoring agent is not a gateway shortcut; it makes no LLM calls.
+    with pytest.raises(ValueError, match="local authoring agent"):
+        create_agent("integration_test")
+
+
 def test_create_agent_unknown_model_raises(monkeypatch: pytest.MonkeyPatch) -> None:
     # No gateway models available -> a bare unknown model can't be resolved.
     monkeypatch.setattr("hud.agents.list_gateway_models", list)

hud/agents/tests/test_integration_test.py

@@ -0,0 +1,159 @@
+"""The local authoring agent (integration_test)."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from hud.agents.integration_test import IntegrationTestAgent
+from hud.agents.types import IntegrationTestConfig, ToolStep
+from hud.types import MCPToolCall
+
+
+def _fake_run(*, validation: list[Any] | None, tools: dict[str, str]) -> Any:
+    """Minimal Run stand-in: manifest + open() returning per-capability MCP clients."""
+
+    class FakeClient:
+        async def call_tool(self, name: str, args: dict[str, Any]):
+            served = tools.get(name)
+            if served is None:
+                from mcp.types import TextContent
+
+                from hud.types import MCPToolResult
+
+                return MCPToolResult(
+                    content=[TextContent(type="text", text=f"unknown tool: {name!r}")],
+                    isError=True,
+                )
+            from mcp.types import TextContent
+
+            from hud.types import MCPToolResult
+
+            return MCPToolResult(content=[TextContent(type="text", text=served)])
+
+    run = SimpleNamespace()
+    run.validation = validation
+    run.recorded = []  # type: list[Any]
+    run.trace = SimpleNamespace(status="running", stop_reason=None)
+
+    def record(step):
+        run.recorded.append(step)
+
+    run.record = record
+    run.client = SimpleNamespace()
+    run.client.manifest = SimpleNamespace(
+        bindings=[
+            SimpleNamespace(name="bash", protocol="mcp"),
+        ]
+    )
+    run.client.open = AsyncMock(return_value=FakeClient())
+    return run
+
+
+@pytest.mark.asyncio
+async def test_dispatches_validation_calls_and_records_tool_steps() -> None:
+    agent = IntegrationTestAgent(IntegrationTestConfig())
+    run = _fake_run(
+        validation=[
+            MCPToolCall(name="bash", arguments={"command": "echo 'golden' > answer.txt"}),
+            {"name": "bash", "arguments": {"command": "chmod +x answer.txt"}},
+        ],
+        tools={"bash": "ok"},
+    )
+
+    await agent(run)
+
+    assert len(run.recorded) == 2
+    assert all(isinstance(step, ToolStep) for step in run.recorded)
+    assert run.recorded[0].call.name == "bash"
+
+
+@pytest.mark.asyncio
+async def test_empty_validation_is_a_noop() -> None:
+    agent = IntegrationTestAgent(IntegrationTestConfig())
+    run = _fake_run(validation=[], tools={})
+
+    await agent(run)
+
+    assert run.recorded == []
+
+
+@pytest.mark.asyncio
+async def test_invalid_entries_are_skipped_without_crashing() -> None:
+    agent = IntegrationTestAgent(IntegrationTestConfig())
+    run = _fake_run(
+        validation=[{"not": "a tool call"}, 42, MCPToolCall(name="bash", arguments={})],
+        tools={"bash": "ok"},
+    )
+
+    await agent(run)
+
+    assert len(run.recorded) == 1
+
+
+def _fake_run_ssh(*, validation: list[Any]) -> Any:
+    """Run stand-in whose workspace is published over SSH (ssh/2)."""
+
+    from hud.capabilities import SSHClient
+
+    class FakeSSHClient(SSHClient):
+        def __init__(self) -> None:
+            pass
+
+        async def run(self, *args: object, **kwargs: Any) -> Any:
+            from types import SimpleNamespace
+
+            assert args == ("bash -lc echo 'golden' > answer.txt",)
+            return SimpleNamespace(stdout=b"staged\n", stderr=b"", returncode=0)
+
+    run = _fake_run(validation=validation, tools={})
+    run.client.manifest = SimpleNamespace(
+        bindings=[SimpleNamespace(name="workspace", protocol="ssh/2")]
+    )
+    run.client.open = AsyncMock(return_value=FakeSSHClient())
+    return run
+
+
+@pytest.mark.asyncio
+async def test_dispatches_validation_over_ssh_workspace() -> None:
+    from hud.agents.types import ToolStep
+
+    agent = IntegrationTestAgent(IntegrationTestConfig())
+    run = _fake_run_ssh(
+        validation=[MCPToolCall(name="bash", arguments={"command": "echo 'golden' > answer.txt"})]
+    )
+
+    await agent(run)
+
+    assert len(run.recorded) == 1
+    step = run.recorded[0]
+    assert isinstance(step, ToolStep)
+    assert step.result.isError is False
+    assert step.result.content[0].text == "staged"
+
+
+@pytest.mark.asyncio
+async def test_ssh_failure_surfaces_as_error_result() -> None:
+    from types import SimpleNamespace
+
+    from hud.capabilities import SSHClient
+
+    class FailingSSHClient(SSHClient):
+        def __init__(self) -> None:
+            pass
+
+        async def run(self, *args: object, **kwargs: Any) -> Any:
+            return SimpleNamespace(stdout=b"", stderr=b"no such file", returncode=127)
+
+    run = _fake_run(validation=[MCPToolCall(name="bash", arguments={"command": "nope"})], tools={})
+    run.client.manifest = SimpleNamespace(
+        bindings=[SimpleNamespace(name="workspace", protocol="ssh/2")]
+    )
+    run.client.open = AsyncMock(return_value=FailingSSHClient())
+
+    await IntegrationTestAgent(IntegrationTestConfig())(run)
+
+    assert run.recorded[0].result.isError is True