chore(harnessx-removal): delete HarnessX code and packaging (task 6)

Deletes harness_ext.py and hunter_harness.py with their HarnessX-only tests (the judge path, verify_judge.py, went in task 5). The harnessx extra, its mypy override and its uv.lock entries are removed (uv lock drops only packages; no remaining version changes). The conftest fixture becomes assert_cold_import; test_gate's two never-import checks become one allow-list (stdlib, openultrasast, yaml) over the governance planes' module-level imports, and the skipped live-extra tolerance test is deleted. Audit rows for the two modules are removed, stage_processors is reworded, and the module-audit manifest is regenerated (no orphan). test_removed_plane_references implements Req 2.4 over src, tests and pyproject.toml: a match must carry a `retired 2026-..` tag (the spec's own slug is not counted). The README part is a strict xfail until task 8 rewrites the README's plane sections. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
2026-10-16 19:59:26 +00:00 · 2026-09-30 12:18:42 +02:00
parent cbadc38d5e
commit b9afb66250
16 changed files with 105 additions and 2943 deletions
@@ -44,7 +44,7 @@ and passed its fixture test. Every task is a commit gated on the full suite's ow
    tagged retirement text allowed in `src`, enforced by a test); tests renamed to the behaviour they assert.
  - _Requirements: 1.1, 1.2, 1.3, 3.3_
- [ ] 6. Delete HarnessX code and packaging
+- [x] 6. Delete HarnessX code and packaging
  - Delete `harness_ext.py`, `hunter_harness.py`, `verify_judge.py`'s judge path, the HarnessX-only tests; remove
    the extra, the mypy override and the `uv.lock` entry; rename the conftest fixture; audit rows and manifest; the
    reference search test (Req 2.4).
@@ -1,20 +1,18 @@
 {
-  "modules": 136,
+  "modules": 134,
  "counts": {
    "load_bearing": 127,
-    "standalone_capability": 9
+    "standalone_capability": 7
  },
  "orphaned": [],
  "standalone": {
    "fusion": "two-panel adjudication engine; reached from the scan's report stage and used standalone",
    "harness_ext": "HarnessX extension surface; loaded by name when the optional harness is present",
    "hunter_harness": "harness wiring for the tool hunter; used by the hunter path and by mcp",
    "mcp": "narrow MCP server over stdio; its own `ousast mcp` entry point and OpenCode integration",
    "push_inputs": "pre-push-safety-net task 1.2: offline input provenance validator; python -m openultrasast.push_inputs",
    "push_scoring": "pre-push-safety-net task 1.3: frozen-profile diagnostic outcome scorer; python -m openultrasast.push_scoring",
    "skills": "skill router; consumed by mcp and by the OpenCode integration, not by the scan path",
    "slot_contract": "the slot/contract protocol stage_processors implements",
-    "stage_processors": "HarnessX slot-contracted scan stages; pinned by the zero-dependency guard in test_gate"
+    "stage_processors": "deterministic slot-contracted scan stages; pinned by the zero-dependency guard in test_gate"
  },
  "rows": [
    {
@@ -226,7 +224,6 @@
        "fusion",
        "gate",
        "hunter",
        "hunter_harness",
        "hunter_tools",
        "improve.evolve",
        "map_gate",
@@ -272,29 +269,14 @@
        "slot_contract"
      ]
    },
    {
      "module": "harness_ext",
      "classification": "standalone_capability",
      "reason": "HarnessX extension surface; loaded by name when the optional harness is present",
      "importers": [
        "hunter_harness"
      ]
    },
    {
      "module": "hunter",
      "classification": "load_bearing",
      "reason": "reachable from gate, map_gate, pair_gate or the CLI",
      "importers": [
-        "cli",
+        "cli"
        "hunter_harness"
      ]
    },
    {
      "module": "hunter_harness",
      "classification": "standalone_capability",
      "reason": "harness wiring for the tool hunter; used by the hunter path and by mcp",
      "importers": []
    },
    {
      "module": "hunter_tools",
      "classification": "load_bearing",
@@ -849,7 +831,6 @@
        "cli",
        "findings",
        "hunter",
        "hunter_harness",
        "improve.evolve",
        "improve.validator",
        "plane.tasks.loop",
@@ -878,7 +859,6 @@
        "findings",
        "gate",
        "hunter",
        "hunter_harness",
        "hunter_tools",
        "improve.evolve",
        "index",
@@ -1064,7 +1044,6 @@
        "findings",
        "gate",
        "hunter",
        "hunter_harness",
        "hunter_tools",
        "improve.evolve",
        "map_gate",
@@ -1156,7 +1135,6 @@
        "cli",
        "findings",
        "hunter",
        "hunter_harness",
        "improve.evolve",
        "improve.memory",
        "improve.validator",
@@ -1468,8 +1446,7 @@
      "classification": "standalone_capability",
      "reason": "skill router; consumed by mcp and by the OpenCode integration, not by the scan path",
      "importers": [
-        "hunter",
+        "hunter"
        "hunter_harness"
      ]
    },
    {
@@ -1483,7 +1460,7 @@
    {
      "module": "stage_processors",
      "classification": "standalone_capability",
-      "reason": "HarnessX slot-contracted scan stages; pinned by the zero-dependency guard in test_gate",
+      "reason": "deterministic slot-contracted scan stages; pinned by the zero-dependency guard in test_gate",
      "importers": []
    },
    {
@@ -20,7 +20,7 @@ RUN apt-get update \
 WORKDIR /opt/openultrasast
 COPY pyproject.toml README.md ./
 COPY src ./src
-# Core install only: PyYAML for the manifests, no extras (no HarnessX, no tree-sitter).
+# Core install only: PyYAML for the manifests, no extras (no tree-sitter).
 RUN pip install --no-cache-dir . \
 && printf '#!/bin/sh\nexec python -m openultrasast.plane.runner "$@"\n' > /usr/local/bin/ax-task-runner \
 && chmod 0755 /usr/local/bin/ax-task-runner
@@ -11,15 +11,6 @@ requires-python = ">=3.11"
 dependencies = ["pyyaml>=6.0"]
 [project.optional-dependencies]
 # HarnessX is the optional agentic plane (LLM hunter/verifier + self-improvement).
 # SHA-pinned for reproducibility; the heavy transitive footprint (see
 # .kiro/specs/harnessx-self-improving-rulesets/dependency-audit.md) lands ONLY when
 # this extra is installed. The core install declares only PyYAML (plane manifests) and every HarnessX
 # import is guarded by openultrasast.harness_ext (lazy). pip does not pull the
 # recipe/verl_harnessX submodule (it is outside the packaged harnessx* modules).
 harnessx = [
    "harnessx @ git+https://github.com/Darwin-Agent/HarnessX.git@274c512eea67c64701f992d3b6fe06f0fd590ef0",
 ]
 # The plane's memory store on MinIO (OUSAST_MEMORY=minio://<bucket>, plane/memory.py). Imported lazily; the
 # core install and the file store need nothing beyond PyYAML.
 minio = [
@@ -69,12 +60,6 @@ warn_redundant_casts = true
 warn_return_any = true
 warn_unused_configs = true
 # HarnessX is an optional extra, absent in the type-check environment; its lazy
 # imports inside guarded functions must not fail mypy.
 [[tool.mypy.overrides]]
 module = "harnessx.*"
 ignore_missing_imports = true
 [[tool.mypy.overrides]]
 module = "tree_sitter"
 ignore_missing_imports = true
@@ -171,7 +171,7 @@ def within_tolerance(
 ) -> bool:
    """True if ``candidate`` recall/FP stay within tolerance of ``baseline``.
-    Used to confirm a HarnessX-backed stage does not drift detection beyond the
+    Used to confirm a model-backed stage does not drift detection beyond the
    configured band of the zero-dependency baseline. Recall may not fall more than
    ``recall_tolerance`` below baseline; FP may not rise more than ``fp_tolerance``
    above it.
@@ -1,97 +0,0 @@
 """Capability guard for the optional HarnessX agentic plane.
 HarnessX is an optional extra (``pip install 'openultrasast[harnessx]'``). The core
 install is zero-dependency, so nothing here imports ``harnessx`` at module load:
 :func:`has_harnessx` only probes importability via ``find_spec`` and
 :func:`require_harnessx` performs the import lazily, on demand. Every HarnessX-backed
 code path (Phase 3+: build_sast, the hunter sub-harness, the llm-judge verifier, the
 MetaAgent.evolve loop) routes through this seam.
 """
 from __future__ import annotations
 import importlib
 import importlib.util
 from types import ModuleType
 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."""
 def has_harnessx() -> bool:
    """Return True if the HarnessX extra is importable, without importing it."""
    return importlib.util.find_spec("harnessx") is not None
 def require_harnessx() -> ModuleType:
    """Lazily import and return the ``harnessx`` module, or raise with install guidance."""
    if not has_harnessx():
        raise HarnessXUnavailableError(f"HarnessX is not installed. Install the optional agentic extra: pip install '{HARNESSX_EXTRA}'")
    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).
    The single, strictly-localized HarnessX composition site (mirrors DocuHarnessX
    ``make_docgen``): every ``harnessx`` import is inside this function so importing
    this module never touches the optional extra. Raises :class:`HarnessXUnavailableError`
    when the extra is absent. The per-task step budget is enforced on the task at run
    time, not composed here.
    """
    require_harnessx()
    from harnessx.bundles.context import make_context
    from harnessx.bundles.control import make_control
    from harnessx.bundles.tools import make_tools
    from harnessx.core.builder import HarnessBuilder
    from harnessx.processors.context.strategies.system_prompt.default import DefaultSystemPromptBuilder
    builder = (
        HarnessBuilder()
        | make_context(system_builder=DefaultSystemPromptBuilder())
        | make_tools(skill_loading=False)
        | make_control(
            include_reliability=True,
            include_budget=True,
            token_threshold=token_threshold,
            max_cost_usd=max_cost_usd,
        )
    )
    return builder.build()
@@ -1,197 +0,0 @@
 """HarnessX-backed hunter pool — a real LLM agent loop per ranked target.
 A drop-in for :func:`hunter.run_hunter_pool` (same call shape, returns the same
 :class:`hunter.HunterPoolResult`) that runs the optional HarnessX agentic plane. It
 is lazy/guarded: importing this module never imports ``harnessx``; the extra is only
 touched once :meth:`HxScanOrchestrator.run_pool` runs. The agent ``.run()`` path makes
 LLM API calls and is therefore exercised only with the extra + credentials present;
 the default scan keeps using the deterministic :func:`hunter.run_hunter_pool`.
 """
 from __future__ import annotations
 import json
 import re
 from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 from .findings import SEVERITY_LABEL, StaticFinding, _reachability_conditions, _reachability_for_line
 from .harness_ext import build_provider, build_sast, require_harnessx
 from .hunter import (
    HunterPoolResult,
    HunterTask,
    HunterTrajectory,
    build_retrieval_context_by_path,
    schedule_hunter_tasks,
 )
 from .policy import CwePolicy, load_policy, resolve_severity
 from .preprocess import FileTarget
 from .rank import RankingScore
 from .ruleset import PatternRule
 # The hunter agent is asked to emit findings as a JSON array of these objects.
 _FINDINGS_JSON = re.compile(r"\[\s*\{.*\}\s*\]", re.DOTALL)
@dataclass(frozen=True)
 class ScanTask:
    """Thin description of one bounded hunter task (wraps, not subclasses, BaseTask)."""
    description: str
    success_criteria: str
    max_steps: int
    token_budget: int | None
    max_cost_usd: float | None
 class HxScanOrchestrator:
    """Runs the hunter pool on a real HarnessX agent loop, one bounded task per target."""
    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
        config = build_sast(max_cost_usd=self._max_cost_usd, token_threshold=self._token_threshold)
        return ModelConfig(main=build_provider(self._provider_model, self._provider)).agentic(config)
    def run_pool(
        self,
        root: Path,
        targets: list[FileTarget],
        rankings: list[RankingScore],
        *,
        scan_id: str,
        ruleset: tuple[PatternRule, ...] | None = None,
        policy: dict[str, CwePolicy] | None = None,
        emit: Callable[..., None] | None = None,
    ) -> HunterPoolResult:
        import asyncio
        tasks = schedule_hunter_tasks(targets, rankings, retrieval_context_by_path=build_retrieval_context_by_path(root, targets))
        resolved_policy = policy if policy is not None else load_policy()
        self._agent = self._agent or self._build_agent()
        return asyncio.run(self._run_all(scan_id, tasks, resolved_policy, emit))
    async def _run_all(
        self,
        scan_id: str,
        tasks: list[HunterTask],
        policy: dict[str, CwePolicy],
        emit: Callable[..., None] | None,
    ) -> HunterPoolResult:
        from harnessx.core.harness import BaseTask
        findings: list[StaticFinding] = []
        trajectories: list[HunterTrajectory] = []
        agent = self._agent
        for task in tasks:
            base_task = BaseTask(
                description=self._describe(task),
                success_criteria="Enumerate each reachable tainted sink with file:line and a one-line rationale.",
                max_steps=task.budget.max_findings_per_target,
                token_budget=task.budget.retrieval_budget_chars * 4,
                max_cost_usd=self._max_cost_usd,
            )
            result = await agent.run(base_task, session_id=scan_id)
            task_findings = self._findings_from_result(task, result, policy)
            findings.extend(task_findings)
            trajectories.append(self._trajectory_from_result(scan_id, task, result, task_findings))
            if emit is not None:
                emit(
                    "stage_end",
                    f"hunter_pool:{task.target.path}",
                    {
                        "exit_reason": getattr(result, "exit_reason", "unknown"),
                        "total_steps": getattr(result, "total_steps", 0),
                        "total_cost_usd": getattr(result, "total_cost_usd", 0.0),
                    },
                )
        await agent.cleanup()
        return HunterPoolResult(findings=sorted(findings, key=lambda item: item.finding_id), trajectories=trajectories)
    def _describe(self, task: HunterTask) -> str:
        from .skills import select_skill_context
        context = task.retrieval_context or f"[repo_code] {task.target.path}"
        skills = select_skill_context(task.target, stage="hunt", budget_chars=task.budget.skill_budget_chars)
        skill_block = f"\n\nRelevant security skills:\n{skills}" if skills else ""
        return (
            f"Audit `{task.target.path}` ({task.target.language}) for security-relevant vulnerabilities. "
            f"Report ONLY evidence-backed findings as a JSON array, each object: "
            f'{{"line": <int>, "title": <str>, "cwe": "CWE-NN" (optional), "rationale": <str>}}. '
            f"Emit `[]` if none.{skill_block}\n\nSource context:\n{context}"
        )
    def _findings_from_result(self, task: HunterTask, result: object, policy: dict[str, CwePolicy]) -> list[StaticFinding]:
        items = _extract_findings(getattr(result, "final_output", "") or "")
        findings: list[StaticFinding] = []
        for item in items:
            line = item.get("line")
            line = int(line) if isinstance(line, int) else None
            reachability_status, function_name, reachability_evidence = (
                _reachability_for_line(task.target, line) if line is not None else ("unknown", None, [])
            )
            cwe = str(item.get("cwe", "")) or None
            severity = SEVERITY_LABEL.get(resolve_severity(policy, cwe), "medium") if cwe else "medium"
            line_part = line if line is not None else 0
            findings.append(
                StaticFinding(
                    finding_id=f"hx-hunter:{task.target.path}:{line_part}",
                    path=task.target.path,
                    title=str(item.get("title", "HarnessX hunter finding")),
                    severity=severity,
                    confidence="medium",
                    evidence_level="static_corroboration",
                    rationale=str(item.get("rationale", "Reported by the HarnessX hunter; manual verification required.")),
                    line=line,
                    function_name=function_name,
                    reachability_status=reachability_status,
                    reachability_evidence=reachability_evidence,
                    reachability_conditions=_reachability_conditions(reachability_evidence),
                    tags=sorted(set(task.target.tags)),
                    ranking_priority=task.ranking.priority,
                )
            )
        return findings
    def _trajectory_from_result(
        self, scan_id: str, task: HunterTask, result: object, task_findings: list[StaticFinding]
    ) -> HunterTrajectory:
        return HunterTrajectory(
            scan_id=scan_id,
            target_path=task.target.path,
            tier=task.tier,
            priority=task.ranking.priority,
            selected_skills=task.selected_skills,
            retrieval_context_chars=len(task.retrieval_context),
            findings=[finding.finding_id for finding in task_findings],
            status=str(getattr(result, "exit_reason", "completed")),
        )
 def _extract_findings(text: str) -> list[dict[str, object]]:
    """Best-effort extraction of the findings JSON array from the agent's final output."""
    match = _FINDINGS_JSON.search(text)
    if match is None:
        return []
    try:
        parsed = json.loads(match.group(0))
    except json.JSONDecodeError:
        return []
    return [item for item in parsed if isinstance(item, dict)] if isinstance(parsed, list) else []
@@ -77,9 +77,7 @@ STANDALONE = {
    "mcp": "narrow MCP server over stdio; its own `ousast mcp` entry point and OpenCode integration",
    "skills": "skill router; consumed by mcp and by the OpenCode integration, not by the scan path",
    "fusion": "two-panel adjudication engine; reached from the scan's report stage and used standalone",
-    "harness_ext": "HarnessX extension surface; loaded by name when the optional harness is present",
+    "stage_processors": "deterministic slot-contracted scan stages; pinned by the zero-dependency guard in test_gate",
    "hunter_harness": "harness wiring for the tool hunter; used by the hunter path and by mcp",
    "stage_processors": "HarnessX slot-contracted scan stages; pinned by the zero-dependency guard in test_gate",
    "slot_contract": "the slot/contract protocol stage_processors implements",
 }
@@ -5,12 +5,6 @@ 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(autouse=True)
 def _pointer_pairs_offline(monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.TempPathFactory) -> None:
@@ -31,17 +25,16 @@ def _repos_offline(monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.Tem
@pytest.fixture
-def assert_cold_of_harnessx():  # type: ignore[no-untyped-def]
+def assert_cold_import():  # type: ignore[no-untyped-def]
-    """Run ``code`` in a fresh interpreter and assert it imported no ``harnessx`` module.
+    """Run ``code`` in a fresh interpreter and assert it exits 0.
-    Hermetic by construction: the lazy-import guarantee is verified the same way
+    Hermetic by construction: an import-laziness check (``code`` asserts on ``sys.modules`` itself) is not
-    whether or not the optional extra is installed in this environment (a global
+    polluted by modules other tests have already imported into this process.
    ``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],
+            [sys.executable, "-c", code],
            capture_output=True,
            text=True,
        )
@@ -2,15 +2,15 @@
 8.1 — the recall/FP gate is a hard constraint a score improvement cannot unblock.
 8.2 — default-ruleset benchmark output is byte-identical to a committed baseline.
-8.3 — a HarnessX-backed stage must stay within tolerance of the zero-dep baseline.
+8.3 — the tolerance band a changed stage is held to against the zero-dep baseline.
-8.4 — the governance/scoring/benchmark planes never import the HarnessX extra.
+8.4 — the governance/scoring/benchmark planes import only stdlib, the package and PyYAML at module scope.
 """
 import ast
 import json
 import sys
 from pathlib import Path
 import pytest
 from openultrasast.gate import (
    FALSE_POSITIVE_CEILING,
    RECALL_FLOOR,
@@ -20,7 +20,6 @@ from openultrasast.gate import (
    merge_gate,
    within_tolerance,
 )
 from openultrasast.harness_ext import has_harnessx
 GOLDEN = Path("tests/golden/detection_baseline.json")
@@ -108,7 +107,7 @@ def test_benchmark_output_is_byte_identical_to_golden_baseline() -> None:
    )
-# ---- 8.3 HarnessX-backed stage stays within detection tolerance -------------
+# ---- 8.3 detection tolerance band -------------------------------------------
 def test_within_tolerance_band_logic() -> None:
@@ -121,15 +120,6 @@ def test_within_tolerance_band_logic() -> None:
    assert not within_tolerance(base, DetectionMetrics(100, 93, 100, 10), recall_tolerance=0.05, fp_tolerance=0.05)
@pytest.mark.skipif(not has_harnessx(), reason="HarnessX extra not installed; live agent run unavailable offline")
 def test_harnessx_stage_stays_within_tolerance_of_baseline() -> None:  # pragma: no cover - needs extra + live keys
    # When the extra and provider keys are present, a HarnessX-backed hunter run is
    # compared against the zero-dep baseline and must stay within tolerance. Live
    # LLM execution is unavailable in CI, so this gates on the capability.
    baseline = detection_gate(evaluate_corpus()).overall
    assert within_tolerance(baseline, baseline)
 # ---- 8.4 zero-dependency guard ----------------------------------------------
 GOVERNANCE_SCORING_BENCHMARK_PLANES = (
@@ -152,24 +142,33 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = (
 )
-def test_governance_scoring_benchmark_planes_never_import_harnessx(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
+_ALLOWED_TOP_LEVEL = frozenset(sys.stdlib_module_names) | {"openultrasast", "yaml"}
    # Hermetic: a fresh interpreter imports every plane and must pull no harnessx.
    imports = "\n".join(f"import {module}" for module in GOVERNANCE_SCORING_BENCHMARK_PLANES)
    assert_cold_of_harnessx(imports)
-def test_planes_have_no_module_level_harnessx_import() -> None:
+def _module_level_imports(tree: ast.Module) -> list[str]:
-    # Flags only column-0 (module-level) imports; an indented lazy import inside a
+    """Top-level names imported outside any function or class body (a lazy import inside a function is fine)."""
-    # capability-guarded function (e.g. stage_processors.host_under_harnessx) is fine.
+    names: list[str] = []
    pending: list[ast.AST] = list(tree.body)
    while pending:
        node = pending.pop()
        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef):
            continue
        if isinstance(node, ast.Import):
            names.extend(alias.name.split(".")[0] for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            names.append("openultrasast" if node.level else (node.module or "").split(".")[0])
        else:
            pending.extend(ast.iter_child_nodes(node))
    return names
 def test_governance_planes_import_only_stdlib_and_declared_dependencies() -> None:
    # An allow-list, not a deny-list: any optional extra at module scope fails without being named here.
    src = Path("src/openultrasast")
    offenders = []
    for module in GOVERNANCE_SCORING_BENCHMARK_PLANES:
        rel = module.removeprefix("openultrasast.").replace(".", "/")
-        candidates = [src / f"{rel}.py", src / rel / "__init__.py"]
+        path = next((c for c in (src / f"{rel}.py", src / rel / "__init__.py") if c.exists()), None)
-        path = next((c for c in candidates if c.exists()), None)
+        assert path is not None, f"{module} vanished from the guard's list"
-        if path is None:
+        offenders += [f"{module}: {name}" for name in _module_level_imports(ast.parse(path.read_text())) if name not in _ALLOWED_TOP_LEVEL]
-            continue
+    assert offenders == [], f"governance planes import beyond stdlib, openultrasast and yaml at module scope: {offenders}"
        for line in path.read_text().splitlines():
            if line.startswith(("import harnessx", "from harnessx")):
                offenders.append(f"{path}: {line}")
    assert offenders == [], f"plane modules import HarnessX at module scope: {offenders}"
@@ -1,73 +0,0 @@
 """HarnessX optional-extra packaging + capability guard (Phase 3 task 5.2)."""
 import re
 import tomllib
 from pathlib import Path
 import pytest
 from openultrasast.harness_ext import (
    HARNESSX_EXTRA,
    SUPPORTED_PROVIDERS,
    HarnessXUnavailableError,
    build_provider,
    has_harnessx,
    require_harnessx,
 )
 PYPROJECT = tomllib.loads(Path("pyproject.toml").read_text())
 def test_core_dependencies_are_pyyaml_only() -> None:
    assert PYPROJECT["project"]["dependencies"] == ["pyyaml>=6.0"]  # ai-service-plane task 1: manifests are YAML; HarnessX stays an extra
 def test_harnessx_is_a_sha_pinned_optional_extra() -> None:
    extras = PYPROJECT["project"]["optional-dependencies"]["harnessx"]
    assert len(extras) == 1
    assert re.fullmatch(
        r"harnessx @ git\+https://github\.com/Darwin-Agent/HarnessX\.git@[0-9a-f]{40}",
        extras[0],
    ), extras[0]
 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.
    assert_cold_of_harnessx("import openultrasast.cli\nimport openultrasast.harness_ext")
 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:
    if has_harnessx():
        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
@@ -1,44 +0,0 @@
 """HarnessX hunter sub-harness: guarding, findings parsing, construction smoke (task 6.1)."""
 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(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:
    text = 'Here are the findings:\n[{"line": 9, "title": "SQLi", "cwe": "CWE-89", "rationale": "concat"}]\nDone.'
    items = _extract_findings(text)
    assert items == [{"line": 9, "title": "SQLi", "cwe": "CWE-89", "rationale": "concat"}]
    assert _extract_findings("no findings here") == []
    assert _extract_findings("[not, valid, json]") == []
@pytest.mark.skipif(has_harnessx(), reason="exercises the extra-absent guard")
 def test_build_sast_raises_without_extra() -> None:
    with pytest.raises(HarnessXUnavailableError):
        build_sast()
@pytest.mark.skipif(has_harnessx(), reason="exercises the extra-absent guard")
 def test_orchestrator_construction_raises_without_extra() -> None:
    with pytest.raises(HarnessXUnavailableError):
        HxScanOrchestrator(provider_model="claude-sonnet-4-6")
@pytest.mark.skipif(not has_harnessx(), reason="construction smoke; requires the harnessx extra")
 def test_build_sast_returns_harness_config_offline() -> None:
    from harnessx.core.harness import HarnessConfig
    assert isinstance(build_sast(), HarnessConfig)
@pytest.mark.skipif(not has_harnessx(), reason="construction smoke; requires the harnessx extra")
 def test_orchestrator_builds_agent_offline() -> None:
    # Constructs the model + harness without any LLM call (provider client is lazy).
    agent = HxScanOrchestrator(provider_model="claude-sonnet-4-6")._build_agent()
    assert type(agent).__name__ == "Harness"
@@ -25,13 +25,15 @@ def test_every_module_is_classified() -> None:
 def test_the_named_subsystems_each_carry_an_explicit_decision() -> None:
-    """Req 3.3: the five older-phase subsystems get a decision, not silence."""
+    """Req 3.3: the older-phase subsystems still in the tree get a decision, not silence."""
    from openultrasast.model.audit import audit
    rows = {row.module: row for row in audit(SRC)}
-    for name in ("fusion", "mcp", "skills", "harness_ext", "hunter_harness"):
+    for name in ("fusion", "mcp", "skills"):
        assert name in rows, f"{name} vanished without a recorded decision"
        assert rows[name].reason, f"{name} has no stated reason"
    # The removed agentic plane's modules are gone from the tree, not carried as rows (harnessx-removal Req 2.1).
    assert not {"harness_ext", "hunter_harness", "verify_judge"} & set(rows)
 def test_the_manifest_matches_the_tree() -> None:
@@ -0,0 +1,54 @@
 """Requirement 2.4 of the removal spec: the removed agentic plane is named only in its retirement notes.
 A case-insensitive search of ``src/``, ``tests/``, ``pyproject.toml`` and the README may match only lines tagged
 ``retired 2026-..``: the retired-key table and the one warning in ``config.py``, the retirement tests, and the README's
 plane section. The spec's own slug (its directory name, the equality instrument's file name) is a citation of the
 removal, not a use of the plane, and is not counted. The pattern is built at runtime so this file does not match it.
 """
 from __future__ import annotations
 import re
 from pathlib import Path
 import pytest
 ROOT = Path(__file__).resolve().parents[1]
 _NAME = "harness" + "x"
 _PATTERN = re.compile(_NAME, re.IGNORECASE)
 _SPEC_SLUG = re.compile(_NAME + r"[-_]removal", re.IGNORECASE)
 _TAG = re.compile(r"retired 2026-\d\d-\d\d", re.IGNORECASE)
 def _untagged(paths: list[Path]) -> list[str]:
    offenders = []
    for path in paths:
        try:
            text = path.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError):
            continue
        for number, line in enumerate(text.splitlines(), 1):
            if _PATTERN.search(_SPEC_SLUG.sub("", line)) and not _TAG.search(line):
                offenders.append(f"{path.relative_to(ROOT)}:{number}: {line.strip()[:120]}")
    return offenders
 def _tree(*roots: str) -> list[Path]:
    return sorted(p for root in roots for p in (ROOT / root).rglob("*") if p.is_file() and "__pycache__" not in p.parts)
 def test_no_harnessx_outside_retirement_notes() -> None:  # the Req 2.4 check itself, retired 2026-09-30
    offenders = _untagged([*_tree("src", "tests"), ROOT / "pyproject.toml"])
    assert offenders == [], "the removed plane is named outside a tagged retirement note:\n" + "\n".join(offenders)
 def test_the_retirement_notes_exist() -> None:
    """The search is not vacuous: the one warning and the retired keys are where the design puts them."""
    config = (ROOT / "src/openultrasast/config.py").read_text()
    assert _PATTERN.search(config) and _TAG.search(config)
@pytest.mark.xfail(strict=True, reason="the README's plane sections are rewritten in task 8; drop this mark there")
 def test_the_readme_names_the_removed_plane_only_in_its_retirement_note() -> None:
    offenders = _untagged([ROOT / "README.md"])
    assert offenders == [], "\n".join(offenders)
@@ -25,14 +25,14 @@ def test_core_dependencies_are_pyyaml_only_and_semantic_is_optional() -> None:
    assert "semantic:" in " ".join(payload["tool"]["pytest"]["ini_options"]["markers"])
-def test_cli_import_does_not_load_tree_sitter(assert_cold_of_harnessx) -> None:  # type: ignore[no-untyped-def]
+def test_cli_import_does_not_load_tree_sitter(assert_cold_import) -> None:  # type: ignore[no-untyped-def]
    proc_code = (
        "import sys\n"
        "import openultrasast.cli\n"
        "leaked = sorted(m for m in sys.modules if m == 'tree_sitter' or m.startswith('tree_sitter_'))\n"
        "assert not leaked, leaked\n"
    )
-    assert_cold_of_harnessx(proc_code)
+    assert_cold_import(proc_code)
 def test_probe_off_yields_no_javascript_grammar(monkeypatch: pytest.MonkeyPatch) -> None: