feat(harness): add NativeHarnessProvider seam foundation (PR 1.1)

designs/harness-modular-registry-proposal.md

@@ -353,6 +353,16 @@ Only starts once Phase 1 has every built-in running *through* the seam.
   2.2 → 2.3 (web); the risk center is **PR 1.5**, where the `_supervise_*_bridges`
   invariants live.
 
+### Implementation progress
+
+Append-only ledger — one line per PR as it opens, updated to `landed` on merge.
+The plan tables above stay the stable target; this tracks what has actually
+shipped. **12 PRs total** (Phase 1: 1.1–1.8, Phase 2: 2.1–2.4).
+
+| PR | Status | Link |
+|---|---|---|
+| 1.1 Provider model + resolver | in review | #3239 |
+
 ## Risks and open questions
 
 - **Runner extraction is the risk center.** The `_supervise_*_bridges` mirrors

omnigent/harness_plugins.py

@@ -75,6 +75,30 @@ class BackgroundTitleGeneratorSpec:
     resolver_harness: str | None = None
 
 
+@dataclass(frozen=True)
+class NativeHarnessProvider:
+    """Import paths for a native harness's lifecycle hooks.
+
+    ``NativeCodingAgent`` is pure identity data; behavior lives here as a
+    sibling row keyed by the same ``key``. Every value is a dotted import path
+    (``module:attr`` or ``module.attr``) resolved lazily at dispatch time via
+    :mod:`omnigent.native_dispatch`, so building the registry never imports the
+    runner / CLI / native-harness stack. Optional hooks are ``None`` when the
+    behavior is not yet a module-level function the resolver can reach (e.g.
+    interrupt/stop handlers that are still runner closures, or the inline
+    spawn-env dispatch); those hubs migrate onto the seam in later phases.
+    """
+
+    key: str  # matches NativeCodingAgent.key
+    run_native: str  # CLI + resume launch entry point
+    auto_create_terminal: str  # runner terminal builder
+    spawn_env_builder: str | None = None
+    interrupt_handler: str | None = None
+    stop_handler: str | None = None
+    materialize_agent_spec: str | None = None  # built-in agent seeding
+    bridge_dir: str | None = None  # cost-popup bridge-dir lookup
+
+
 @dataclass(frozen=True)
 class HarnessContribution:
     """One package's harness registry contribution."""
@@ -85,6 +109,7 @@ class HarnessContribution:
     aliases: dict[str, str] = field(default_factory=dict)
     native_harnesses: frozenset[str] = frozenset()
     native_agents: tuple[NativeCodingAgent, ...] = ()
+    native_providers: tuple[NativeHarnessProvider, ...] = ()
     install_specs: dict[str, HarnessInstallSpec] = field(default_factory=dict)
     harness_install_keys: dict[str, str] = field(default_factory=dict)
     model_env_keys: dict[str, str] = field(default_factory=dict)
@@ -210,6 +235,48 @@ class HarnessPluginState:
 )
 
 
+def _builtin_native_provider(key: str) -> NativeHarnessProvider:
+    """Build a built-in provider row from the ``omnigent.<key>_native`` module.
+
+    The built-in native harnesses follow a uniform module layout: each exports
+    ``run_<key>_native`` (CLI + resume launch) and ``_materialize_<key>_agent_spec``
+    (agent seeding), and re-exports ``_auto_create_<key>_terminal`` from
+    ``omnigent.runner.native``. The remaining hooks (spawn-env, interrupt, stop,
+    bridge-dir) are still runner-local closures / inline dispatch, so they stay
+    ``None`` until those hubs migrate onto the seam.
+    """
+    module = f"omnigent.{key}_native"
+    return NativeHarnessProvider(
+        key=key,
+        run_native=f"{module}:run_{key}_native",
+        auto_create_terminal=f"omnigent.runner.native:_auto_create_{key}_terminal",
+        materialize_agent_spec=f"{module}:_materialize_{key}_agent_spec",
+    )
+
+
+# Behavior side-channel for the built-in native agents. One row per
+# NativeCodingAgent above, keyed by the same ``key``; resolved lazily so this
+# module stays import-light. See designs/harness-modular-registry-proposal.md
+# (Phase 1). Populated uniformly because every built-in native harness shares
+# the omnigent.<key>_native module layout.
+_BUILTIN_NATIVE_PROVIDERS: tuple[NativeHarnessProvider, ...] = tuple(
+    _builtin_native_provider(agent.key)
+    for agent in (
+        CLAUDE_NATIVE_CODING_AGENT,
+        CODEX_NATIVE_CODING_AGENT,
+        PI_NATIVE_CODING_AGENT,
+        OPENCODE_NATIVE_CODING_AGENT,
+        CURSOR_NATIVE_CODING_AGENT,
+        KIRO_NATIVE_CODING_AGENT,
+        GOOSE_NATIVE_CODING_AGENT,
+        ANTIGRAVITY_NATIVE_CODING_AGENT,
+        QWEN_NATIVE_CODING_AGENT,
+        KIMI_NATIVE_CODING_AGENT,
+        HERMES_NATIVE_CODING_AGENT,
+    )
+)
+
+
 # Declared capabilities for the built-in harnesses. Each value is backed by the
 # module that implements it; the derivable axes (model_family, subagents) are
 # asserted against their source in tests/test_harness_capabilities.py so the
@@ -628,6 +695,7 @@ class HarnessPluginState:
         KIMI_NATIVE_CODING_AGENT,
         HERMES_NATIVE_CODING_AGENT,
     ),
+    native_providers=_BUILTIN_NATIVE_PROVIDERS,
     model_env_keys={
         "acp": "HARNESS_ACP_MODEL",
         "antigravity": "HARNESS_ANTIGRAVITY_MODEL",
@@ -885,6 +953,22 @@ def native_agents() -> tuple[NativeCodingAgent, ...]:
     return tuple(agents)
 
 
+def native_providers() -> tuple[NativeHarnessProvider, ...]:
+    """Return native-harness behavior provider rows, merged across contributions."""
+    providers: list[NativeHarnessProvider] = []
+    for contribution in plugin_state().contributions:
+        providers.extend(contribution.native_providers)
+    return tuple(providers)
+
+
+def native_provider_for_key(key: str) -> NativeHarnessProvider | None:
+    """Return the provider row whose ``key`` matches, or ``None``."""
+    for provider in native_providers():
+        if provider.key == key:
+            return provider
+    return None
+
+
 def harness_modules() -> dict[str, str]:
     """Return runtime harness module mapping, aliases included."""
     modules = _merge_dict("harness_modules")

omnigent/native_dispatch.py

@@ -0,0 +1,63 @@
+"""Lazy resolver for native-harness provider hooks.
+
+``NativeHarnessProvider`` rows hold dotted import *strings*, never live
+callables — building the harness registry must not import the runner / CLI /
+native-harness stack (see designs/harness-plugin-interface.md § import rules).
+This module is the single place those strings become callables, and only at
+dispatch time. Each dispatch hub (resume, CLI, runner launch/interrupt/stop,
+seeding) resolves its hook here instead of branching on ``key == "<x>"``.
+
+Resolution is cached per import path so a hot dispatch loop imports each target
+module at most once.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from omnigent.harness_plugins import (
+    NativeHarnessProvider,
+    load_object,
+    native_provider_for_key,
+)
+
+_RESOLVE_CACHE: dict[str, Any] = {}
+
+
+def resolve(import_path: str) -> Any:
+    """Resolve a ``module:attr`` / ``module.attr`` path to its object, cached.
+
+    Thin caching wrapper over :func:`omnigent.harness_plugins.load_object`.
+    """
+    cached = _RESOLVE_CACHE.get(import_path)
+    if cached is None:
+        cached = load_object(import_path)
+        _RESOLVE_CACHE[import_path] = cached
+    return cached
+
+
+def resolve_hook(provider: NativeHarnessProvider, hook: str) -> Any | None:
+    """Resolve one named hook on a provider, or ``None`` if it is unset.
+
+    ``hook`` is a field name on :class:`NativeHarnessProvider` (e.g.
+    ``"run_native"``, ``"auto_create_terminal"``). Optional hooks that the
+    provider leaves ``None`` resolve to ``None`` rather than raising, so callers
+    can treat "no such hook yet" and "hook present" uniformly.
+    """
+    import_path = getattr(provider, hook)
+    if import_path is None:
+        return None
+    return resolve(import_path)
+
+
+def resolve_hook_for_key(key: str, hook: str) -> Any | None:
+    """Resolve a hook by native-agent ``key``, or ``None`` if unknown/unset."""
+    provider = native_provider_for_key(key)
+    if provider is None:
+        return None
+    return resolve_hook(provider, hook)
+
+
+def reset_resolve_cache_for_tests() -> None:
+    """Clear the per-path resolution cache."""
+    _RESOLVE_CACHE.clear()

tests/test_harness_plugins.py

@@ -252,3 +252,44 @@ def _contribution() -> hp.HarnessContribution:
 
     generator = hp.background_title_generators()["foo"]
     assert generator.generator == ("omnigent.community.harness.foo.background_titles:generate")
+
+
+def test_builtin_native_providers_cover_every_native_agent() -> None:
+    """Every native agent has exactly one provider row keyed by the same key."""
+    agent_keys = sorted(agent.key for agent in hp.native_agents())
+    provider_keys = sorted(provider.key for provider in hp.native_providers())
+    assert provider_keys == agent_keys
+    # No duplicate provider keys.
+    assert len(provider_keys) == len(set(provider_keys))
+
+
+def test_native_provider_for_key_lookup() -> None:
+    assert hp.native_provider_for_key("claude") is not None
+    assert hp.native_provider_for_key("claude").key == "claude"
+    assert hp.native_provider_for_key("does-not-exist") is None
+
+
+def test_builtin_native_providers_have_required_hooks() -> None:
+    """run_native and auto_create_terminal are mandatory on every built-in row."""
+    for provider in hp.native_providers():
+        assert provider.run_native, provider.key
+        assert provider.auto_create_terminal, provider.key
+
+
+def test_builtin_native_provider_paths_resolve() -> None:
+    """Every populated built-in provider hook resolves to a real callable.
+
+    This is the guard that keeps the provider rows honest: a typo'd import path
+    or a renamed run_<x>_native symbol fails here rather than at dispatch time.
+    """
+    from omnigent import native_dispatch
+
+    native_dispatch.reset_resolve_cache_for_tests()
+    for provider in hp.native_providers():
+        for hook in (
+            "run_native",
+            "auto_create_terminal",
+            "materialize_agent_spec",
+        ):
+            resolved = native_dispatch.resolve_hook(provider, hook)
+            assert callable(resolved), f"{provider.key}.{hook} did not resolve to a callable"

tests/test_native_dispatch.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+import pytest
+
+import omnigent.harness_plugins as hp
+from omnigent import native_dispatch
+
+
+@pytest.fixture(autouse=True)
+def _reset_caches() -> Iterator[None]:
+    hp.reset_plugin_state_for_tests()
+    native_dispatch.reset_resolve_cache_for_tests()
+    yield
+    hp.reset_plugin_state_for_tests()
+    native_dispatch.reset_resolve_cache_for_tests()
+
+
+def test_resolve_colon_and_dot_paths() -> None:
+    # module:attr form
+    assert native_dispatch.resolve("omnigent.harness_plugins:load_object") is hp.load_object
+    # module.attr form
+    assert native_dispatch.resolve("omnigent.harness_plugins.load_object") is hp.load_object
+
+
+def test_resolve_is_cached() -> None:
+    first = native_dispatch.resolve("omnigent.harness_plugins:native_agents")
+    second = native_dispatch.resolve("omnigent.harness_plugins:native_agents")
+    assert first is second
+    assert "omnigent.harness_plugins:native_agents" in native_dispatch._RESOLVE_CACHE
+
+
+def test_resolve_hook_returns_none_for_unset_optional_hook() -> None:
+    provider = hp.native_provider_for_key("claude")
+    assert provider is not None
+    # interrupt_handler is not yet a module-level function, so it stays None.
+    assert provider.interrupt_handler is None
+    assert native_dispatch.resolve_hook(provider, "interrupt_handler") is None
+
+
+def test_resolve_hook_resolves_populated_hook() -> None:
+    provider = hp.native_provider_for_key("pi")
+    assert provider is not None
+    run_native = native_dispatch.resolve_hook(provider, "run_native")
+    assert callable(run_native)
+    assert run_native.__name__ == "run_pi_native"
+
+
+def test_resolve_hook_for_key() -> None:
+    run_native = native_dispatch.resolve_hook_for_key("codex", "run_native")
+    assert callable(run_native)
+    assert run_native.__name__ == "run_codex_native"
+
+
+def test_resolve_hook_for_unknown_key_returns_none() -> None:
+    assert native_dispatch.resolve_hook_for_key("does-not-exist", "run_native") is None