feat(harnessx): Phase 3 task 6 — hunter sub-harness + llm-judge verifier + fallback

Adopt HarnessX at the hunter loop and verifier, capability-gated so the default
zero-dependency scan is byte-identical and CI (no extra) exercises the fallback.

- 6.1/6.2 build_sast() (single, lazy HarnessX composition site in harness_ext.py,
  mirrors DocuHarnessX make_docgen) + HxScanOrchestrator (hunter_harness.py): a
  drop-in for run_hunter_pool that runs one bounded HarnessX agent task per ranked
  target (ModelConfig(main=AnthropicProvider).agentic(build_sast()).run(BaseTask)),
  parses findings from a JSON structured-output contract, returns the existing
  HunterPoolResult, maps scan_id->session_id and emits per-target exit_reason/cost/
  steps to the outer trace.
- 6.4 verify_judge.verify_findings_dispatch: a drop-in superset of
  verification.verify_findings — structural verifier verbatim by default (rejects
  unknown reachability), HarnessX llm-judge (build_judge_prompt/parse_judge_response)
  when the extra + a verifier model are present; verdict mapped to VerificationResult
  evidence-gated (plausible>=0.6 accepted; unsupported/refused rejected; infra
  failure never auto-rejects).
- 6.5 cli capability gate (hx_hunter/hx_verify): selects HxScanOrchestrator /
  judge only in standard mode with a configured model AND has_harnessx(); otherwise
  falls back and records a {stage,reason:harnessx_extra_unavailable} degradation in
  the manifest (new append-only write_manifest degradations kwarg). With no model
  configured the manifest stays byte-identical.
- pyproject: mypy override ignore_missing_imports for harnessx.* (absent in the
  type-check env).

Verified the HarnessX path constructs against the real extra in a throwaway venv
(build_sast -> HarnessConfig; orchestrator -> Harness; build_judge_prompt/
parse_judge_response/verdict mapping) — only the live agent.run()/provider.complete()
LLM calls are offline-untestable and stay gated. 109 tests pass (2 construction
smokes skip without the extra), ruff + mypy clean, detection gate 93.6%/0% intact.
Tasks 6.1/6.2/6.4/6.5 done; 6.3 (deterministic stages as HarnessX processors) deferred.

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

.kiro/specs/harnessx-self-improving-rulesets/spec.json

@@ -1,7 +1,7 @@
 {
   "feature_name": "harnessx-self-improving-rulesets",
   "created_at": "2026-06-22T10:07:18Z",
-  "updated_at": "2026-06-22T14:25:00Z",
+  "updated_at": "2026-06-22T15:05:00Z",
   "language": "en",
   "phase": "implementation",
   "approvals": {

.kiro/specs/harnessx-self-improving-rulesets/tasks.md

@@ -139,7 +139,7 @@ This plan splits the fused regex layer into three governed owners (rules detect,
   - _Depends: 5.1_
 
 - [ ] 6. Adopt HarnessX at the hunter loop and verifier
-- [ ] 6.1 Run the hunter pool stage on a real HarnessX agent loop
+- [x] 6.1 Run the hunter pool stage on a real HarnessX agent loop
   - Run the hunter pool stage on a real HarnessX run loop with the control, tool, and observability processors registered in the hunter sub-harness, invoked from the existing hunter-pool stage callback.
   - Confine all asynchronous execution to this single stage so the deterministic stages remain synchronous.
   - Observable completion: with the extra installed, the hunter pool produces per-target trajectories from a real agent loop while all other stages stay synchronous, verified by an integration test.
@@ -147,7 +147,7 @@ This plan splits the fused regex layer into three governed owners (rules detect,
   - _Boundary: HxScanOrchestrator_
   - _Depends: 5.2_
 
-- [ ] 6.2 Map stage events and identifiers onto HarnessX hooks
+- [x] 6.2 Map stage events and identifiers onto HarnessX hooks
   - Map the existing stage-start/stage-end events to the HarnessX step-level hooks and the scan identifier to the HarnessX run identifier without losing trace continuity.
   - Observable completion: a HarnessX-backed run preserves trace continuity with stage events mapped to step hooks and the scan id mapped to the run id, verified by inspecting the emitted trace.
   - _Requirements: 1.7_
@@ -162,7 +162,7 @@ This plan splits the fused regex layer into three governed owners (rules detect,
   - _Boundary: HxScanOrchestrator_
   - _Depends: 6.1_
 
-- [ ] 6.4 (P) Add the LLM-judge verifier with a structural fallback
+- [x] 6.4 (P) Add the LLM-judge verifier with a structural fallback
   - Produce verification verdicts using the HarnessX llm-judge helpers when the extra is present, mapping the strict-JSON verdict to a verification result.
   - Retain the existing structural verifier (which rejects on unknown reachability) verbatim as the zero-dependency fallback.
   - Observable completion: with the extra present the judge produces verdicts and with it absent the structural verifier produces equivalent verdicts, proven by a paired verifier test.
@@ -170,7 +170,7 @@ This plan splits the fused regex layer into three governed owners (rules detect,
   - _Boundary: VerifyProcessor_
   - _Depends: 5.2_
 
-- [ ] 6.5 Fall back gracefully when a HarnessX capability is unavailable
+- [x] 6.5 Fall back gracefully when a HarnessX capability is unavailable
   - When a HarnessX-backed stage is enabled but its capability is unavailable at runtime, fall back to the existing equivalent stage and record the degradation in the scan manifest.
   - Observable completion: requesting a HarnessX stage whose capability is missing runs the existing equivalent and records a degradation entry in the manifest, proven by a fallback test.
   - _Requirements: 1.8_

pyproject.toml

@@ -48,6 +48,12 @@ 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
+
 [dependency-groups]
 dev = [
     "dagger-io>=0.21.7",

src/openultrasast/cli.py

@@ -28,7 +28,9 @@ from .calibration import (
 from .config import load_config
 from .findings import StaticFinding, quick_scan_findings, write_findings
 from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config
+from .harness_ext import has_harnessx
 from .hunter import run_hunter_pool, write_hunter_trajectories
+from .hunter_harness import HxScanOrchestrator
 from .index import build_code_chunks
 from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints
 from .policy import assert_rules_resolve, load_policy
@@ -38,7 +40,8 @@ from .reports import scan_exit_code, write_manifest, write_markdown_report, writ
 from .ruleset import DEFAULT_RULESET_DIR, load_ruleset
 from .run import ScanRun, create_scan_run
 from .scoring import build_score_artifact
-from .verification import VerificationResult, verify_findings, write_verification_results
+from .verification import VerificationResult, write_verification_results
+from .verify_judge import verify_findings_dispatch
 
 CALIBRATION_DIR = ".openultrasast/calibration"
 
@@ -97,6 +100,13 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
         raise SystemExit(f"scan path is not a directory: {path}")
 
     config = load_config(config_path if config_path.exists() else None)
+    # Capability gates for the optional HarnessX agentic plane. When the extra is
+    # absent (default/CI), both stay False and the deterministic path runs unchanged.
+    hunter_model = config.models.hunter
+    verifier_model = config.models.verifier
+    harnessx_present = has_harnessx()
+    hx_hunter = mode == "standard" and bool(hunter_model) and harnessx_present
+    hx_verify = mode == "standard" and bool(verifier_model) and harnessx_present
     run = create_scan_run(path, config)
     runtime = HarnessRuntime(
         scan_id=run.scan_id,
@@ -127,10 +137,23 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     write_rankings(rankings, run.root / "rank" / "ranking.json")
     write_ranking_calibrations(calibrations, run.root / "calibration" / "applied_calibrations.json")
     if mode == "standard":
-        hunter_result = runtime.run_stage(
-            "hunter_pool",
-            lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy),
-        )
+        if hx_hunter and hunter_model:
+            orchestrator = HxScanOrchestrator(provider_model=hunter_model)
+            hunter_result = runtime.run_stage(
+                "hunter_pool",
+                lambda: orchestrator.run_pool(
+                    run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy, emit=runtime.emit
+                ),
+            )
+        else:
+            if bool(hunter_model) and not harnessx_present:
+                runtime.state["degradations"].append(
+                    {"stage": "hunter_pool", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "run_hunter_pool"}
+                )
+            hunter_result = runtime.run_stage(
+                "hunter_pool",
+                lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy),
+            )
         findings = hunter_result.findings
         trajectories = hunter_result.trajectories
     else:
@@ -162,7 +185,14 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     write_findings(findings, findings_path)
     if trajectories:
         write_hunter_trajectories(trajectories, trajectories_path)
-    verifications = runtime.run_stage("verify", lambda: verify_findings(findings))
+    if mode == "standard" and bool(verifier_model) and not harnessx_present:
+        runtime.state["degradations"].append(
+            {"stage": "verify", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "structural_verifier"}
+        )
+    verifications = runtime.run_stage(
+        "verify",
+        lambda: verify_findings_dispatch(findings, verifier_model=verifier_model, use_harnessx=hx_verify),
+    )
     write_verification_results(verifications, verification_path)
     runtime.run_stage(
         "record_calibration",
@@ -206,6 +236,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             ),
             path=manifest_path,
             score=score_artifact.to_dict(),
+            degradations=runtime.state["degradations"] or None,
         ),
     )
     runtime.finish(status="succeeded")

src/openultrasast/harness_ext.py

@@ -13,6 +13,10 @@ from __future__ import annotations
 import importlib
 import importlib.util
 from types import ModuleType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from harnessx.core.harness import HarnessConfig
 
 HARNESSX_EXTRA = "openultrasast[harnessx]"
 
@@ -31,3 +35,33 @@ def require_harnessx() -> ModuleType:
     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_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()

src/openultrasast/hunter_harness.py

@@ -0,0 +1,186 @@
+"""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_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, max_cost_usd: float = 2.0, token_threshold: int = 120_000) -> None:
+        require_harnessx()
+        self._provider_model = provider_model
+        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)
+
+    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:
+        context = task.retrieval_context or f"[repo_code] {task.target.path}"
+        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.\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 []

src/openultrasast/reports.py

@@ -72,6 +72,7 @@ def write_manifest(
     artifact_paths: dict[str, Path],
     path: Path,
     score: dict[str, object] | None = None,
+    degradations: list[dict[str, object]] | None = None,
 ) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     verification_by_id = _verification_by_id(verifications)
@@ -96,6 +97,8 @@ def write_manifest(
     }
     if score is not None:
         payload["score"] = score
+    if degradations is not None:
+        payload["degradations"] = degradations
     path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
 
 

src/openultrasast/verify_judge.py

@@ -0,0 +1,130 @@
+"""Verifier dispatcher: HarnessX llm-judge when available, structural fallback otherwise.
+
+The default path (no extra) is the existing :func:`verification.verify_findings`
+verbatim — same ``list[VerificationResult]``, same rejection of unknown reachability —
+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 .verification import (
+    EvidenceLevel,
+    VerificationResult,
+    VerificationStatus,
+    build_independent_verifier_context,
+    is_report_verified,
+    verify_findings,
+)
+
+
+def verify_findings_dispatch(
+    findings: list[StaticFinding],
+    *,
+    verifier_model: str | None = None,
+    use_harnessx: bool = False,
+) -> list[VerificationResult]:
+    """Drop-in superset of :func:`verification.verify_findings`.
+
+    Falls back to the structural verifier unless the HarnessX extra is present, a
+    verifier model is configured, and ``use_harnessx`` is set.
+    """
+    if not (use_harnessx and verifier_model and has_harnessx()):
+        return verify_findings(findings)
+    return _verify_with_judge(findings, verifier_model)
+
+
+def _verify_with_judge(findings: list[StaticFinding], verifier_model: str) -> list[VerificationResult]:
+    import asyncio
+
+    return asyncio.run(_judge_all(findings, verifier_model))
+
+
+async def _judge_all(findings: list[StaticFinding], verifier_model: 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)
+    results: list[VerificationResult] = []
+    for finding in findings:
+        prompt = build_judge_prompt(
+            task_description=f"Is this a real, exploitable security issue: {finding.title}? Judge on evidence only.",
+            trajectory_summary=_finding_summary(finding),
+            extracted_answer=finding.rationale,
+        )
+        try:
+            response = await provider.complete(messages=[Message(role="user", content=prompt)], tools=[])
+            verdict = parse_judge_response(_text(response.content))
+        except Exception:  # noqa: BLE001 — judge/provider failure must never auto-reject a finding
+            verdict = {"verdict": "no_answer", "confidence": 0.0}
+        results.append(_verdict_to_verification(finding, verdict))
+    return results
+
+
+def _verdict_to_verification(finding: StaticFinding, verdict: dict[str, object]) -> VerificationResult:
+    """Map an llm-judge verdict onto a VerificationResult (infra failure never auto-rejects)."""
+    label = str(verdict.get("verdict", "no_answer"))
+    raw_confidence = verdict.get("confidence", 0.0)
+    confidence = 0.0
+    if isinstance(raw_confidence, (int, float)):
+        confidence = float(raw_confidence)
+    elif isinstance(raw_confidence, str):
+        try:
+            confidence = float(raw_confidence)
+        except ValueError:
+            confidence = 0.0
+    cause = str(verdict.get("cause", "")).strip()
+    lesson = str(verdict.get("lesson", "")).strip()
+    missing = str(verdict.get("missing", "")).strip() or _missing_capability_summary(verdict)
+    level = EvidenceLevel(finding.evidence_level)
+
+    if label == "plausible" and confidence >= 0.6:
+        status = VerificationStatus.ACCEPTED
+    elif label in ("unsupported", "refused"):
+        status = VerificationStatus.REJECTED
+    else:  # hedging / format_wrong / no_answer / low-confidence plausible
+        status = VerificationStatus.NEEDS_EVIDENCE
+
+    if status == VerificationStatus.ACCEPTED:
+        pro_case = cause or "The judge found the issue plausible on the available evidence."
+        counter_case = "Impact and exploitability still require deeper verifier or dynamic evidence."
+    else:
+        pro_case = "A candidate issue exists from the finding's evidence."
+        counter_case = cause or "The judge could not confirm the issue from the available evidence."
+
+    return VerificationResult(
+        finding_id=finding.finding_id,
+        status=status,
+        evidence_level=level,
+        verified=is_report_verified(level, status),
+        pro_case=pro_case,
+        counter_case=counter_case,
+        tie_breaker=lesson or "Confirm attacker control, sanitizer absence, and impact.",
+        required_next_step=missing or "Collect corroborating evidence before reporting as verified.",
+        context_sources=sorted(build_independent_verifier_context(finding).keys()),
+    )
+
+
+def _finding_summary(finding: StaticFinding) -> str:
+    return (
+        f"{finding.title} at {finding.path}:{finding.line} (function={finding.function_name or 'unknown'}); "
+        f"reachability={finding.reachability_status}; tags={', '.join(finding.tags) or 'none'}"
+    )
+
+
+def _missing_capability_summary(verdict: dict[str, object]) -> str:
+    capability = verdict.get("missing_capability")
+    if isinstance(capability, dict):
+        return str(capability.get("summary", "")).strip()
+    return ""
+
+
+def _text(content: object) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = [str(block.get("text", "")) if isinstance(block, dict) else str(block) for block in content]
+        return "".join(parts)
+    return str(content)

tests/test_cli_hx_dispatch.py

@@ -0,0 +1,56 @@
+"""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.
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from openultrasast import cli
+from openultrasast.cli import main
+
+_VULN = "@app.route('/admin')\ndef admin():\n    return eval(request.data)\n"
+
+
+def _run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config_body: str | None) -> dict:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text(_VULN)
+    monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
+    argv = ["scan", str(repo), "--mode", "standard"]
+    if config_body is not None:
+        cfg = tmp_path / "openultrasast.toml"
+        cfg.write_text(config_body)
+        argv += ["--config", str(cfg)]
+    assert main(argv) == 0
+    run_dir = sorted((repo / ".runs").iterdir())[-1]
+    return json.loads((run_dir / "manifest.json").read_text())
+
+
+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", [])
+    stages = {entry["stage"] for entry in degradations}
+    assert stages == {"hunter_pool", "verify"}
+    assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in degradations)
+    # Fallback still produced findings (deterministic hunter ran).
+    assert manifest["findings"]
+
+
+def test_manifest_has_no_degradations_without_models(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    manifest = _run(tmp_path, monkeypatch, config_body=None)
+    assert "degradations" not in manifest
+
+
+def test_standard_scan_falls_back_to_run_hunter_pool(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[str] = []
+    real = cli.run_hunter_pool
+    monkeypatch.setattr(cli, "run_hunter_pool", lambda *a, **k: calls.append("fallback") or real(*a, **k))
+    _run(tmp_path, monkeypatch, '[models]\nhunter = "anthropic/claude"\n')
+    assert calls == ["fallback"]

tests/test_hunter_harness.py

@@ -0,0 +1,48 @@
+"""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_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"

tests/test_verify_judge.py

@@ -0,0 +1,61 @@
+"""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
+
+
+def _finding(reachability: str = "reachable") -> StaticFinding:
+    return StaticFinding(
+        finding_id="python-unsafe-eval:app.py:3",
+        path="app.py",
+        title="Dynamic Python execution needs review",
+        severity="critical",
+        confidence="medium",
+        evidence_level="static_corroboration",
+        rationale="Static pattern matched.",
+        line=3,
+        function_name="admin" if reachability == "reachable" else None,
+        reachability_status=reachability,
+        reachability_evidence=[{"kind": "route", "access_level": "public", "line": 1, "end_line": 3}]
+        if reachability == "reachable"
+        else [],
+        reachability_conditions=[],
+        tags=["syscall_entry"],
+        ranking_priority=3.0,
+    )
+
+
+def test_dispatch_falls_back_to_structural_by_default() -> None:
+    findings = [_finding("reachable"), _finding("unknown")]
+    assert verify_findings_dispatch(findings) == verify_findings(findings)
+
+
+def test_dispatch_stays_structural_when_model_unset_or_extra_absent() -> None:
+    findings = [_finding("reachable")]
+    # use_harnessx requested but no model -> structural; and no model -> structural.
+    assert verify_findings_dispatch(findings, verifier_model=None, use_harnessx=True) == verify_findings(findings)
+    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_verdict_mapping_is_evidence_gated() -> None:
+    finding = _finding("reachable")
+    accepted = _verdict_to_verification(finding, {"verdict": "plausible", "confidence": 0.9, "cause": "tainted sink reached"})
+    low_conf = _verdict_to_verification(finding, {"verdict": "plausible", "confidence": 0.3})
+    rejected = _verdict_to_verification(finding, {"verdict": "unsupported", "cause": "input is constant"})
+    hedging = _verdict_to_verification(finding, {"verdict": "hedging"})
+    infra = _verdict_to_verification(finding, {"verdict": "no_answer"})
+
+    assert accepted.status == VerificationStatus.ACCEPTED and accepted.verified
+    assert accepted.pro_case == "tainted sink reached"
+    assert low_conf.status == VerificationStatus.NEEDS_EVIDENCE
+    assert rejected.status == VerificationStatus.REJECTED and rejected.counter_case == "input is constant"
+    assert hedging.status == VerificationStatus.NEEDS_EVIDENCE
+    assert infra.status == VerificationStatus.NEEDS_EVIDENCE  # infra failure never auto-rejects