feat(harnessx): configurable provider + per-task budget caps

Add a [harnessx] config block so the agentic plane can be steered from
openultrasast.toml, and make the LLM provider selectable (the hunter and verifier
previously hardcoded AnthropicProvider).

- config.HarnessxConfig: provider ("anthropic"|"openai"|"litellm", default
  anthropic), max_cost_usd (2.0), token_threshold (120000); parsed from [harnessx].
- harness_ext.build_provider: single lazy, capability-guarded provider factory.
  The provider name is validated before any import, so an unknown choice fails with
  a clear ValueError even when the extra is absent; each provider reads its own
  standard key env var (openai -> OPENAI_API_KEY, anthropic -> ANTHROPIC_API_KEY).
- HxScanOrchestrator and verify_findings_dispatch take a provider and route through
  build_provider; cli threads config.harnessx.{provider,max_cost_usd,token_threshold}
  into both the hunter and the verifier.
- README: "Configuring the HarnessX agentic plane" section (install, models,
  provider, budgets, fallback/degradation).

Test isolation: convert the "harnessx not in sys.modules" import-guards to a
hermetic subprocess fixture (tests/conftest.py assert_cold_of_harnessx) and
simulate extra-absent in the CLI dispatch tests, so the suite is deterministic
whether or not the optional extra is installed (it now is, locally). Verified the
OpenAI path constructs end-to-end (build_provider -> OpenAIProvider; orchestrator
-> Harness) with the budget threaded through.

137 passed, 3 skipped; ruff + mypy clean; detection gate PASS.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

README.md

@@ -243,6 +243,43 @@ addressed, it stays closed because the recall/precision gate runs in CI.
 `external_baseline_deltas.json` additionally shows where another tool found a
 vulnerability OpenUltraSAST missed (or vice-versa).
 
+## Configuring the HarnessX agentic plane
+
+The LLM hunter pool and the llm-judge verifier are optional. The core install is
+zero-dependency and `--mode quick` never calls a model. To enable the agentic
+plane, install the extra, name the models, and pick a provider:
+
+```bash
+uv sync --extra harnessx          # SHA-pinned optional dependency
+export OPENAI_API_KEY=sk-...       # or ANTHROPIC_API_KEY, matching the provider
+uv run ousast scan /path/to/code --mode standard --config openultrasast.toml
+```
+
+```toml
+# openultrasast.toml
+[models]
+hunter   = "gpt-4o"   # LLM hunter pool (empty -> deterministic hunter)
+verifier = "gpt-4o"   # llm-judge verifier (empty -> structural verifier)
+
+[harnessx]
+provider        = "openai"   # "anthropic" (default) | "openai" | "litellm"
+max_cost_usd    = 2.0        # per-task spend cap
+token_threshold = 120000     # per-task token budget
+```
+
+The provider selects the HarnessX backend for **both** the hunter and the
+verifier; each reads its own standard key env var (`openai` → `OPENAI_API_KEY`,
+`anthropic` → `ANTHROPIC_API_KEY`). The HarnessX path activates only when the
+extra is installed, a model is configured, and the mode is `standard` — all three
+gated through one capability seam (`harness_ext.has_harnessx`). If any condition
+fails, the scan transparently falls back to the deterministic hunter / structural
+verifier and records a `degradations` entry in `manifest.json`, so a run is never
+silently downgraded:
+
+```bash
+jq '.degradations' "$run/manifest.json"   # null = HarnessX ran; entries = fell back
+```
+
 ## The HarnessX self-improving cycle
 
 Each scan is a composed harness of typed processors with read/write **state

src/openultrasast/cli.py

@@ -169,7 +169,12 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     write_ranking_calibrations(calibrations, run.root / "calibration" / "applied_calibrations.json")
     if mode == "standard":
         if hx_hunter and hunter_model:
-            orchestrator = HxScanOrchestrator(provider_model=hunter_model)
+            orchestrator = HxScanOrchestrator(
+                provider_model=hunter_model,
+                provider=config.harnessx.provider,
+                max_cost_usd=config.harnessx.max_cost_usd,
+                token_threshold=config.harnessx.token_threshold,
+            )
             hunter_result = runtime.run_stage(
                 "hunter_pool",
                 lambda: orchestrator.run_pool(
@@ -222,7 +227,12 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
         )
     verifications = runtime.run_stage(
         "verify",
-        lambda: verify_findings_dispatch(findings, verifier_model=verifier_model, use_harnessx=hx_verify),
+        lambda: verify_findings_dispatch(
+            findings,
+            verifier_model=verifier_model,
+            verifier_provider=config.harnessx.provider,
+            use_harnessx=hx_verify,
+        ),
     )
     write_verification_results(verifications, verification_path)
     runtime.run_stage(

src/openultrasast/config.py

@@ -66,6 +66,17 @@ class RulesetConfig:
     min_emit_precision: float = 0.0
 
 
+@dataclass(frozen=True)
+class HarnessxConfig:
+    # Optional agentic plane (the openultrasast[harnessx] extra). `provider`
+    # selects the HarnessX LLM provider used by both the hunter pool and the
+    # llm-judge verifier; the budget caps bound per-task cost and token spend.
+    # Inert unless the extra is installed and a model is configured in [models].
+    provider: str = "anthropic"
+    max_cost_usd: float = 2.0
+    token_threshold: int = 120_000
+
+
 @dataclass(frozen=True)
 class ResolvedConfig:
     models: ModelConfig = ModelConfig()
@@ -76,6 +87,7 @@ class ResolvedConfig:
     static_analysis: StaticAnalysisConfig = StaticAnalysisConfig()
     score: ScoreConfig = ScoreConfig()
     ruleset: RulesetConfig = RulesetConfig()
+    harnessx: HarnessxConfig = HarnessxConfig()
     runs_dir: str = ".openultrasast/runs"
 
 
@@ -94,6 +106,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig:
         static_analysis=_load_static_analysis(data.get("static_analysis", {})),
         score=_load_score(data.get("score", {})),
         ruleset=_load_ruleset_config(data.get("ruleset", {})),
+        harnessx=_load_harnessx(data.get("harnessx", {})),
         runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"),
     )
 
@@ -188,6 +201,15 @@ def _load_ruleset_config(value: object) -> RulesetConfig:
     )
 
 
+def _load_harnessx(value: object) -> HarnessxConfig:
+    data = _section(value)
+    return HarnessxConfig(
+        provider=_string(data.get("provider")) or "anthropic",
+        max_cost_usd=_float_value(data.get("max_cost_usd"), 2.0),
+        token_threshold=_int_value(data.get("token_threshold"), 120_000),
+    )
+
+
 def _float_value(value: object, default: float) -> float:
     if isinstance(value, bool):
         return default

src/openultrasast/harness_ext.py

@@ -13,13 +13,17 @@ from __future__ import annotations
 import importlib
 import importlib.util
 from types import ModuleType
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from harnessx.core.harness import HarnessConfig
 
 HARNESSX_EXTRA = "openultrasast[harnessx]"
 
+# HarnessX providers OpenUltraSAST can drive for the hunter / llm-judge stages.
+# Each reads its own standard API-key env var (ANTHROPIC_API_KEY / OPENAI_API_KEY).
+SUPPORTED_PROVIDERS = ("anthropic", "openai", "litellm")
+
 
 class HarnessXUnavailableError(RuntimeError):
     """Raised when a HarnessX-backed capability is requested but the extra is absent."""
@@ -37,6 +41,32 @@ def require_harnessx() -> ModuleType:
     return importlib.import_module("harnessx")
 
 
+def build_provider(model: str, provider: str = "anthropic") -> Any:
+    """Build a HarnessX LLM provider for ``model`` (lazy, capability-guarded).
+
+    The single place that knows HarnessX provider class names. The provider name is
+    validated before any import so an unknown choice fails loudly even when the extra
+    is absent; each provider reads its own standard API-key env var. Raises
+    :class:`ValueError` on an unknown provider and :class:`HarnessXUnavailableError`
+    when the extra is not installed.
+    """
+    key = provider.lower()
+    if key not in SUPPORTED_PROVIDERS:
+        raise ValueError(f"unknown HarnessX provider {provider!r}; choose one of {SUPPORTED_PROVIDERS}")
+    require_harnessx()
+    if key == "anthropic":
+        from harnessx.providers.anthropic_provider import AnthropicProvider
+
+        return AnthropicProvider(model=model)
+    if key == "openai":
+        from harnessx.providers.openai_provider import OpenAIProvider
+
+        return OpenAIProvider(model=model)
+    from harnessx.providers.litellm_provider import LiteLLMProvider
+
+    return LiteLLMProvider(model=model)
+
+
 def build_sast(*, max_cost_usd: float = 2.0, token_threshold: int = 120_000) -> HarnessConfig:
     """Compose the SAST hunter HarnessConfig (behaviour pipeline, model-free).
 

src/openultrasast/hunter_harness.py

@@ -18,7 +18,7 @@ from pathlib import Path
 from typing import Any
 
 from .findings import SEVERITY_LABEL, StaticFinding, _reachability_conditions, _reachability_for_line
-from .harness_ext import build_sast, require_harnessx
+from .harness_ext import build_provider, build_sast, require_harnessx
 from .hunter import (
     HunterPoolResult,
     HunterTask,
@@ -49,19 +49,26 @@ class ScanTask:
 class HxScanOrchestrator:
     """Runs the hunter pool on a real HarnessX agent loop, one bounded task per target."""
 
-    def __init__(self, *, provider_model: str, max_cost_usd: float = 2.0, token_threshold: int = 120_000) -> None:
+    def __init__(
+        self,
+        *,
+        provider_model: str,
+        provider: str = "anthropic",
+        max_cost_usd: float = 2.0,
+        token_threshold: int = 120_000,
+    ) -> None:
         require_harnessx()
         self._provider_model = provider_model
+        self._provider = provider
         self._max_cost_usd = max_cost_usd
         self._token_threshold = token_threshold
         self._agent: Any = None
 
     def _build_agent(self) -> Any:
         from harnessx.core.model_config import ModelConfig
-        from harnessx.providers.anthropic_provider import AnthropicProvider
 
         config = build_sast(max_cost_usd=self._max_cost_usd, token_threshold=self._token_threshold)
-        return ModelConfig(main=AnthropicProvider(model=self._provider_model)).agentic(config)
+        return ModelConfig(main=build_provider(self._provider_model, self._provider)).agentic(config)
 
     def run_pool(
         self,

src/openultrasast/verify_judge.py

@@ -8,7 +8,7 @@ so behaviour is byte-identical without HarnessX. The judge path is lazy/guarded.
 from __future__ import annotations
 
 from .findings import StaticFinding
-from .harness_ext import has_harnessx
+from .harness_ext import build_provider, has_harnessx
 from .verification import (
     EvidenceLevel,
     VerificationResult,
@@ -23,6 +23,7 @@ def verify_findings_dispatch(
     findings: list[StaticFinding],
     *,
     verifier_model: str | None = None,
+    verifier_provider: str = "anthropic",
     use_harnessx: bool = False,
 ) -> list[VerificationResult]:
     """Drop-in superset of :func:`verification.verify_findings`.
@@ -32,21 +33,20 @@ def verify_findings_dispatch(
     """
     if not (use_harnessx and verifier_model and has_harnessx()):
         return verify_findings(findings)
-    return _verify_with_judge(findings, verifier_model)
+    return _verify_with_judge(findings, verifier_model, verifier_provider)
 
 
-def _verify_with_judge(findings: list[StaticFinding], verifier_model: str) -> list[VerificationResult]:
+def _verify_with_judge(findings: list[StaticFinding], verifier_model: str, verifier_provider: str) -> list[VerificationResult]:
     import asyncio
 
-    return asyncio.run(_judge_all(findings, verifier_model))
+    return asyncio.run(_judge_all(findings, verifier_model, verifier_provider))
 
 
-async def _judge_all(findings: list[StaticFinding], verifier_model: str) -> list[VerificationResult]:
+async def _judge_all(findings: list[StaticFinding], verifier_model: str, verifier_provider: str) -> list[VerificationResult]:
     from harnessx.core.events import Message
     from harnessx.processors.evaluation.llm_judge import build_judge_prompt, parse_judge_response
-    from harnessx.providers.anthropic_provider import AnthropicProvider
 
-    provider = AnthropicProvider(model=verifier_model)
+    provider = build_provider(verifier_model, verifier_provider)
     results: list[VerificationResult] = []
     for finding in findings:
         prompt = build_judge_prompt(

tests/conftest.py

@@ -0,0 +1,33 @@
+"""Shared test fixtures."""
+
+import subprocess
+import sys
+
+import pytest
+
+_HARNESSX_GUARD = (
+    "\nimport sys as _sys\n"
+    "_leaked = sorted(m for m in _sys.modules if m == 'harnessx' or m.startswith('harnessx.'))\n"
+    "assert not _leaked, 'leaked harnessx modules: ' + repr(_leaked)\n"
+)
+
+
+@pytest.fixture
+def assert_cold_of_harnessx():  # type: ignore[no-untyped-def]
+    """Run ``code`` in a fresh interpreter and assert it imported no ``harnessx`` module.
+
+    Hermetic by construction: the lazy-import guarantee is verified the same way
+    whether or not the optional extra is installed in this environment (a global
+    ``sys.modules`` check would be polluted by other tests once the extra is present).
+    """
+
+    def _check(code: str) -> str:
+        proc = subprocess.run(
+            [sys.executable, "-c", code + _HARNESSX_GUARD],
+            capture_output=True,
+            text=True,
+        )
+        assert proc.returncode == 0, f"exit={proc.returncode}\nstdout={proc.stdout}\nstderr={proc.stderr}"
+        return proc.stdout
+
+    return _check

tests/test_cli_hx_dispatch.py

@@ -1,8 +1,10 @@
 """CLI capability-gated dispatch + graceful degradation (Phase 3 task 6.5).
 
-These run in the default (extra-absent) environment: standard mode with models
-configured must fall back to the deterministic hunter/structural verifier and
-record a degradation; with no models configured the manifest stays byte-identical.
+Simulates the extra-absent environment (via a monkeypatched capability probe so the
+behaviour is deterministic whether or not the optional extra is installed): standard
+mode with models configured must fall back to the deterministic hunter/structural
+verifier and record a degradation; with no models configured the manifest stays
+byte-identical.
 """
 
 import json
@@ -20,6 +22,7 @@ def _run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config_body: str | Non
     repo = tmp_path / "repo"
     repo.mkdir()
     (repo / "app.py").write_text(_VULN)
+    monkeypatch.setattr(cli, "has_harnessx", lambda: False)  # simulate the extra being absent
     monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
     argv = ["scan", str(repo), "--mode", "standard"]
     if config_body is not None:
@@ -32,7 +35,6 @@ def _run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config_body: str | Non
 
 
 def test_degradation_recorded_when_models_set_but_extra_absent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-    assert not cli.has_harnessx()  # this CI env has no harnessx extra
     manifest = _run(tmp_path, monkeypatch, '[models]\nhunter = "anthropic/claude"\nverifier = "anthropic/claude"\n')
 
     degradations = manifest.get("degradations", [])

tests/test_config.py

@@ -28,6 +28,33 @@ def test_load_config_reads_toml(tmp_path: Path) -> None:
     assert config.dynamic.network_scope == ("127.0.0.1:8080",)
 
 
+def test_harnessx_defaults_when_section_absent() -> None:
+    config = load_config(None)
+    assert config.harnessx.provider == "anthropic"
+    assert config.harnessx.max_cost_usd == 2.0
+    assert config.harnessx.token_threshold == 120_000
+
+
+def test_load_config_reads_harnessx_section(tmp_path: Path) -> None:
+    config_path = tmp_path / "openultrasast.toml"
+    config_path.write_text(
+        "\n".join(
+            [
+                "[harnessx]",
+                'provider = "openai"',
+                "max_cost_usd = 5.0",
+                "token_threshold = 200000",
+            ]
+        )
+    )
+
+    config = load_config(config_path)
+
+    assert config.harnessx.provider == "openai"
+    assert config.harnessx.max_cost_usd == 5.0
+    assert config.harnessx.token_threshold == 200_000
+
+
 def test_write_resolved_config_creates_json(tmp_path: Path) -> None:
     output = tmp_path / "run" / "resolved_config.json"
 

tests/test_harness_ext.py

@@ -1,13 +1,19 @@
 """HarnessX optional-extra packaging + capability guard (Phase 3 task 5.2)."""
 
 import re
-import sys
 import tomllib
 from pathlib import Path
 
 import pytest
 
-from openultrasast.harness_ext import HARNESSX_EXTRA, HarnessXUnavailableError, has_harnessx, require_harnessx
+from openultrasast.harness_ext import (
+    HARNESSX_EXTRA,
+    SUPPORTED_PROVIDERS,
+    HarnessXUnavailableError,
+    build_provider,
+    has_harnessx,
+    require_harnessx,
+)
 
 PYPROJECT = tomllib.loads(Path("pyproject.toml").read_text())
 
@@ -25,18 +31,15 @@ def test_harnessx_is_a_sha_pinned_optional_extra() -> None:
     ), extras[0]
 
 
-def test_importing_openultrasast_does_not_import_harnessx() -> None:
+def test_importing_openultrasast_does_not_import_harnessx(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
     # The big modules (cli, harness_ext) must not eagerly import the heavy extra.
-    import openultrasast.cli  # noqa: F401
-    import openultrasast.harness_ext  # noqa: F401
-
-    assert "harnessx" not in sys.modules
+    assert_cold_of_harnessx("import openultrasast.cli\nimport openultrasast.harness_ext")
 
 
-def test_has_harnessx_probes_without_importing() -> None:
-    available = has_harnessx()
-    assert isinstance(available, bool)
-    assert "harnessx" not in sys.modules  # find_spec must not import the module
+def test_has_harnessx_probes_without_importing(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
+    assert isinstance(has_harnessx(), bool)
+    # find_spec must not import the module.
+    assert_cold_of_harnessx("from openultrasast.harness_ext import has_harnessx\nassert isinstance(has_harnessx(), bool)")
 
 
 def test_require_harnessx_raises_with_install_guidance_when_absent() -> None:
@@ -44,3 +47,27 @@ def test_require_harnessx_raises_with_install_guidance_when_absent() -> None:
         pytest.skip("HarnessX extra is installed in this environment")
     with pytest.raises(HarnessXUnavailableError, match=re.escape(HARNESSX_EXTRA)):
         require_harnessx()
+
+
+def test_build_provider_rejects_unknown_provider_before_importing(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
+    # Name is validated before any import, so this fails loudly even without the extra.
+    with pytest.raises(ValueError, match="unknown HarnessX provider"):
+        build_provider("gpt-4o", "bogus")
+    assert_cold_of_harnessx(
+        "from openultrasast.harness_ext import build_provider\n"
+        "try:\n    build_provider('gpt-4o', 'bogus')\n"
+        "except ValueError:\n    pass\n"
+        "else:\n    raise SystemExit('expected ValueError')\n"
+    )
+
+
+def test_supported_providers_cover_anthropic_and_openai() -> None:
+    assert {"anthropic", "openai"} <= set(SUPPORTED_PROVIDERS)
+
+
+@pytest.mark.parametrize(("provider", "class_name"), [("anthropic", "AnthropicProvider"), ("openai", "OpenAIProvider")])
+def test_build_provider_constructs_selected_provider(provider: str, class_name: str) -> None:
+    if not has_harnessx():
+        pytest.skip("HarnessX extra not installed in this environment")
+    built = build_provider("test-model", provider)
+    assert type(built).__name__ == class_name

tests/test_hunter_harness.py

@@ -1,17 +1,13 @@
 """HarnessX hunter sub-harness: guarding, findings parsing, construction smoke (task 6.1)."""
 
-import sys
-
 import pytest
 
 from openultrasast.harness_ext import HarnessXUnavailableError, build_sast, has_harnessx
 from openultrasast.hunter_harness import HxScanOrchestrator, _extract_findings
 
 
-def test_module_import_does_not_pull_harnessx() -> None:
-    import openultrasast.hunter_harness  # noqa: F401
-
-    assert "harnessx" not in sys.modules
+def test_module_import_does_not_pull_harnessx(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
+    assert_cold_of_harnessx("import openultrasast.hunter_harness")
 
 
 def test_extract_findings_parses_json_array_and_tolerates_garbage() -> None:

tests/test_verify_judge.py

@@ -1,7 +1,5 @@
 """Verifier dispatcher: structural fallback (default) + judge verdict mapping (Phase 3 task 6.4)."""
 
-import sys
-
 from openultrasast.findings import StaticFinding
 from openultrasast.verification import VerificationStatus, verify_findings
 from openultrasast.verify_judge import _verdict_to_verification, verify_findings_dispatch
@@ -40,9 +38,20 @@ def test_dispatch_stays_structural_when_model_unset_or_extra_absent() -> None:
     assert verify_findings_dispatch(findings, verifier_model="claude-x", use_harnessx=False) == verify_findings(findings)
 
 
-def test_dispatch_does_not_import_harnessx_on_fallback() -> None:
-    verify_findings_dispatch([_finding()], verifier_model="claude-x", use_harnessx=True)
-    assert "harnessx" not in sys.modules
+def test_dispatch_does_not_import_harnessx_on_fallback(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
+    # Even with a model and use_harnessx=True, when the capability is absent the
+    # dispatch must fall back without importing the extra. Simulate absence so the
+    # guarantee holds whether or not the extra is installed in this environment.
+    assert_cold_of_harnessx(
+        "import openultrasast.verify_judge as vj\n"
+        "vj.has_harnessx = lambda: False\n"
+        "from openultrasast.findings import StaticFinding\n"
+        "f = StaticFinding(finding_id='x:app.py:1', path='app.py', title='t', severity='low',\n"
+        "    confidence='low', evidence_level='static_corroboration', rationale='r', line=1,\n"
+        "    function_name=None, reachability_status='unknown', reachability_evidence=[],\n"
+        "    reachability_conditions=[], tags=[], ranking_priority=1.0)\n"
+        "vj.verify_findings_dispatch([f], verifier_model='claude-x', use_harnessx=True)\n"
+    )
 
 
 def test_verdict_mapping_is_evidence_gated() -> None: