# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 import asyncio import json import re import tomllib from pathlib import Path from types import SimpleNamespace import pytest ROOT = Path(__file__).resolve().parents[2] ROOT_README = ROOT / "README.md" CALCULATOR_ROOT = ROOT / "examples" / "harbor" / "calculator" CALCULATOR_README = CALCULATOR_ROOT / "README.md" CALCULATOR_DOCKERFILE = CALCULATOR_ROOT / "task" / "environment" / "Dockerfile" CALCULATOR_SOLUTION = CALCULATOR_ROOT / "task" / "solution" / "solve.sh" CALCULATOR_FABRIC_ROOT = CALCULATOR_ROOT / "task" / "environment" / "fabric" SWEBENCH_ROOT = ROOT / "examples" / "harbor" / "swebench" SWEBENCH_README = SWEBENCH_ROOT / "README.md" SWEBENCH_MCP_CONFIG = SWEBENCH_ROOT / "mcp" / "repo-inspector.mcp.json" SWEBENCH_CLINE_DOCKERFILE = SWEBENCH_ROOT / "cline" / "Dockerfile" SWEBENCH_OPENCODE_DOCKERFILE = SWEBENCH_ROOT / "opencode" / "Dockerfile" INTEGRATION_README = ROOT / "examples" / "harbor" / "README.md" SDK_INTEGRATION_README = ( ROOT / "sdk" / "python" / "nemo-fabric-runtime" / "src" / "nemo_fabric" / "integrations" / "harbor" / "README.md" ) HARBOR_PACKAGE_INIT = SDK_INTEGRATION_README.parent / "__init__.py" pytestmark = pytest.mark.usefixtures("requires_harbor") def documented_root_extras(text: str) -> set[str]: extras = set(re.findall(r"--extra\s+([a-z0-9-]+)", text)) for selector in re.findall(r"nemo-fabric\[([^]]+)\]", text): extras.update(extra.strip() for extra in selector.split(",")) return extras def load_codex_adapter(): from nemo_fabric_adapters.codex import adapter return adapter def test_harbor_builder_constructs_complete_config_from_harbor_inputs(): from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config from nemo_fabric.integrations.harbor.models import HarborMcpServer config = build_harbor_config( adapter_id="demo.fabric.smoke", workspace="/testbed", model_name="openai/gpt-5.4", mcp_servers=( HarborMcpServer( name="remote", transport="streamable-http", url="https://mcp.example.test", ), HarborMcpServer( name="local", transport="stdio", command="mcp-server", args=("--stdio",), ), ), ) assert config.models["default"].to_mapping() == { "provider": "openai", "model": "openai/gpt-5.4", } assert config.mcp is not None assert set(config.mcp.servers) == {"remote", "local"} assert config.mcp.servers["local"].url == "mcp-server" assert config.mcp.servers["local"].args == ["--stdio"] assert "args" not in config.mcp.servers["local"].extra_fields assert config.skills is None assert ( json.loads(json.dumps(config.to_mapping()))["metadata"]["name"] == "harbor-smoke" ) def test_harbor_builder_leaves_optional_capabilities_unset(): from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config config = build_harbor_config( adapter_id="demo.fabric.smoke", workspace="/testbed", ) assert config.models == {} assert config.mcp is None assert config.skills is None def test_harbor_transport_models_validate_mcp_targets(): from nemo_fabric import RunRequest from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config from nemo_fabric.integrations.harbor.models import FabricRunPayload from nemo_fabric.integrations.harbor.models import HarborMcpServer from pydantic import ValidationError server = HarborMcpServer( name="github", transport="streamable-http", url="https://mcp.example.test", ) payload = FabricRunPayload.model_validate_json( FabricRunPayload( config=build_harbor_config( adapter_id="demo.fabric.smoke", workspace="/testbed" ), config_base_dir="/workspace", request=RunRequest(input="fix it"), ).model_dump_json() ) assert payload.request.input == "fix it" assert server.name == "github" payload_properties = FabricRunPayload.model_json_schema()["properties"] assert set(payload_properties) == { "config", "config_base_dir", "logs_dir", "skills_dir", "request", "environment_env_names", "adapter_descriptor_sha256", } assert payload_properties["logs_dir"]["default"] == "/logs/agent" with pytest.raises(ValidationError, match="require url"): HarborMcpServer(name="missing", transport="sse") with pytest.raises(ValidationError, match="require command"): HarborMcpServer(name="missing", transport="stdio") with pytest.raises(ValidationError, match="Extra inputs"): FabricRunPayload.model_validate( { "config": {}, "config_base_dir": "/workspace", "request": {"input": "fix it"}, "profile_paths": [], } ) with pytest.raises(ValidationError, match="Field required"): FabricRunPayload.model_validate( { "config_base_dir": "/workspace", "request": {"input": "fix it"}, } ) def test_each_harbor_job_delegates_to_an_independent_fabric_run( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ): from nemo_fabric.integrations.harbor import runner from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config from nemo_fabric.integrations.harbor.models import FabricRunPayload calls: list[dict[str, object]] = [] class FakeResult: def __init__(self, runtime_id: str) -> None: self.runtime_id = runtime_id def to_mapping(self) -> dict[str, str]: return {"runtime_id": self.runtime_id} class FakeFabric: async def run(self, config, *, base_dir, request): runtime_id = f"runtime-{len(calls) + 1}" calls.append( { "runtime_id": runtime_id, "request": request.to_mapping(), } ) return FakeResult(runtime_id) monkeypatch.setattr(runner, "Fabric", FakeFabric) monkeypatch.setattr( runner, "publish_telemetry_evidence", lambda result, path, **kwargs: {}, ) payloads = [ FabricRunPayload( config=build_harbor_config( adapter_id="demo.fabric.smoke", workspace="/testbed" ), config_base_dir="/workspace", request={ "input": f"job {job_id}", "context": {"job_id": job_id}, }, ) for job_id in ("job-1", "job-2") ] async def run_payloads(): return await asyncio.gather(*(runner.run(payload) for payload in payloads)) results = asyncio.run(run_payloads()) assert [result.runtime_id for result in results] == ["runtime-1", "runtime-2"] requests = [call["request"] for call in calls] assert [request["context"]["job_id"] for request in requests] == [ # type: ignore[index] "job-1", "job-2", ] def test_codex_adapter_maps_fabric_request_to_sdk(tmp_path): from nemo_fabric_adapter_contract.models import AgentConfig from nemo_fabric_adapter_contract.models import RuntimeContext adapter = load_codex_adapter() payload = { "base_dir": str(tmp_path), "config": { "harness": { "settings": { "sandbox": "workspace-write", "reasoning_effort": "high", } }, "models": { "default": { "provider": "openai", "model": "openai/gpt-5.4", } }, "runtime": {}, }, "runtime_context": { "runtime_id": "harbor-test", "invocation_id": "harbor-invocation", "request_id": "harbor-request", "environment": { "environment_id": "harbor-environment", "provider": "local", "control_location": "in_env_control", "ownership": "caller_owned", "workspace": str(tmp_path), }, "artifacts": {}, }, "request": {"input": "Fix the calculator."}, } config = AgentConfig.from_mapping(payload["config"]) context = RuntimeContext.from_mapping(payload["runtime_context"]) assert adapter.selected_model(config) == "gpt-5.4" assert adapter.sandbox(config) == adapter.Sandbox.workspace_write assert adapter._reasoning_effort(config) == adapter.ReasoningEffort.high assert adapter.thread_config(config, context, relay=None) == { "mcp_optional_startup_grace_ms": 0 } assert adapter.resolve_cwd(context, payload["base_dir"]) == tmp_path def test_claude_calculator_run_uses_current_adapter_contract(): from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config config = build_harbor_config( adapter_id="nvidia.fabric.claude", workspace="/app", model_name="anthropic/claude-sonnet-4-5", harness_settings={"permission_mode": "bypassPermissions"}, max_turns=20, timeout_seconds=600, ) settings = config.harness.settings assert config.harness.adapter_id == "nvidia.fabric.claude" assert settings["permission_mode"] == "bypassPermissions" assert config.runtime.max_turns == 20 assert config.runtime.timeout_seconds == 600 assert config.models["default"].provider == "anthropic" dockerfile = CALCULATOR_DOCKERFILE.read_text(encoding="utf-8") assert "-e /opt/nemo-fabric/adapter-contract/python" in dockerfile assert "-e /opt/nemo-fabric/adapters/python/claude" in dockerfile assert ( "git clone https://github.com/NousResearch/hermes-agent.git" not in dockerfile ) assert "-e /opt/nemo-fabric/adapters/python/openclaw" in dockerfile assert "-e /opt/nemo-fabric/sdk/python/nemo-fabric-runtime" in dockerfile assert '-e "/opt/nemo-fabric/sdk/python/nemo-fabric[' in dockerfile assert "nemo-fabric[claude,openclaw]" in dockerfile assert "node:24.16.0-bookworm-slim" in dockerfile assert ( "npm install --global openclaw@2026.9.4 --allow-scripts=openclaw" in dockerfile ) assert "@openai/codex" not in dockerfile def test_harbor_calculator_uses_agent_inputs_without_config_files(): assert not list(CALCULATOR_FABRIC_ROOT.glob("*.py")) assert not list(CALCULATOR_FABRIC_ROOT.rglob("*.yaml")) def test_harbor_smoke_config_resolves_its_local_adapter(): from nemo_fabric import Fabric from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config config = build_harbor_config( adapter_id="demo.fabric.scripted", workspace="/app", discovery_paths=("adapters",), ) plan = Fabric().plan(config, base_dir=CALCULATOR_FABRIC_ROOT) assert plan.adapter.adapter_id == "demo.fabric.scripted" provenance = plan["adapter_descriptor"]["provenance"][0] assert provenance["source"] == "explicit_local" assert Path(provenance["root"]).parts[-2:] == ( "adapters", "scripted", ) def test_harbor_calculator_documents_explicit_cli_commands(): calculator = CALCULATOR_README.read_text(encoding="utf-8") dockerfile = CALCULATOR_DOCKERFILE.read_text(encoding="utf-8") landing = INTEGRATION_README.read_text(encoding="utf-8") swebench = SWEBENCH_README.read_text(encoding="utf-8") with (ROOT / "sdk/python/nemo-fabric/pyproject.toml").open("rb") as file: sdk_project = tomllib.load(file)["project"] package_version = sdk_project["version"] declared_extras = set(sdk_project["optional-dependencies"]) assert "run.sh" not in calculator assert calculator.count(" harbor run \\") == 5 assert calculator.count("uv run --extra harbor harbor run \\") == 5 assert '--ak "fabric_config_bundle=$TASK_DIR/environment/fabric"' in calculator assert "uv run --extra harbor --extra" not in calculator assert landing.count("uv run --extra harbor harbor run") == 0 assert swebench.count("uv run --extra harbor harbor run") == 9 assert "--agent-import-path" not in landing + calculator + swebench assert "fabric_config_path" not in calculator assert "fabric_config_path" not in landing assert "fabric_config_path" not in swebench assert "fabric_config_factory" not in calculator assert "fabric_config_factory" not in landing assert "fabric_config_factory" not in swebench assert ( 'fabric_harness_settings={"permission_mode":"bypassPermissions"}' in calculator ) assert ( 'fabric_harness_settings={"sandbox":"workspace-write","approval_mode":"deny_all"}' in calculator ) assert "fabric_workspace=/app" in calculator assert "--model nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" in calculator assert "--model anthropic/claude-sonnet-4-5" in calculator assert "fabric_adapter_id=nvidia.fabric.openclaw" in calculator assert "fabric_adapter_id=nvidia.fabric.pi" in calculator pi_command = calculator.split("## 4. Pi", 1)[1].split("## 5. Codex", 1)[0] assert "fabric_max_turns" not in pi_command assert "fabric_adapter_id=nvidia.fabric.codex" in calculator assert "fabric_model_api_key_env=OPENAI_API_KEY" in calculator codex_command = calculator.split("## 5. Codex", 1)[1].split( "## Inspect Results", 1 )[0] assert "fabric_max_turns" not in codex_command assert "fabric_discovery_paths=" in calculator assert "pi.fabric-adapter.json" in calculator assert "-e '/opt/nemo-fabric/adapters/python/codex[harness]'" in dockerfile assert "--workspace nemo-fabric-adapters-pi" in dockerfile assert "fabric_model_api_key_env=NVIDIA_API_KEY" in calculator assert 'CALCULATOR_DIR="$PWD/examples/harbor/calculator"' in calculator assert "calculator/README.md" in landing assert "swebench/README.md" in landing assert f"nemo-fabric[harbor]=={package_version}" in landing assert f"nemo-fabric[claude]=={package_version}" in landing assert f"nemo-fabric=={package_version}" in landing assert f"nemo-fabric-adapters-hermes=={package_version}" in landing assert "Hermes Agent 0.20 and later is no longer installable from PyPI" in landing assert ( documented_root_extras("\n".join((calculator, dockerfile, landing, swebench))) <= declared_extras ) assert "fabric_adapter_id" in landing assert 'export TMPDIR="$HOME/harbor-tmp"' in landing assert "raw.githubusercontent.com/NVIDIA/NeMo-Relay/main/install.sh" in swebench assert ( "FABRIC_PACKAGE=" f"'nemo-fabric[claude,hermes-agent,relay]=={package_version}'" in swebench ) assert "PIP_FIND_LINKS" not in swebench assert "PATH=/tmp/nemo-fabric-config/.relay/bin:$PATH" in swebench assert "--dataset swe-bench/swe-bench-verified" in swebench for value in ( '--path "$OPENCODE_SWEBENCH_TASK"', "fabric_adapter_id=nvidia.fabric.opencode", ): assert swebench.count(value) == 2 assert swebench.count("fabric_config_target=/opt/nemo-fabric-config") == 4 assert swebench.count("fabric_python=/opt/nemo-fabric-venv/bin/python") == 6 assert "--job-name django-13741-opencode-install" not in swebench assert "--job-name django-13741-opencode" not in swebench assert ( 'OPENCODE_JOB_NAME="django-13741-opencode-$(date +%Y%m%d-%H%M%S)"' in swebench ) assert '--job-name "${OPENCODE_JOB_NAME}-install"' in swebench assert '--job-name "$OPENCODE_JOB_NAME"' in swebench for value in ( '--path "$CLINE_SWEBENCH_TASK"', "fabric_adapter_id=nvidia.fabric.cline", ): assert swebench.count(value) == 2 assert swebench.count("--model nvidia/nemotron-3-super-120b-a12b") == 1 assert 'CLINE_JOB_NAME="django-13741-cline-$(date +%Y%m%d-%H%M%S)"' in swebench assert '--job-name "${CLINE_JOB_NAME}-install"' in swebench assert '--job-name "$CLINE_JOB_NAME"' in swebench assert "--ae 'NVIDIA_API_KEY=${NVIDIA_API_KEY}'" in swebench assert '"$RUNS_DIR/$OPENCODE_JOB_NAME/result.json"' in swebench assert '"$RUNS_DIR/$CLINE_JOB_NAME/result.json"' in swebench assert "export JOB_NAME=django-13741-hermes" in swebench for flag in ( "--path", "--agent", "--ak", "--job-name", ): assert flag in calculator for value in ( "swe-bench/swe-bench-verified", "django__django-13741", "--task swe-bench/django__django-13741", "--model", "--skill", "--mcp-config", "fabric_blocked_tools", "fabric_telemetry=relay", "harbor job resume", "telemetry-validation.json", "agent/trajectory.json", ): assert value in swebench def test_harbor_generated_paths_are_ignored(): ignore = (CALCULATOR_ROOT / ".gitignore").read_text(encoding="utf-8") assert "runs/" in ignore assert "task/environment/vendor/" in ignore assert "task/environment/.vendor.*/" in ignore def test_swebench_setup_pins_a_supported_relay_cli(): from nemo_fabric_adapters.common.relay_version import RELAY_MAXIMUM_VERSION from nemo_fabric_adapters.common.relay_version import RELAY_MINIMUM_VERSION swebench = SWEBENCH_README.read_text(encoding="utf-8") pinned = re.search(r"NEMO_RELAY_VERSION=(\d+)\.(\d+)\.(\d+) ", swebench) assert pinned is not None version = tuple(int(part) for part in pinned.groups()) assert RELAY_MINIMUM_VERSION <= version < RELAY_MAXIMUM_VERSION assert '--install-dir "$FABRIC_BUNDLE/.relay/bin"' in swebench def test_swebench_opencode_image_uses_locked_npm_dependencies(): dockerfile = SWEBENCH_OPENCODE_DOCKERFILE.read_text(encoding="utf-8") assert "npm ci" in dockerfile assert "--ignore-scripts" in dockerfile assert "package-lock.json" in dockerfile assert "--no-package-lock" not in dockerfile def test_swebench_cline_image_uses_supported_sdk_and_locked_adapter_dependencies(): dockerfile = SWEBENCH_CLINE_DOCKERFILE.read_text(encoding="utf-8") assert "npm ci" in dockerfile assert "--workspace nemo-fabric-adapters-cline" in dockerfile assert "@cline/sdk@0.0.83" in dockerfile assert "--ignore-scripts" in dockerfile assert "package-lock.json" in dockerfile def test_harbor_calculator_setup_and_solution_fail_fast(): dockerfile = CALCULATOR_DOCKERFILE.read_text(encoding="utf-8") solution = CALCULATOR_SOLUTION.read_text(encoding="utf-8") assert ( "curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs |" not in dockerfile ) assert "-o /tmp/rustup-init.sh" in dockerfile assert "if updated == source:" in solution assert "raise SystemExit" in solution def test_harbor_relay_telemetry_exports_direct_atof_and_atif(): from nemo_fabric import Fabric from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config config = build_harbor_config( adapter_id="nvidia.fabric.hermes", workspace="/app", model_name="nvidia/nemotron-3-nano-omni-30b-a3b-reasoning", telemetry="relay", ) assert config.harness.settings == {} plan = Fabric().plan(config, base_dir=CALCULATOR_FABRIC_ROOT) assert plan.adapter.adapter_id == "nvidia.fabric.hermes" mapping = config.to_mapping() assert "relay" in mapping["telemetry"]["providers"] observability = mapping["relay"]["observability"] swebench = SWEBENCH_README.read_text(encoding="utf-8") assert "openinference" not in observability assert observability["atof"]["enabled"] is True assert observability["atif"]["enabled"] is True assert not (CALCULATOR_ROOT / "host-gateway.compose.yaml").exists() assert "direct Relay ATOF and ATIF" in swebench assert "telemetry-validation.json" in swebench assert "canonical ATIF" in swebench def test_harbor_sdk_package_points_to_the_public_example(): from nemo_fabric.integrations.harbor import FabricAgent readme = SDK_INTEGRATION_README.read_text(encoding="utf-8") package_init = HARBOR_PACKAGE_INIT.read_text(encoding="utf-8") assert FabricAgent.name() == "fabric" assert FabricAgent.__module__ == "nemo_fabric.integrations.harbor.fabric_agent" assert "nemo_fabric.integrations.harbor:FabricAgent" in readme assert "examples/harbor/README.md" in readme assert "FabricConfig" in readme assert "Internals" not in readme assert "class FabricAgent" not in package_init assert "fabric_agent import FabricAgent" in package_init def test_swebench_matrix_translates_harbor_inputs_to_typed_config(tmp_path: Path): from harbor.cli.utils import load_mcp_servers from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config from nemo_fabric.integrations.harbor.models import HarborMcpServer base = build_harbor_config( adapter_id="nvidia.fabric.hermes", workspace="/testbed", ) relay = build_harbor_config( adapter_id="nvidia.fabric.hermes", workspace="/testbed", telemetry="relay", model_name="nvidia/nemotron-3-nano-omni-30b-a3b-reasoning", mcp_servers=tuple( HarborMcpServer.model_validate(server.model_dump(mode="python")) for server in load_mcp_servers(SWEBENCH_MCP_CONFIG) ), ) tools = build_harbor_config( adapter_id="nvidia.fabric.deepagents", workspace="/testbed", blocked_tools=["browser"], telemetry="relay", ) selected_tools = build_harbor_config( adapter_id="nvidia.fabric.hermes", workspace="/testbed", enabled_tools=[], blocked_tools=["browser"], telemetry="relay", ) claude = build_harbor_config( adapter_id="nvidia.fabric.claude", workspace="/testbed", harness_settings={"permission_mode": "bypassPermissions"}, environment_env={"IS_SANDBOX": "1"}, ) assert base.environment is not None assert str(base.environment.workspace) == "/testbed" assert base.models == {} assert base.skills is None assert base.mcp is None assert base.tools is None assert base.telemetry is None assert ( relay.models["default"].model == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" ) assert relay.skills is None assert relay.mcp is not None assert set(relay.mcp.servers) == {"fabric-repo-inspector"} assert relay.mcp.servers["fabric-repo-inspector"].args == [ "/tmp/nemo-fabric-config/mcp/repo_inspector.py" ] assert "args" not in relay.mcp.servers["fabric-repo-inspector"].extra_fields assert tools.tools is not None assert tools.tools.blocked == ["browser"] assert tools.tools.enabled is None assert selected_tools.tools is not None assert selected_tools.tools.enabled == [] assert selected_tools.tools.blocked == ["browser"] assert relay.telemetry is not None assert "relay" in relay.telemetry.providers assert relay.relay is not None assert relay.relay.observability.atif.enabled is True assert relay.relay.observability.atof.enabled is True assert claude.harness.settings["permission_mode"] == "bypassPermissions" assert claude.environment is not None assert claude.environment.env == {"IS_SANDBOX": "1"} assert not list(SWEBENCH_ROOT.rglob("*.yaml")) assert not (SWEBENCH_ROOT / "harbor_swebench_config.py").exists() # TODO: Remove the bundled copies and these equality checks after Fabric # discovers adapter descriptors directly from source checkouts and wheels. assert ( SWEBENCH_ROOT / "adapters/hermes/hermes.fabric-adapter.json" ).read_text() == ( ROOT / "adapters/python/hermes/hermes.fabric-adapter.json" ).read_text() assert ( SWEBENCH_ROOT / "adapters/claude/claude.fabric-adapter.json" ).read_text() == ( ROOT / "adapters/python/claude/claude.fabric-adapter.json" ).read_text() hermes_descriptor = json.loads( (SWEBENCH_ROOT / "adapters/hermes/hermes.fabric-adapter.json").read_text() ) assert "models" in hermes_descriptor["config"]["accepts"] readme = SWEBENCH_README.read_text(encoding="utf-8") assert readme.count("django__django-13741") >= 4 assert "--n-tasks 5" in readme def test_harbor_lifecycle_populates_context_after_run(tmp_path: Path): from harbor.models.agent.context import AgentContext from harbor.trial.trial import Trial from nemo_fabric.integrations.harbor import FabricAgent result = { "agent_name": "harbor-hermes", "harness": "hermes", "adapter_kind": "python", "adapter_id": "nvidia.fabric.hermes", "status": "succeeded", "runtime_id": "runtime-1", "invocation_id": "invocation-1", "request_id": "request-1", "output": {"response": "done"}, "error": None, "artifacts": {"root": "/logs/agent", "artifacts": []}, "telemetry": [], "events": [], "metadata": {}, } class FakeEnvironment: async def upload_file(self, _source, _destination): return None async def exec(self, *_args, **_kwargs): return SimpleNamespace(return_code=0, stdout="", stderr="") async def download_file(self, _source, destination): destination.write_text(json.dumps(result), encoding="utf-8") agent = FabricAgent( logs_dir=tmp_path, fabric_adapter_id="nvidia.fabric.hermes", ) context = AgentContext() asyncio.run(agent.run("fix it", FakeEnvironment(), context)) assert context.is_empty() (tmp_path / "trajectory.json").write_text( json.dumps( { "schema_version": "ATIF-v1.7", "session_id": "runtime-1", "agent": {"name": "fabric", "version": "0.1.0"}, "steps": [{"step_id": 1, "source": "agent", "message": "done"}], "final_metrics": { "total_prompt_tokens": 12, "total_cached_tokens": 3, "total_completion_tokens": 4, "total_cost_usd": 0.25, }, } ), encoding="utf-8", ) Trial._populate_agent_context( SimpleNamespace(agent=agent, user_agent=None), context ) assert context.metadata["fabric"]["status"] == "succeeded" atif_metrics = context.metadata["fabric"]["atif_final_metrics"] assert atif_metrics["source"] == "atif" assert atif_metrics["coverage"] == "unverified" assert atif_metrics["final_metrics"]["total_prompt_tokens"] == 12 assert atif_metrics["final_metrics"]["total_cached_tokens"] == 3 assert atif_metrics["final_metrics"]["total_completion_tokens"] == 4 assert atif_metrics["final_metrics"]["total_cost_usd"] == 0.25 assert context.n_input_tokens is None assert context.n_cache_tokens is None assert context.n_output_tokens is None assert context.cost_usd is None def test_harbor_023_options_schema_and_preflight(): from harbor.agents.factory import AgentFactory from harbor.models.trial.config import AgentConfig from nemo_fabric.integrations.harbor import FabricAgent schema = FabricAgent.options_schema() assert schema["required"] == ["fabric_adapter_id"] assert set(schema["properties"]) == { "fabric_adapter_descriptor", "fabric_adapter_id", "fabric_blocked_tools", "fabric_config_base_dir", "fabric_config_bundle", "fabric_config_target", "fabric_cwd", "fabric_discovery_paths", "fabric_enabled_tools", "fabric_environment_env", "fabric_harness_settings", "fabric_install_command", "fabric_max_turns", "fabric_model_api_key_env", "fabric_model_base_url", "fabric_package", "fabric_python", "fabric_runtime_timeout_seconds", "fabric_system_instruction", "fabric_system_stats", "fabric_telemetry", "fabric_timeout_sec", "fabric_target_descriptor", "fabric_venv_path", "fabric_workspace", "fabric_workflow_settings", "fabric_workflow_target_id", } assert FabricAgent.capabilities.atif is False for field in ("skills", "mcp_servers"): if field in type(FabricAgent.capabilities).model_fields: assert getattr(FabricAgent.capabilities, field) is False agent = AgentConfig( import_path="nemo_fabric.integrations.harbor:FabricAgent", kwargs={"fabric_adapter_id": "nvidia.fabric.hermes"}, ) AgentFactory.run_preflight(agent) agent.kwargs["unexpected_option"] = True with pytest.raises(ValueError, match="Unknown option 'unexpected_option'"): AgentFactory.run_preflight(agent) del agent.kwargs["unexpected_option"] agent.kwargs["fabric_workspace"] = "relative/path" with pytest.raises(ValueError, match="fabric_workspace must be an absolute"): AgentFactory.run_preflight(agent) @pytest.mark.parametrize( ("arguments", "expected"), [ ( [], {"collect": ["process", "cgroup.cpu", "cgroup.memory", "cgroup.pressure"]}, ), (["fabric_system_stats=null"], None), ( ["fabric_system_stats={}"], {"collect": ["process", "cgroup.cpu", "cgroup.memory", "cgroup.pressure"]}, ), ( [ 'fabric_system_stats={"collect":["process","cgroup.cpu"],"sampling":{"interval_ms":100,"max_samples":2}}' ], { "collect": ["process", "cgroup.cpu"], "sampling": {"interval_ms": 100, "max_samples": 2}, }, ), ], ) def test_harbor_stats_cli_options_survive_runner_transport( tmp_path, arguments, expected ): from harbor.agents.factory import AgentFactory from harbor.cli.utils import parse_kwargs from nemo_fabric.integrations.harbor.models import FabricRunPayload agent = AgentFactory.create_agent_from_import_path( "nemo_fabric.integrations.harbor:FabricAgent", logs_dir=tmp_path, fabric_adapter_id="nvidia.fabric.langchain.deepagents", **parse_kwargs(arguments), ) payload = agent._build_spec("review the task") wire = payload.model_dump_json(exclude_none=True) assert json.loads(wire)["config"]["system_stats"] == expected decoded = FabricRunPayload.model_validate_json(wire) assert decoded.config.to_mapping()["system_stats"] == expected assert agent._build_config().harness.settings == {} @pytest.mark.parametrize( "stats", [ {"collect": []}, {"collect": ["unknown"]}, {"collect": ["process", "process"]}, {"sampling": {"interval_ms": 99}}, {"sampling": {"max_samples": 1}}, {"sampling": {"interval_ms": True}}, {"unknown": True}, ], ) def test_harbor_rejects_invalid_stats_during_preflight(stats): from harbor.agents.factory import AgentFactory from harbor.models.trial.config import AgentConfig agent = AgentConfig( import_path="nemo_fabric.integrations.harbor:FabricAgent", kwargs={ "fabric_adapter_id": "nvidia.fabric.langchain.deepagents", "fabric_system_stats": stats, }, ) with pytest.raises(ValueError): AgentFactory.run_preflight(agent) def test_harbor_023_factory_loads_fabric_agent(tmp_path: Path): from harbor.agents.factory import AgentFactory agent = AgentFactory.create_agent_from_import_path( "nemo_fabric.integrations.harbor:FabricAgent", logs_dir=tmp_path, fabric_adapter_id="nvidia.fabric.hermes", fabric_max_turns="12", ) assert agent.name() == "fabric" assert agent.SUPPORTS_ATIF is False assert agent.fabric_max_turns == 12 def test_swebench_mcp_config_uses_the_bundled_repo_inspector(): from harbor.cli.utils import load_mcp_servers [server] = load_mcp_servers(SWEBENCH_MCP_CONFIG) assert server.transport == "stdio" assert server.command == "python3" assert server.args == ["/tmp/nemo-fabric-config/mcp/repo_inspector.py"] def test_root_readme_routes_to_primary_documentation(): readme = ROOT_README.read_text(encoding="utf-8") primary_documentation = ( "docs/getting-started/install.mdx", "docs/getting-started/quickstart.mdx", "docs/about-nemo-fabric/overview.mdx", "adapters/README.md", "CONTRIBUTING.md", ) for path in primary_documentation: assert f"]({path})" in readme def test_registered_workflow_options_preserve_selection_and_settings(tmp_path): from nemo_fabric import inspect_adapter from nemo_fabric.integrations.harbor import FabricAgent from nemo_fabric_adapter_catalog import ( get_adapter_descriptor, get_target_descriptor, ) agent = FabricAgent( logs_dir=tmp_path, fabric_adapter_id="nvidia.fabric.nooa", fabric_workflow_target_id="nvidia.nooa.coding-agent", fabric_workflow_settings={}, ) payload = agent._build_spec("Solve the task.") assert payload.config.harness.adapter_id == "nvidia.fabric.nooa" assert payload.config.workflow.target_id == "nvidia.nooa.coding-agent" assert payload.config.workflow.settings == {} profile = inspect_adapter( payload.config, get_adapter_descriptor("nvidia.fabric.nooa"), target_descriptor=get_target_descriptor("nvidia.nooa.coding-agent"), ) assert payload.adapter_descriptor_sha256 == profile.descriptor_sha256 assert agent._capability_profile.mcp and agent._capability_profile.skills @pytest.mark.parametrize( "arguments", [ {"fabric_workflow_settings": {}}, {"fabric_workflow_target_id": " "}, {"fabric_workflow_target_id": "nvidia.nooa.coding-agent "}, { "fabric_workflow_target_id": "nvidia.nooa.coding-agent", "fabric_workflow_settings": [], }, {"fabric_target_descriptor": "target.json"}, ], ) def test_workflow_preflight_rejects_incomplete_selection(arguments): from nemo_fabric.integrations.harbor.fabric_agent import FabricAgentOptions with pytest.raises(ValueError): FabricAgentOptions(fabric_adapter_id="nvidia.fabric.nooa", **arguments) def test_workflow_preflight_rejects_wrong_binding_and_target_settings(tmp_path): from nemo_fabric import FabricConfigError from nemo_fabric.integrations.harbor import FabricAgent for adapter, settings in [ ("nvidia.fabric.hermes", {}), ("nvidia.fabric.nooa", {"unexpected": "private-value"}), ]: with pytest.raises(FabricConfigError) as error: FabricAgent( logs_dir=tmp_path, fabric_adapter_id=adapter, fabric_workflow_target_id="nvidia.nooa.coding-agent", fabric_workflow_settings=settings, ) assert "private-value" not in str(error.value) def test_external_workflow_metadata_is_not_task_discovery(tmp_path): from nemo_fabric.integrations.harbor import FabricAgent agent = FabricAgent( logs_dir=tmp_path, fabric_adapter_id="nvidia.fabric.nooa", fabric_workflow_target_id="nvidia.tests.nooa.echo", fabric_workflow_settings={"prefix": "test"}, fabric_adapter_descriptor=ROOT / "adapters/python/nooa/nooa.fabric-adapter.json", fabric_target_descriptor=ROOT / "tests/fixtures/nooa/echo.fabric-target.json", fabric_discovery_paths=["/task-only/descriptors"], ) payload = agent._build_spec("hello") assert payload.config.discovery.local_paths == ["/task-only/descriptors"] assert payload.config.workflow.settings == {"prefix": "test"} assert payload.adapter_descriptor_sha256 is not None @pytest.mark.asyncio @pytest.mark.usefixtures("nemo_relay") async def test_registered_workflow_round_trip_and_target_drift(tmp_path, monkeypatch): import os import sys from nemo_fabric import FabricConfigError from nemo_fabric import RunRequest from nemo_fabric import inspect_adapter from nemo_fabric.integrations.harbor import runner from nemo_fabric.integrations.harbor.fabric_agent import build_harbor_config from nemo_fabric.integrations.harbor.models import FabricRunPayload fixture = ROOT / "tests/fixtures/nooa" target_path = tmp_path / "echo.fabric-target.json" target = json.loads((fixture / "echo.fabric-target.json").read_text()) target_path.write_text(json.dumps(target)) adapter_path = ROOT / "adapters/python/nooa/nooa.fabric-adapter.json" descriptor = json.loads(adapter_path.read_text()) monkeypatch.setenv("ADAPTER_PYTHON", sys.executable) monkeypatch.setenv( "PYTHONPATH", os.pathsep.join([str(fixture / "src"), os.environ.get("PYTHONPATH", "")]), ) config = build_harbor_config( adapter_id="nvidia.fabric.nooa", workspace=str(tmp_path), workflow_target_id="nvidia.tests.nooa.echo", workflow_settings={}, discovery_paths=(adapter_path, target_path), ) config.runtime.artifacts = str(tmp_path / "artifacts") config.environment.artifacts = str(tmp_path / "artifacts") config.enable_relay(output_dir=tmp_path / "relay") profile = inspect_adapter(config, descriptor, target_descriptor=target) payload = FabricRunPayload( config=config, config_base_dir=str(tmp_path), logs_dir=str(tmp_path / "logs"), request=RunRequest(input="hello"), adapter_descriptor_sha256=profile.descriptor_sha256, ) result = await runner.run(payload) assert result.status == "succeeded", result.to_mapping() assert result.output["response"] == "reply-1: hello" target["spec"]["entrypoint"]["ref"] = "not_imported.factory:create_agent" target_path.write_text(json.dumps(target)) with pytest.raises(FabricConfigError, match="descriptor mismatch"): await runner.run(payload)