fix(evaluator): guard HarborRunner submissions to be subprocess only

docs/evaluator/agent-eval/harbor-runner.mdx

@@ -16,12 +16,16 @@ the verifier reward through agent-eval — the same
 Unlike the quickstart, this runner is **not** zero-dependency — it shells out to Harbor and Docker:
 
 - **Python ≥ 3.12**
-- **Docker** installed and running
+- **Docker and docker daemon** installed and running
 - **Harbor**, installed separately: `uv pip install "harbor>=0.16.1"`. Harbor is intentionally **not**
   a dependency of `nemo-platform[nemo-evaluator-sdk]`, so the rest of the SDK stays lightweight.
 
 The runner raises a clear error pointing at this install step if `harbor` is missing.
 
+Evaluator plugin submissions (`nemo evaluator agent-evaluate submit`) support the Harbor runner only
+through the host `subprocess` executor. Standalone SDK runs using `run_harbor_eval()` or
+`HarborAgentTaskRunner` invoke Harbor directly and do not use a Jobs execution profile.
+
 </Warning>
 
 ## The dataset

packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py

@@ -54,7 +54,7 @@
 from nemo_platform_plugin.entities import EntityClient
 from nemo_platform_plugin.jobs.client import AsyncJobsClient
 from nemo_platform_plugin.jobs.docker import validate_gpu_available_for_docker
-from nemo_platform_plugin.jobs.exceptions import PlatformJobCompilationError
+from nemo_platform_plugin.jobs.exceptions import PlatformJobCompilationError, PlatformJobDependencyUnavailableError
 from nemo_platform_plugin.jobs.openapi_utils import generate_openapi_extra_params
 from nemo_platform_plugin.jobs.result_manager import download_from_result_info
 from nemo_platform_plugin.jobs.schemas import (
@@ -670,6 +670,7 @@ async def _compile_platform_spec(
 
     Raises:
         HTTPException(422): If the compiler raises PlatformJobCompilationError.
+        HTTPException(503): If the compiler raises PlatformJobDependencyUnavailableError.
         PermissionError: If the compiler raises a PermissionError.
     """
     try:
@@ -688,6 +689,11 @@ async def _compile_platform_spec(
             status_code=status.HTTP_403_FORBIDDEN,
             detail=str(e),
         ) from e
+    except PlatformJobDependencyUnavailableError as e:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Temporarily unable to compile {service_name} job spec: {str(e)}",
+        ) from e
     except PlatformJobCompilationError as e:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,

packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/exceptions.py

@@ -6,3 +6,9 @@ class PlatformJobCompilationError(Exception):
     """Exception raised for errors in the platform job compilation process."""
 
     pass
+
+
+class PlatformJobDependencyUnavailableError(Exception):
+    """A dependency required to compile a platform job is temporarily unavailable."""
+
+    pass

packages/nmp_common/tests/api_factory/test_api_factory.py

@@ -44,6 +44,7 @@
     _validate_job_spec,
     job_route_factory,
 )
+from nemo_platform_plugin.jobs.exceptions import PlatformJobDependencyUnavailableError
 from nmp.common.errors.sdk_exception_handlers import register_sdk_exception_handlers
 from nmp.common.jobs.exceptions import PlatformJobCompilationError
 from nmp.common.jobs.schemas import (
@@ -1744,6 +1745,33 @@ def bad_compiler(workspace, input_spec, output_spec, entity_client, job_name, sd
         assert "my_svc" in exc_info.value.detail
         assert "missing field" in exc_info.value.detail
 
+    @pytest.mark.anyio
+    async def test_dependency_unavailable_error_becomes_503(self):
+        """A compiler dependency outage is reported as a service outage."""
+        from fastapi import HTTPException
+
+        def unavailable_compiler(workspace, input_spec, output_spec, entity_client, job_name, sdk):
+            raise PlatformJobDependencyUnavailableError(
+                "The Jobs service is temporarily unavailable. Retry the submission."
+            )
+
+        spec = FooJobConfig(foo="a", bar=1)
+        with pytest.raises(HTTPException) as exc_info:
+            await _compile_platform_spec(
+                unavailable_compiler,
+                "ws",
+                spec,
+                spec,
+                MagicMock(),
+                "name",
+                "my_svc",
+                MagicMock(),
+            )
+        assert exc_info.value.status_code == 503
+        assert "my_svc" in exc_info.value.detail
+        assert "temporarily unavailable" in exc_info.value.detail
+        assert "Retry the submission" in exc_info.value.detail
+
     @pytest.mark.anyio
     async def test_validate_job_spec_is_called(self):
         """_validate_job_spec is invoked on the compiled result (catches non-serializable config)."""

plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py

@@ -20,7 +20,7 @@
 import logging
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, ClassVar, cast
+from typing import Any, ClassVar, Literal, cast
 from urllib.parse import urlsplit
 
 import nemo_evaluator.agent_seeds  # noqa: F401 - registers the platform 'fileset' workspace-seed handler
@@ -52,10 +52,15 @@
 from nemo_evaluator_sdk.metrics.protocol import Metric
 from nemo_evaluator_sdk.values import RunConfigOnline, RunConfigOnlineModel
 from nemo_platform import AsyncNeMoPlatform, NeMoPlatform
+from nemo_platform_plugin.client.adapter import client_from_platform
+from nemo_platform_plugin.client.errors import InternalServerError, NemoResponseValidationError, NemoTransportError
 from nemo_platform_plugin.entities import EntityClient
 from nemo_platform_plugin.job import NemoJob
 from nemo_platform_plugin.job_context import JobContext
-from nemo_platform_plugin.jobs.api_factory import PlatformJobSpec
+from nemo_platform_plugin.jobs.api_factory import PlatformJobSpec, SubprocessExecutionProviderSpec
+from nemo_platform_plugin.jobs.client import AsyncJobsClient
+from nemo_platform_plugin.jobs.exceptions import PlatformJobCompilationError, PlatformJobDependencyUnavailableError
+from nemo_platform_plugin.jobs.execution_profiles import SubprocessJobExecutionProfile
 from pydantic import BaseModel
 
 logger = logging.getLogger(__name__)
@@ -66,6 +71,26 @@
 AGENT_BUNDLE_DIR = "agent-eval"
 SUMMARY_FILE_NAME = "summary.json"
 
+# Shared tail for every Harbor backend-compatibility rejection
+_HARBOR_BACKEND_REQUIREMENT = (
+    "Harbor targets currently require local execution or the subprocess backend with access to the host Docker daemon."
+)
+_SUBPROCESS_PROVIDER: Literal["subprocess"] = "subprocess"
+
+
+def _harbor_backend_error(reason: str) -> PlatformJobCompilationError:
+    """A Harbor backend rejection: the specific cause followed by the shared requirement."""
+    return PlatformJobCompilationError(f"{reason} {_HARBOR_BACKEND_REQUIREMENT}")
+
+
+def _harbor_dependency_unavailable(profile: str) -> PlatformJobDependencyUnavailableError:
+    """A retryable failure while resolving the backend for a Harbor profile."""
+    return PlatformJobDependencyUnavailableError(
+        f"Unable to resolve execution profile '{profile}': the Jobs service is temporarily unavailable. "
+        "Retry the submission."
+    )
+
+
 #: Identity headers forwarded from the job's platform SDK to online inference so a platform-routed
 #: target authenticates as the job's principal (``get_task_sdk`` emits these). An explicit allowlist
 #: — not an ``X-NMP-*`` prefix match — so trace/metadata headers the SDK may add later never leak to
@@ -187,9 +212,60 @@ async def compile(
         options: dict | None = None,
     ) -> PlatformJobSpec:
         """Compile the canonical spec into a plugin-native agent-evaluation job."""
-        del workspace, entity_client, job_name, async_sdk, options
+        del workspace, entity_client, job_name, options
         canonical_spec = spec if isinstance(spec, AgentEvalSpec) else AgentEvalSpec.model_validate(spec.model_dump())
-        return compile_agent_eval_job(canonical_spec, profile=profile)
+        platform_spec = compile_agent_eval_job(canonical_spec, profile=profile)
+        if isinstance(canonical_spec.target, HarborRunnerTarget):
+            step = next(iter(platform_spec["steps"]))
+            executor = cast(dict[str, Any], step["executor"])
+            step["executor"] = await cls._resolve_harbor_subprocess_executor(
+                executor=executor,
+                async_sdk=async_sdk,
+            )
+        return platform_spec
+
+    @staticmethod
+    async def _resolve_harbor_subprocess_executor(
+        *, executor: dict[str, Any], async_sdk: AsyncNeMoPlatform | None
+    ) -> SubprocessExecutionProviderSpec:
+        """Resolve Harbor's selected profile to an explicit host subprocess executor."""
+        profile = cast(str, executor["profile"])
+        provider = cast(str, executor["provider"])
+        if async_sdk is None:
+            raise _harbor_dependency_unavailable(profile)
+
+        try:
+            profiles = (await client_from_platform(async_sdk, AsyncJobsClient).get_execution_profiles()).data()
+        except (NemoTransportError, NemoResponseValidationError, InternalServerError) as exc:
+            raise _harbor_dependency_unavailable(profile) from exc
+
+        # The concrete profile type fixes the backend to "subprocess".
+        if any(
+            isinstance(execution_profile, SubprocessJobExecutionProfile) and execution_profile.profile == profile
+            for execution_profile in profiles
+        ):
+            container = cast(dict[str, Any], executor["container"])
+            command = [*(container.get("entrypoint") or []), *(container.get("command") or [])]
+            if not command:
+                raise _harbor_backend_error(
+                    f"Unable to compile execution profile '{profile}' for subprocess execution: the step command is empty."
+                )
+            return SubprocessExecutionProviderSpec(provider=_SUBPROCESS_PROVIDER, profile=profile, command=command)
+
+        # Jobs keys execution profiles by (provider, profile), so at most one backend can match.
+        resolved_backend = next(
+            (
+                execution_profile.backend
+                for execution_profile in profiles
+                if execution_profile.profile == profile and execution_profile.provider == provider
+            ),
+            None,
+        )
+        raise _harbor_backend_error(
+            f"Execution profile '{profile}' resolves to backend '{resolved_backend}'."
+            if resolved_backend is not None
+            else f"Execution profile '{profile}' does not resolve to a subprocess backend."
+        )
 
     @staticmethod
     def _endpoint_url(target: Target | None) -> str | None:

plugins/nemo-evaluator/tests/integration/conftest.py

@@ -239,6 +239,13 @@ def _materialize_docker_config(work_root: Path, *, base_url: str) -> Path:
     silently run on subprocess. ``platform.runtime: docker`` + a ``cpu/default`` docker executor
     keeps the agent-eval step on the docker backend. The jobs-launcher binary is optional (the
     backend falls back to the container's own entrypoint when it's absent), so it isn't built here.
+
+    ``enable_subprocess_executor: False`` is what makes that true, and is load-bearing — do not drop
+    it as redundant with the ``executors`` list. Jobs auto-appends a ``subprocess/default`` profile
+    for any non-Kubernetes runtime unless this is set (see
+    ``get_default_executor_profiles_for_runtime``). The Harbor compiler deliberately selects any
+    advertised subprocess profile, so leaving this unset would make the fixture subprocess-capable
+    instead of Docker-only.
     """
     docker_executor_config = {
         "launcher_tool_path": str(REPO_ROOT / "services/core/jobs/jobs-launcher/jobs-launcher"),
@@ -251,6 +258,7 @@ def _materialize_docker_config(work_root: Path, *, base_url: str) -> Path:
         "platform": {"runtime": "docker", "base_url": base_url},
         "auth": {"enabled": False, "allow_unsigned_jwt": True},
         "jobs": {
+            "enable_subprocess_executor": False,
             "executors": [
                 {"provider": "cpu", "profile": "default", "backend": "docker", "config": docker_executor_config},
             ],

plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py

@@ -50,6 +50,7 @@
     AgentEvalTaskInput,
     AgentTarget,
     CodexRunnerTarget,
+    HarborRunnerTarget,
     ModelTarget,
 )
 from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateJob
@@ -438,6 +439,20 @@ def _codex_eval_input_spec() -> dict:
     ).model_dump(mode="json")
 
 
+def _harbor_eval_input_spec() -> dict:
+    """Minimal Harbor target submission; compilation must reject it before task execution."""
+    return AgentEvalInputSpec(
+        tasks=[
+            AgentEvalTaskInput(
+                id="harbor-task",
+                intent="Exercise Harbor backend compatibility validation.",
+                inputs=TaskInputs(instruction="Reply with DONE."),
+            )
+        ],
+        target=HarborRunnerTarget(agent_name="oracle"),
+    ).model_dump(mode="json")
+
+
 @requires_codex
 @pytest.mark.timeout(600)
 def test_submit_to_subprocess_backend_runs_agent_eval(subprocess_platform: str) -> None:
@@ -638,6 +653,37 @@ def test_submit_model_target_under_auth_forwards_identity_to_igw(auth_subprocess
     assert result.bundle_ref
 
 
+@pytest.mark.timeout(300)
+def test_submit_harbor_target_to_docker_backend_fails_fast(docker_platform: str) -> None:
+    workspace = _unique("harbor-docker-guard")
+    client = NeMoPlatform(base_url=docker_platform, max_retries=2)
+    client.workspaces.create(name=workspace, exist_ok=True)
+
+    with pytest.raises(httpx.HTTPStatusError) as exc_info:
+        NemoJobScheduler().submit_remote(
+            AgentEvalJob,
+            _harbor_eval_input_spec(),
+            base_url=docker_platform,
+            workspace=workspace,
+            profile="default",
+        )
+
+    response = exc_info.value.response
+    assert response.status_code == 422
+    detail = response.json()["detail"]
+    assert "profile 'default'" in detail
+    assert "backend 'docker'" in detail
+    assert "Harbor targets currently require local execution or the subprocess backend" in detail
+
+    jobs = httpx.get(
+        f"{docker_platform}/apis/evaluator/v2/workspaces/{workspace}/agent-evaluate/jobs",
+        params={"page_size": 100},
+        timeout=30,
+    )
+    assert jobs.status_code == 200, jobs.text
+    assert jobs.json()["data"] == []
+
+
 @pytest.mark.timeout(600)
 @pytest.mark.xfail(
     reason="agent-eval can't run under the docker backend until the cpu-tasks image is rebuilt with "