test(harness-bench): drop double-import in provisioning-failure test

Addresses the review nit: the new test imported tests.harness_bench.bench both
via the top-level `from ... import run_harness` and an inner `import ... as
bench_mod`. Patch resolve_driver_class via monkeypatch's string target instead,
and drop the redundant inner Verdict import (already imported at top). No
behavior change.

tests/harness_bench/bench.py

@@ -8,6 +8,8 @@ subprocess and gateway load bounded.
 
 from __future__ import annotations
 
+import contextlib
+import logging
 from collections.abc import Callable
 from dataclasses import dataclass, field
 
@@ -16,6 +18,8 @@ from tests.harness_bench.profile import BenchProfile
 from tests.harness_bench.transport import resolve_driver_class
 from tests.harness_bench.verdict import Applicability, Priority, ProbeResult, Verdict, reconcile
 
+_logger = logging.getLogger(__name__)
+
 # A progress sink: the bench calls it with human-readable status lines as it
 # spawns harnesses and runs probes. ``None`` (the default) stays silent, which
 # is what the pytest layer wants; the CLI passes a stderr writer so a live run
@@ -176,7 +180,29 @@ async def run_harness(
         f"(model={profile.model}); first turn may take ~10-30s...",
     )
     cells: list[CellResult] = []
-    async with driver_cls(profile, databricks_profile=databricks_profile) as driver:
+    driver_cm = driver_cls(profile, databricks_profile=databricks_profile)
+    try:
+        entered = await driver_cm.__aenter__()
+    except Exception as exc:
+        # Provisioning failed (e.g. an own-auth native whose vendor CLI is
+        # installed but not logged in, so its terminal never wires up). Report
+        # a capability-neutral skip for this harness rather than aborting the
+        # whole run — a multi-harness run must survive one unrunnable harness.
+        #
+        # __aenter__ may have already spawned the server + daemon and opened a
+        # client before raising, so tear those down here or they leak for the
+        # rest of the run (_teardown null-checks each, so a half-provisioned
+        # driver is safe to tear down). Log the traceback: this branch also
+        # catches genuine driver bugs (e.g. an AssertionError), which must not
+        # vanish silently behind a green-looking skip.
+        _logger.warning("provisioning failed for %s", profile.harness, exc_info=True)
+        with contextlib.suppress(Exception):
+            await driver_cm.__aexit__(type(exc), exc, exc.__traceback__)
+        reason = f"provisioning failed: {exc}"
+        _emit(progress, f"[{profile.harness}] skipped: {reason}")
+        return _uniform_report(profile, probes, ProbeResult.skipped(reason), skipped_reason=reason)
+    try:
+        driver = entered
         prereq_skip: str | None = None
         for probe in probes:
             if not _applicable(probe, profile):
@@ -202,6 +228,8 @@ async def run_harness(
             # they would only re-hit the same failure and pollute the matrix.
             if probe.name == _PREREQ_PROBE and cell.observed is not Verdict.SUPPORTED:
                 prereq_skip = f"prerequisite '{probe.title}' did not pass ({observed.note})"
+    finally:
+        await driver_cm.__aexit__(None, None, None)
     return HarnessReport(profile=profile, cells=cells)
 
 

tests/harness_bench/manifest.py

@@ -152,32 +152,50 @@ OFFICIAL_PROFILES: dict[str, BenchProfile] = {
 # ── native-tui harnesses ─────────────────────────────────────────
 #
 # Native harnesses are not in HARNESS_PROBES (that matrix is the SDK-wrap
-# e2e set), so their profiles are built directly here. Both shipped natives
-# are OMNIGENT_CREDENTIAL vendors the native-tui driver can run and observe
-# (see native_tui_driver for the per-vendor provisioning). OWN_AUTH natives
-# (cursor-native, kiro-native, ...) need a vendor login the bench cannot
-# provision, so they are left to a --harness <ref> opt-in.
+# e2e set), so their profiles are derived here directly from the capability
+# model: every harness with integration_mode == NATIVE_TUI is registered, so
+# the shipped natives and any community-plugin native (harness_capabilities()
+# discovers plugins via entry points) are probeable by name with no bench edit.
 #
-# model: native harnesses take the model as a launch --model, not a
-# HARNESS_<H>_MODEL env var, so they are absent from model_env_keys() and
-# their model_override declares UNKNOWN (honest — the probe confirms it live
-# once native model-override observation is wired).
-_NATIVE_PROFILES: dict[str, tuple[str, str]] = {
-    # harness: (model, marker)
-    "claude-native": ("databricks-claude-sonnet-4-6", "CLAUDE_NATIVE_OK"),
-    "codex-native": ("databricks-gpt-5-4-mini", "CODEX_NATIVE_OK"),
+# What the bench can actually *run* is a separate axis from what it registers.
+# OMNIGENT_CREDENTIAL natives (claude, codex) route through the run's Databricks
+# profile, so the bench runs them unattended. OWN_AUTH / session-scoped natives
+# need a vendor login the bench cannot provision; they are still registered
+# (visible, resolvable, honest declared matrix) but skip-gate at the driver's
+# unavailable() on a host without that login.
+#
+# model: an OMNIGENT_CREDENTIAL native routes its launch --model through the
+# gateway, so it takes a databricks-* model; an own-auth native's model lives
+# in the vendor's namespace the bench does not control, and is unused in
+# practice (the harness skip-gates before a turn). model_override still
+# declares UNKNOWN for all natives (absent from model_env_keys()), confirmed
+# live by the probe.
+_NATIVE_CREDENTIAL_MODELS: dict[str, str] = {
+    "claude-native": "databricks-claude-sonnet-4-6",
+    "codex-native": "databricks-gpt-5-4-mini",
+}
+_NATIVE_DEFAULT_MODEL = "databricks-claude-sonnet-4-6"
+
+# The vendor CLI the driver skip-gates on. Usually the harness id minus
+# "-native" (claude-native -> "claude"), but several vendors ship a
+# differently-named binary (the _DEFAULT_*_COMMAND in each omnigent/*_native.py),
+# so those are listed explicitly. A missing/unlisted native falls back to the
+# suffix convention.
+_NATIVE_CLI_BINARY: dict[str, str] = {
+    "cursor-native": "cursor-agent",
+    "kiro-native": "kiro-cli",
 }
 
 
-def _native_profile(harness: str, model: str, marker: str) -> BenchProfile:
-    """Build a native-tui :class:`BenchProfile`, columns/verdicts from capabilities."""
+def _native_profile(harness: str) -> BenchProfile:
+    """Build a native-tui :class:`BenchProfile`; all fields from convention/capabilities."""
     caps = harness_capabilities().get(harness)
-    # The vendor CLI the driver skip-gates on (claude-native -> "claude").
-    cli_binary = harness.removesuffix("-native")
+    cli_binary = _NATIVE_CLI_BINARY.get(harness, harness.removesuffix("-native"))
     env_prefix = "HARNESS_" + harness.upper().replace("-", "_") + "_"
+    marker = harness.upper().replace("-", "_") + "_OK"
     return BenchProfile(
         harness=harness,
-        model=model,
+        model=_NATIVE_CREDENTIAL_MODELS.get(harness, _NATIVE_DEFAULT_MODEL),
         env_prefix=env_prefix,
         marker=marker,
         cli_binary=cli_binary,
@@ -189,8 +207,17 @@ def _native_profile(harness: str, model: str, marker: str) -> BenchProfile:
     )
 
 
-for _h, (_model, _marker) in _NATIVE_PROFILES.items():
-    OFFICIAL_PROFILES[_h] = _native_profile(_h, _model, _marker)
+def _native_tui_harnesses() -> list[str]:
+    """Every harness the capability model marks as native-tui (plugins included)."""
+    return [
+        harness
+        for harness, caps in harness_capabilities().items()
+        if caps.integration_mode is IntegrationMode.NATIVE_TUI
+    ]
+
+
+for _h in _native_tui_harnesses():
+    OFFICIAL_PROFILES[_h] = _native_profile(_h)
 
 
 __all__ = ["OFFICIAL_PROFILES"]

tests/harness_bench/native_tui_driver.py

@@ -33,13 +33,17 @@ records, so adding a harness is a config entry, not a new driver — until a
 vendor diverges in kind (codex-native is RPC-delivered; opencode-native is
 ``native-server`` not ``native-tui``), which will want its own handling.
 
-Scope: this driver ships **claude-native** and **codex-native**, both
-live-verified end to end (basic turn, delta streaming, model override,
-interrupt) on a host with the vendor CLI logged in. They reach the same shared
-observe path by different means: claude-native tails a transcript (forwarder
-auto-starts on bind), codex-native rides an app-server-RPC forwarder that the
-driver wires up via an explicit runner launch/bind + native terminal ensure +
-provider config (``needs_terminal_ensure``; see ``_provision``).
+Scope: this driver runs **any** native-tui harness — the two shipped
+(claude-native, codex-native) and any other in-repo or community-plugin native
+harness — with no per-vendor table. It derives what it needs (agent name,
+terminal name, whether the vendor self-authenticates) from the capability model
+via :func:`native_vendor`, and provisions every native uniformly: launch/bind a
+runner, ensure the native terminal, and wait for the runner-side forwarder to
+come live (all natives stamp ``external_session_id`` once their terminal thread
+starts) before driving turns on the shared observe path. An OMNIGENT_CREDENTIAL
+native (claude, codex) is routed through the run's Databricks profile via a
+written config home; an own-auth native runs only where its vendor CLI is
+already logged in.
 """
 
 from __future__ import annotations
@@ -114,33 +118,51 @@ _READER_TERMINAL = frozenset({_OUTPUT_DONE_EVENT, _FAILED_EVENT, _INTERRUPTED_EV
 class NativeVendor:
     """Per-vendor facts a native-tui harness needs beyond the shared path.
 
+    Derived from the capability model (see :func:`native_vendor`), so a native
+    harness — in-repo or a community plugin — is probeable with no bench edit.
+
     :param harness: The native harness id, e.g. ``"claude-native"``.
-    :param agent_name: The server's auto-registered UI agent, e.g.
-        ``"claude-native-ui"``.
-    :param own_auth: ``True`` when the vendor uses its own login (cannot take
-        a minted bearer); such a harness is only runnable pre-logged-in.
-    :param needs_terminal_ensure: ``True`` when a turn's output only reaches
-        the shared session stream after the vendor's runner-side forwarder is
-        wired up, which requires an explicit runner launch/bind + native
-        terminal ensure during provisioning (codex-native). claude-native's
-        forwarder auto-starts on session bind, so it needs none of this.
+    :param agent_name: The server's auto-registered UI agent, by convention
+        ``"<harness>-ui"`` (e.g. ``"claude-native-ui"``).
+    :param terminal_name: The native terminal to ensure, by convention the
+        vendor CLI name (``"<harness>" minus "-native"``, e.g. ``"codex"``).
+    :param own_auth: ``True`` when the vendor logs in itself (auth is not
+        ``OMNIGENT_CREDENTIAL``), so the bench cannot provision it — runnable
+        only on a host where the vendor CLI is already logged in.
     """
 
     harness: str
     agent_name: str
+    terminal_name: str
     own_auth: bool = False
-    needs_terminal_ensure: bool = False
 
 
-# Both shipped vendors surface output on the shared session stream; codex-native
-# needs extra provisioning first (see ``needs_terminal_ensure``). OWN_AUTH
-# natives are absent (login the bench cannot provision).
-_VENDORS: dict[str, NativeVendor] = {
-    "claude-native": NativeVendor("claude-native", "claude-native-ui", own_auth=False),
-    "codex-native": NativeVendor(
-        "codex-native", "codex-native-ui", own_auth=False, needs_terminal_ensure=True
-    ),
-}
+def native_vendor(harness: str) -> NativeVendor | None:
+    """Derive the :class:`NativeVendor` for *harness* from its capabilities.
+
+    Returns ``None`` unless the harness declares ``integration_mode ==
+    NATIVE_TUI`` in :func:`omnigent.harness_plugins.harness_capabilities`
+    (which already discovers community plugins via entry points), so any
+    native-tui harness is drivable by name with no per-vendor table here.
+    ``native-server`` harnesses (e.g. opencode-native) are a different
+    transport and return ``None``.
+    """
+    from omnigent.harness_capabilities import AuthModel, IntegrationMode
+    from omnigent.harness_plugins import harness_capabilities
+
+    caps = harness_capabilities().get(harness)
+    if caps is None or caps.integration_mode is not IntegrationMode.NATIVE_TUI:
+        return None
+    # agent_name and terminal_name are convention (``<harness>-ui`` and the
+    # vendor CLI name), which holds for every in-repo native. A community
+    # plugin whose registered terminal/agent name diverges would need an
+    # override map here, mirroring the manifest's _NATIVE_CLI_BINARY.
+    return NativeVendor(
+        harness=harness,
+        agent_name=f"{harness}-ui",
+        terminal_name=harness.removesuffix("-native"),
+        own_auth=caps.auth is not AuthModel.OMNIGENT_CREDENTIAL,
+    )
 
 
 class NativeTuiDriver:
@@ -158,7 +180,7 @@ class NativeTuiDriver:
     def __init__(self, profile: BenchProfile, *, databricks_profile: str) -> None:
         self._profile = profile
         self._db_profile = databricks_profile
-        self._vendor = _VENDORS.get(profile.harness)
+        self._vendor = native_vendor(profile.harness)
         self._proc: subprocess.Popen[bytes] | None = None
         self._daemon: subprocess.Popen[bytes] | None = None
         self._client: httpx.Client | None = None
@@ -169,9 +191,9 @@ class NativeTuiDriver:
     @staticmethod
     def unavailable(profile: BenchProfile, *, databricks_profile: str | None) -> str | None:
         """Return a skip reason if this driver cannot run *profile*, else None."""
-        vendor = _VENDORS.get(profile.harness)
+        vendor = native_vendor(profile.harness)
         if vendor is None:
-            return f"no native-tui vendor entry for {profile.harness!r}"
+            return f"{profile.harness!r} is not a native-tui harness"
         if not databricks_profile:
             return "no --profile / databricks profile provided; native-tui needs a gateway route"
         if lookup_databricks_host(databricks_profile) is None:
@@ -239,10 +261,11 @@ class NativeTuiDriver:
             "DATABRICKS_CONFIG_PROFILE": self._db_profile,
             "OMNIGENT_RUNNER_TUNNEL_TOKEN": binding_token,
         }
-        # codex reads its provider from omnigent's global config, not from
-        # DATABRICKS_CONFIG_PROFILE; without it the TUI hits the vendor login
-        # screen and never starts a thread.
-        if self._vendor.needs_terminal_ensure:
+        # An omnigent-credential native resolves its provider from omnigent's
+        # global config, not DATABRICKS_CONFIG_PROFILE; without it some vendors
+        # (codex) hit the login screen and never start a thread. Own-auth
+        # vendors use their own login and are left untouched.
+        if not self._vendor.own_auth:
             base_env["OMNIGENT_CONFIG_HOME"] = str(self._write_provider_config())
         self._proc = spawn_omnigent_server(self._tmp, port, base_env, binding_token)
         self._wait_health()
@@ -263,8 +286,10 @@ class NativeTuiDriver:
         )
         created.raise_for_status()
         self._session_id = str(created.json()["id"])
-        if self._vendor.needs_terminal_ensure:
-            self._wire_native_forwarder(host_id, workspace)
+        # Ensure the native terminal + wait for the forwarder for every
+        # native-tui harness: it is the uniform readiness protocol (all natives
+        # stamp external_session_id once their terminal thread starts).
+        self._wire_native_forwarder(host_id, workspace)
 
     def _write_provider_config(self) -> Path:
         """Write the ``OMNIGENT_CONFIG_HOME`` config that routes the vendor's
@@ -285,16 +310,21 @@ class NativeTuiDriver:
         vendor thread id), which the forwarder sets once its thread starts.
         """
         assert self._client is not None and self._session_id is not None
+        assert self._vendor is not None
         session_id = self._session_id
         self._launch_and_bind_runner(host_id, workspace)
         ensure = self._client.post(
             f"/v1/sessions/{session_id}/resources/terminals",
-            json={"terminal": "codex", "session_key": "main", "ensure_native_terminal": True},
+            json={
+                "terminal": self._vendor.terminal_name,
+                "session_key": "main",
+                "ensure_native_terminal": True,
+            },
             timeout=90.0,
         )
         ensure.raise_for_status()
         # Gate on the forwarder wiring up: it stamps external_session_id (the
-        # codex thread id) on the session once the TUI creates its thread.
+        # vendor thread id) on the session once the TUI creates its thread.
         # Posting a turn before this races ahead of the forwarder subscription.
         deadline = time.monotonic() + _FORWARDER_READY_TIMEOUT_S
         while time.monotonic() < deadline:

tests/harness_bench/test_bench.py

@@ -204,12 +204,55 @@ async def test_full_server_async_shims_delegate_to_sync(monkeypatch: pytest.Monk
     assert any(c.startswith("tool:") and "True" in c for c in calls)
 
 
+async def test_provisioning_failure_skips_and_tears_down(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A driver that raises in __aenter__ yields a skip AND is torn down.
+
+    Provisioning spawns a server + daemon before the step that can fail (an
+    own-auth native whose terminal never wires up), so the failure path must
+    call __aexit__ or those subprocesses leak for the rest of a multi-harness
+    run. Asserts both: the harness is a capability-neutral skip, and teardown ran.
+    """
+    torn_down: list[bool] = []
+
+    class _FailingDriver:
+        transport = "stub"
+
+        def __init__(self, profile: BenchProfile, *, databricks_profile: str) -> None:
+            pass
+
+        @staticmethod
+        def unavailable(profile: BenchProfile, *, databricks_profile: str | None) -> str | None:
+            return None
+
+        async def __aenter__(self):
+            # Simulates _wire_native_forwarder raising after the server/daemon
+            # are already up.
+            raise RuntimeError("native forwarder did not wire up within 90.0s")
+
+        async def __aexit__(self, *exc: object) -> None:
+            torn_down.append(True)
+
+    profile = BenchProfile(
+        harness="stub-native", model="m", env_prefix="HARNESS_STUB_NATIVE_", marker="X"
+    )
+    monkeypatch.setattr(
+        "tests.harness_bench.bench.resolve_driver_class",
+        lambda p, *, override: _FailingDriver,
+    )
+
+    report = await run_harness(profile, databricks_profile="oss", live=True)
+
+    assert report.skipped_reason is not None and "provisioning failed" in report.skipped_reason
+    assert all(c.observed is Verdict.SKIPPED for c in report.cells)
+    assert torn_down == [True], "provisioning-failure path must tear down the driver"
+
+
 # ── native-tui transport (offline) ──────────────────────────────
 
 
 def test_native_tui_registered_and_gates() -> None:
-    """native-tui is in the registry and gates unwired vendors cleanly."""
-    from tests.harness_bench.native_tui_driver import NativeTuiDriver
+    """native-tui is in the registry and derives any native-tui harness."""
+    from tests.harness_bench.native_tui_driver import NativeTuiDriver, native_vendor
     from tests.harness_bench.transport import driver_registry, resolve_driver_class
 
     assert driver_registry()["native-tui"] is NativeTuiDriver
@@ -220,13 +263,19 @@ def test_native_tui_registered_and_gates() -> None:
     )
     assert resolve_driver_class(claude_native, override="native-tui") is NativeTuiDriver
 
-    # A native harness with no vendor entry (not yet wired) is a clean skip,
-    # never a crash — the walking skeleton only wires claude-native.
-    cursor_native = BenchProfile(
-        harness="cursor-native", model="m", env_prefix="HARNESS_CURSOR_NATIVE_", marker="X"
-    )
-    reason = NativeTuiDriver.unavailable(cursor_native, databricks_profile="oss")
-    assert reason is not None and "cursor-native" in reason
+    # Every native-tui harness derives a vendor from the capability model with
+    # no per-vendor table — an own-auth native (cursor) as much as a shipped
+    # credential one (claude). This is what lets a community-plugin native run
+    # by name with no bench edit.
+    assert native_vendor("claude-native") is not None
+    cursor = native_vendor("cursor-native")
+    assert cursor is not None and cursor.own_auth is True
+
+    # A non-native-tui harness derives no vendor and gates cleanly: an SDK
+    # harness, or a native-server one (opencode-native), is not this driver's.
+    assert native_vendor("claude-sdk") is None
+    codex_sdk = BenchProfile(harness="codex", model="m", env_prefix="X_", marker="X")
+    assert NativeTuiDriver.unavailable(codex_sdk, databricks_profile="oss") is not None
 
     # No profile → the same capability-neutral skip contract as other drivers.
     assert NativeTuiDriver.unavailable(claude_native, databricks_profile=None) is not None