[codex] Add Hermes capability hardening coverage

.github/workflows/ci_python.yml

@@ -54,9 +54,9 @@ jobs:
           uv pip install --python .venv/bin/python -e . "pyyaml>=6"
 
       # Dependency-free smokes only: the gated integration smokes
-      # (smoke_hermes_sdk, smoke_relay_integration, smoke_harbor_*) need an
-      # NVIDIA_API_KEY / a running Hermes / a sibling harbor checkout, so they
-      # are excluded.
+      # (smoke_hermes_sdk, smoke_relay_integration, and the Docker-backed
+      # smoke_harbor_swebench_task) need an NVIDIA_API_KEY / a running Hermes /
+      # a sibling harbor checkout, so they are excluded.
       - name: Run dependency-free smokes
         run: |
           set -euo pipefail
@@ -68,6 +68,7 @@ jobs:
             python/tests/smoke_native_sdk.py
             python/tests/smoke_typed_config.py
             python/tests/smoke_consumer_neutral.py
+            python/tests/smoke_harbor_integration.py
             python/tests/smoke_readme_examples.py
             python/tests/smoke_sdk_sessions.py
             tests/smoke_cli.py

python/tests/smoke_harbor_integration.py

@@ -9,20 +9,55 @@
 import json
 import sys
 import tempfile
+import types
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT / "python" / "src"))
 
+
+def install_harbor_stubs() -> None:
+    """Install minimal Harbor stubs for this smoke when Harbor is not present."""
+
+    class BaseAgent:
+        def __init__(self, logs_dir: Path, *args: Any, **kwargs: Any) -> None:
+            self.logs_dir = logs_dir
+            self.model_name = kwargs.get("model_name")
+            self.skills_dir = kwargs.get("skills_dir")
+            self.mcp_servers = kwargs.get("mcp_servers", [])
+
+    class BaseEnvironment:
+        pass
+
+    class AgentContext:
+        def __init__(self) -> None:
+            self.metadata: dict[str, Any] | None = None
+
+    modules = {
+        "harbor": types.ModuleType("harbor"),
+        "harbor.agents": types.ModuleType("harbor.agents"),
+        "harbor.agents.base": types.ModuleType("harbor.agents.base"),
+        "harbor.environments": types.ModuleType("harbor.environments"),
+        "harbor.environments.base": types.ModuleType("harbor.environments.base"),
+        "harbor.models": types.ModuleType("harbor.models"),
+        "harbor.models.agent": types.ModuleType("harbor.models.agent"),
+        "harbor.models.agent.context": types.ModuleType("harbor.models.agent.context"),
+    }
+    modules["harbor.agents.base"].BaseAgent = BaseAgent
+    modules["harbor.environments.base"].BaseEnvironment = BaseEnvironment
+    modules["harbor.models.agent.context"].AgentContext = AgentContext
+    sys.modules.update(modules)
+
+
 try:
     from nemo_fabric.integrations.harbor import FabricAgent
     from harbor.models.agent.context import AgentContext
-except ImportError as exc:
-    raise SystemExit(
-        "Install Harbor before running this smoke, for example: pip install -e ../harbor"
-    ) from exc
+except ImportError:
+    install_harbor_stubs()
+    from nemo_fabric.integrations.harbor import FabricAgent
+    from harbor.models.agent.context import AgentContext
 
 
 @dataclass
@@ -61,7 +96,23 @@ async def exec(
                     "profile": "env_local",
                     "harness_type": "hermes",
                     "adapter_id": "nvidia.fabric.hermes.sdk",
-                    "artifacts": {"artifacts": []},
+                    "artifacts": {
+                        "root": "/workspace/agent/artifacts",
+                        "artifacts": [
+                            {
+                                "name": "stdout",
+                                "kind": "log",
+                                "path": "/workspace/agent/artifacts/stdout.txt",
+                                "media_type": "text/plain",
+                            },
+                            {
+                                "name": "workspace_patch",
+                                "kind": "patch",
+                                "path": "/workspace/agent/artifacts/workspace.patch",
+                                "media_type": "text/x-diff",
+                            },
+                        ],
+                    },
                     "telemetry": None,
                 }
             )
@@ -98,6 +149,8 @@ async def main() -> None:
     assert context.metadata
     assert context.metadata["fabric"]["status"] == "succeeded"
     assert context.metadata["fabric"]["adapter_id"] == "nvidia.fabric.hermes.sdk"
+    artifacts = context.metadata["fabric"]["artifacts"]["artifacts"]
+    assert {artifact["name"] for artifact in artifacts} == {"stdout", "workspace_patch"}
 
 
 if __name__ == "__main__":

python/tests/smoke_sdk.py

@@ -6,20 +6,29 @@
 from __future__ import annotations
 
 import asyncio
+import json
+import subprocess
 import sys
 import tempfile
 from shutil import copytree
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT / "python" / "src"))
+sys.path.insert(0, str(ROOT / "tests"))
 
+from _utils.utils import (  # noqa: E402
+    assert_process_adapter_native_observability,
+    assert_relay_disabled_native_observability,
+)
 from nemo_fabric import FabricClient
 
+COMMAND = ("cargo", "run", "-q", "-p", "fabric-cli", "--")
+
 
 async def main() -> None:
     async with FabricClient(
-        command=("cargo", "run", "-q", "-p", "fabric-cli", "--"),
+        command=COMMAND,
         cwd=ROOT,
     ) as client:
         await smoke(client)
@@ -28,6 +37,7 @@ async def main() -> None:
 async def smoke(client: FabricClient) -> None:
     example_agent = ROOT / "examples" / "code-review-agent"
     fixture_agent = ROOT / "tests" / "fixtures" / "hermes-shim-agent"
+    process_fixture_agent = ROOT / "tests" / "fixtures" / "hermes-cli-agent"
 
     assert client.validate(example_agent).startswith("validated")
 
@@ -47,13 +57,28 @@ async def smoke(client: FabricClient) -> None:
     assert multi_plan["telemetry_plan"]["relay_enabled"] is True
 
     with tempfile.TemporaryDirectory(prefix="fabric-python-sdk-") as tmpdir:
-        temp_agent = Path(tmpdir) / "hermes-shim-agent"
+        temp_agent = Path(tmpdir) / "hermes-shim-agent-sdk"
+        temp_cli_agent = Path(tmpdir) / "hermes-shim-agent-cli"
+        temp_process_agent = Path(tmpdir) / "hermes-cli-agent-sdk"
+        temp_process_cli_agent = Path(tmpdir) / "hermes-cli-agent-cli"
         copytree(fixture_agent, temp_agent)
+        copytree(fixture_agent, temp_cli_agent)
+        copytree(process_fixture_agent, temp_process_agent)
+        copytree(process_fixture_agent, temp_process_cli_agent)
+
         hermes_result = await client.run(
             temp_agent,
             profile="env_local",
             input_text="hello hermes",
         )
+        hermes_cli_result = call_json(
+            "run",
+            temp_cli_agent,
+            "--profile",
+            "env_local",
+            "--input",
+            "hello hermes",
+        )
         structured = await client.run(
             temp_agent,
             profile="env_local",
@@ -63,6 +88,38 @@ async def smoke(client: FabricClient) -> None:
                 "context": {"task": {"source": "sdk-smoke"}},
             },
         )
+        process_result = await client.run(
+            temp_process_agent,
+            profile="env_local",
+            input_text="hello process adapter",
+        )
+        process_cli_result = call_json(
+            "run",
+            temp_process_cli_agent,
+            "--profile",
+            "env_local",
+            "--input",
+            "hello process adapter",
+        )
+
+        assert_sdk_cli_runresult_parity(
+            hermes_cli_result,
+            hermes_result,
+            adapter_kind="python",
+            adapter_id="test.fabric.hermes_shim",
+            adapter_runner="python",
+            mode="shim",
+        )
+        assert_sdk_cli_runresult_parity(
+            process_cli_result,
+            process_result,
+            adapter_kind="process",
+            adapter_id="nvidia.fabric.hermes.cli",
+            adapter_runner="process",
+            mode="hermes_cli_oneshot",
+        )
+        assert_relay_disabled_native_observability(hermes_result)
+        assert_process_adapter_native_observability(process_result)
 
     assert hermes_result["status"] == "succeeded"
     assert hermes_result["adapter_kind"] == "python"
@@ -76,6 +133,65 @@ async def smoke(client: FabricClient) -> None:
     assert structured["request_id"] == "sdk-structured-request"
     assert structured["output"]["received"] == "hello structured sdk"
 
+    process_response = json.loads(process_result["output"]["response"])
+    assert process_response["fake_hermes"] is True
+    assert process_response["prompt"] == "hello process adapter"
+
+
+def assert_sdk_cli_runresult_parity(
+    cli_result: dict,
+    sdk_result: dict,
+    *,
+    adapter_kind: str,
+    adapter_id: str,
+    adapter_runner: str,
+    mode: str,
+) -> None:
+    comparable_fields = [
+        "agent_name",
+        "profile",
+        "harness_type",
+        "adapter_kind",
+        "adapter_id",
+        "status",
+    ]
+    for field in comparable_fields:
+        assert cli_result[field] == sdk_result[field], field
+
+    assert cli_result.get("error") == sdk_result.get("error")
+    assert cli_result["adapter_kind"] == adapter_kind
+    assert cli_result["adapter_id"] == adapter_id
+    assert cli_result["metadata"]["adapter_runner"] == adapter_runner
+    assert sdk_result["metadata"]["adapter_runner"] == adapter_runner
+    assert cli_result["output"]["harness"] == "hermes"
+    assert sdk_result["output"]["harness"] == "hermes"
+    assert cli_result["output"]["mode"] == mode
+    assert sdk_result["output"]["mode"] == mode
+
+    for result in (cli_result, sdk_result):
+        assert result["status"] == "succeeded"
+        assert result["runtime_id"].startswith("runtime-")
+        assert result["invocation_id"].startswith("invocation-")
+        assert result["request_id"].startswith("request-")
+        assert isinstance(result["artifacts"]["artifacts"], list)
+        assert isinstance(result["events"], list)
+        assert result["events"], "RunResult events should not be empty"
+
+
+def call_json(*args: object) -> dict:
+    completed = subprocess.run(
+        [*COMMAND, *(str(arg) for arg in args)],
+        cwd=ROOT,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if completed.returncode != 0:
+        raise AssertionError(
+            f"command failed: {completed.args}\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
+        )
+    return json.loads(completed.stdout)
+
 
 if __name__ == "__main__":
     asyncio.run(main())

tests/_utils/utils.py

@@ -6,6 +6,37 @@
 import yaml
 
 
+def assert_relay_disabled_native_observability(result: dict):
+    """Assert telemetry-off runs still surface native harness evidence."""
+
+    artifact_by_name = {
+        artifact["name"]: artifact
+        for artifact in result["artifacts"]["artifacts"]
+    }
+    assert "stdout" in artifact_by_name
+    assert "relay_config" not in artifact_by_name
+    assert not any(name.startswith("relay_") for name in artifact_by_name)
+
+    stdout_path = Path(artifact_by_name["stdout"]["path"])
+    assert stdout_path.is_file()
+    assert stdout_path.read_text(encoding="utf-8").strip()
+
+    event_kinds = {event["kind"] for event in result["events"]}
+    assert {"runtime_start", "invocation_start", "invocation_end"} <= event_kinds
+
+    telemetry = result["telemetry"]
+    assert telemetry is not None
+    assert telemetry["relay_enabled"] is False
+
+
+def assert_process_adapter_native_observability(result: dict):
+    """Assert process adapters preserve native evidence and clean process output."""
+
+    assert_relay_disabled_native_observability(result)
+    assert result["output"]["returncode"] == 0
+    assert result["output"]["stderr"] == ""
+
+
 def update_hermes_cli_relay_base_url(code_review_agent_dir: Path, api_server: str):
     """
     Update the base URL in the Hermes CLI relay profile.

tests/smoke_cli.py

@@ -11,6 +11,8 @@
 from pathlib import Path
 from shutil import copytree
 
+from _utils.utils import assert_relay_disabled_native_observability
+
 ROOT = Path(__file__).resolve().parents[1]
 COMMAND = ("cargo", "run", "-q", "-p", "fabric-cli", "--")
 
@@ -58,6 +60,28 @@ def main() -> None:
         assert direct_plan["profile"] == str(direct_profile)
         assert direct_plan["adapter_descriptor"]["descriptor"]["adapter_id"] == "nvidia.fabric.hermes.sdk"
 
+        profile_plans = [
+            ("hermes_sdk", "nvidia.fabric.hermes.sdk", "python", False),
+            ("hermes_cli", "nvidia.fabric.hermes.cli", "process", False),
+            ("hermes_relay", "nvidia.fabric.hermes.sdk", "python", True),
+            ("hermes_cli_relay", "nvidia.fabric.hermes.cli", "process", True),
+        ]
+        for profile, adapter_id, adapter_kind, relay_enabled in profile_plans:
+            profile_plan = call_json("plan", temp_example, "--profile", profile)
+            assert profile_plan["profiles"] == [profile]
+            descriptor = profile_plan["adapter_descriptor"]["descriptor"]
+            assert descriptor["adapter_id"] == adapter_id
+            assert descriptor["adapter_kind"] == adapter_kind
+            assert profile_plan["config"]["runtime"]["mode"] == "oneshot"
+            assert profile_plan["capability_plan"]["native"]["skill_paths"]
+            assert "github" in profile_plan["capability_plan"]["native"]["mcp_servers"]
+            telemetry_plan = profile_plan["telemetry_plan"]
+            assert telemetry_plan["relay_enabled"] is relay_enabled
+            if relay_enabled:
+                assert telemetry_plan["relay_output_dir"]
+            else:
+                assert not telemetry_plan.get("relay_output_dir")
+
         multi_plan = call_json(
             "plan",
             temp_fixture,
@@ -82,6 +106,7 @@ def main() -> None:
         assert hermes["output"]["native_mcp_servers"] == ["github"]
         assert hermes["output"]["managed_skill_paths"] == []
         assert hermes["output"]["managed_mcp_servers"] == []
+        assert_relay_disabled_native_observability(hermes)
 
         request = json.dumps(
             {

tests/smoke_hermes_cli.py

@@ -11,6 +11,8 @@
 from pathlib import Path
 from shutil import copytree
 
+from _utils.utils import assert_process_adapter_native_observability
+
 ROOT = Path(__file__).resolve().parents[1]
 COMMAND = ("cargo", "run", "-q", "-p", "fabric-cli", "--")
 
@@ -46,6 +48,7 @@ def main() -> None:
 
         config_path = Path(result["output"]["hermes_config_path"])
         assert config_path.is_file()
+        assert_process_adapter_native_observability(result)
 
 
 def call_json(*args: object) -> dict: