From cbadc38d5ed857aa653431957f7f4ba380ac034a Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 30 Sep 2026 12:13:26 +0200 Subject: [PATCH] refactor(harnessx-removal): retire the HarnessX scan paths (task 5) The standard scan now runs run_hunter_pool, verify_findings and the deterministic fuse_findings directly; the HarnessX hunter orchestrator, the LLM judge dispatcher (verify_judge.py, deleted with its test) and the LLM fusion panels are gone, and so are their extra-absent degradations. host_under_harnessx is dropped from stage_processors. config: [models] verifier and [fusion] panel_model/decider_model raise RetiredConfigError naming the plane replacement (the CLI prints it and exits 2); a [harnessx] section loads with one warning and is otherwise ignored. harness.json keeps "verifier": null. Tests that asserted the fallback are renamed to the behaviour they assert; the module-audit manifest drops verify_judge. Nothing outside the two modules deleted in task 6 imports harness_ext or hunter_harness. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- .kiro/specs/harnessx-removal/tasks.md | 2 +- .../measurements/2026-09-08-module-audit.json | 28 +--- src/openultrasast/cli.py | 80 ++-------- src/openultrasast/config.py | 70 +++++---- src/openultrasast/fusion.py | 146 ++---------------- src/openultrasast/harness.py | 2 +- src/openultrasast/slot_contract.py | 9 +- src/openultrasast/stage_processors.py | 39 +---- src/openultrasast/verify_judge.py | 130 ---------------- tests/test_cli_hx_dispatch.py | 65 -------- tests/test_config.py | 74 +++++---- tests/test_fusion.py | 13 +- tests/test_scan_deep.py | 1 - tests/test_scan_pipeline_integrity.py | 63 +++----- tests/test_slot_contract.py | 27 +--- tests/test_tool_hunter.py | 6 +- tests/test_verify_judge.py | 70 --------- 17 files changed, 152 insertions(+), 673 deletions(-) delete mode 100644 src/openultrasast/verify_judge.py delete mode 100644 tests/test_cli_hx_dispatch.py delete mode 100644 tests/test_verify_judge.py diff --git a/.kiro/specs/harnessx-removal/tasks.md b/.kiro/specs/harnessx-removal/tasks.md index c4ed7a6..a3f93c3 100644 --- a/.kiro/specs/harnessx-removal/tasks.md +++ b/.kiro/specs/harnessx-removal/tasks.md @@ -38,7 +38,7 @@ and passed its fixture test. Every task is a commit gated on the full suite's ow passes it reuses from memory). - _Requirements: 6.3_ -- [ ] 5. Retire the HarnessX scan paths +- [x] 5. Retire the HarnessX scan paths - `cli.py`, `fusion.py`, `verify_judge.py`, `stage_processors.py` changes; `RetiredConfigError` for `[models] verifier`, `[fusion] panel_model`/`decider_model`; one warning for a `[harnessx]` section (the only tagged retirement text allowed in `src`, enforced by a test); tests renamed to the behaviour they assert. diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index cc4ed93..055b47e 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 137, + "modules": 136, "counts": { - "load_bearing": 128, + "load_bearing": 127, "standalone_capability": 9 }, "orphaned": [], @@ -243,8 +243,7 @@ "semantic.prove_filter", "stage_processors", "tool_hunter", - "verification", - "verify_judge" + "verification" ] }, { @@ -278,11 +277,7 @@ "classification": "standalone_capability", "reason": "HarnessX extension surface; loaded by name when the optional harness is present", "importers": [ - "cli", - "fusion", - "hunter_harness", - "stage_processors", - "verify_judge" + "hunter_harness" ] }, { @@ -298,9 +293,7 @@ "module": "hunter_harness", "classification": "standalone_capability", "reason": "harness wiring for the tool hunter; used by the hunter path and by mcp", - "importers": [ - "cli" - ] + "importers": [] }, { "module": "hunter_tools", @@ -1519,16 +1512,7 @@ "calibration", "cli", "fusion", - "reports", - "verify_judge" - ] - }, - { - "module": "verify_judge", - "classification": "load_bearing", - "reason": "reachable from gate, map_gate, pair_gate or the CLI", - "importers": [ - "cli" + "reports" ] } ] diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 3c7e21a..d8dc1a4 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -36,14 +36,12 @@ from .calibration import ( ) from .complexity import map as complexity_map from .complexity.ledger import persist_verdicts -from .config import ModelLayerConfig, ObligationsConfig, ResolvedConfig, load_config, load_dotenv +from .config import ModelLayerConfig, ObligationsConfig, ResolvedConfig, RetiredConfigError, load_config, load_dotenv from .findings import StaticFinding, quick_scan_findings, write_findings -from .fusion import FusionDecision, fuse_findings_dispatch +from .fusion import FusionDecision, fuse_findings from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR 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 .improve import RoundOutcome, run_improvement from .index import build_code_chunks from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints @@ -91,7 +89,6 @@ from .semantic import ( from .semantic.facts import FactLoadError, SemanticFacts from .stages import Stage, plan_for_mode, record_completed, record_skip, skip_as_degradation, stages_payload from .verification import VerificationResult, verify_findings, write_verification_results -from .verify_judge import verify_findings_dispatch CALIBRATION_DIR = ".openultrasast/calibration" @@ -125,6 +122,14 @@ def _configure_logging() -> None: def main(argv: list[str] | None = None) -> int: + try: + return _main(argv) + except RetiredConfigError as exc: # a retired key is an error that names its replacement, exit 2 + print(f"ousast: {exc}", file=sys.stderr) + return 2 + + +def _main(argv: list[str] | None) -> int: if (argv if argv is not None else sys.argv[1:])[:1] != ["pre-push"]: load_dotenv() parser = argparse.ArgumentParser(prog="ousast") @@ -431,13 +436,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut plan = plan_for_mode(mode) 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, @@ -469,28 +468,10 @@ 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": - if hx_hunter and 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( - 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), - ) + 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: @@ -530,46 +511,19 @@ 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) - 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, - verifier_provider=config.harnessx.provider, - use_harnessx=hx_verify, - ), - ) + verifications = runtime.run_stage("verify", lambda: verify_findings(findings)) write_verification_results(verifications, verification_path) runtime.run_stage( "record_calibration", lambda: _persist_calibration_feedback(run, ledger_path, prior_learnings, findings, verifications), ) - # Fusion: two-panel adjudication for triggered findings (standard mode). Runs - # deterministically by default; routes panels through the configured provider when - # a panel model is set and the extra is present, else falls back + records a degradation. + # Fusion: deterministic two-panel adjudication for triggered findings (standard mode). fusion_decisions: list[FusionDecision] = [] fusion_path = run.root / "fusion.json" if mode == "standard" and config.fusion.enabled: - hx_fusion = bool(config.fusion.panel_model) and harnessx_present - if bool(config.fusion.panel_model) and not harnessx_present: - runtime.state["degradations"].append( - {"stage": "fusion", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "deterministic_panels"} - ) fusion_decisions = runtime.run_stage( "fusion", - lambda: fuse_findings_dispatch( - findings, - verifications, - panel_model=config.fusion.panel_model, - decider_model=config.fusion.decider_model, - provider=config.harnessx.provider, - use_harnessx=hx_fusion, - high_assurance=config.fusion.high_assurance, - ), + lambda: fuse_findings(findings, verifications, high_assurance=config.fusion.high_assurance), ) if fusion_decisions: fusion_path.write_text(json.dumps([decision.to_dict() for decision in fusion_decisions], indent=2, sort_keys=True) + "\n") diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index 7864b16..ad86781 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import logging import os import tomllib from dataclasses import asdict, dataclass @@ -11,6 +12,41 @@ from openultrasast.contracts import Contract DEFAULT_VECTOR_STORE = "json-local" +# Retirement notes (harnessx-removal Req 1.3, 3.3). A key that asked for a retired LLM capability fails and names +# the replacement, never a silent downgrade; the old budget section only warns. Every line below that names the +# removed plane carries its retirement date, which the reference-search test holds the rest of the tree to. +_RETIRED_ON = "retired 2026-09-30 with HarnessX" +_PLANE_RUN = "`ousast plane run`, see ops/ax/README.md" +_RETIRED_PANELS = ( + f"LLM fusion panels were retired; fusion is deterministic; independent LLM agreement is the plane's `agree` task ({_PLANE_RUN})" +) +RETIRED_KEYS: dict[tuple[str, str], str] = { + ("models", "verifier"): f"LLM verification runs on the plane (`verify` + `agree`, {_PLANE_RUN})", + ("fusion", "panel_model"): _RETIRED_PANELS, + ("fusion", "decider_model"): _RETIRED_PANELS, +} +RETIRED_SECTION = "harnessx" # retired 2026-09-30: loads with one warning and is otherwise ignored (Req 3.3) +RETIRED_SECTION_WARNING = ( + "`[harnessx]` is ignored (retired 2026-09-30): HarnessX was removed; agentic work runs on the ax plane " + "with per-task budgets, see ops/ax/README.md" +) + + +class RetiredConfigError(ValueError): + """A config key asked for a capability that was retired; the message names its replacement.""" + + +def _check_retired(data: dict[str, object]) -> None: + problems = [] + for (section, key), replacement in RETIRED_KEYS.items(): + table = data.get(section) + if isinstance(table, dict) and key in table: + problems.append(f"`[{section}] {key}` was {_RETIRED_ON}: {replacement}. Remove the key.") + if problems: + raise RetiredConfigError("\n".join(problems)) + if RETIRED_SECTION in data: + logging.getLogger("openultrasast.config").warning(RETIRED_SECTION_WARNING) + def load_dotenv(path: Path | None = None, *, force: bool = False) -> None: """Load KEY=VALUE lines from .env into os.environ. Stdlib only; never overrides.""" @@ -34,7 +70,6 @@ def load_dotenv(path: Path | None = None, *, force: bool = False) -> None: class ModelConfig: ranker: str | None = None hunter: str | None = None - verifier: str | None = None patcher: str | None = None judge: str | None = None # second, independent judgement before anything is published as proven chat_base_url: str | None = None # chat endpoint; independent of the embedding endpoint @@ -95,17 +130,6 @@ 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 HardeningConfig: # Release-readiness controls. redact_secrets masks credentials in traces/reports; @@ -116,12 +140,9 @@ class HardeningConfig: @dataclass(frozen=True) class FusionConfig: - # Two-panel fusion adjudication for triggered findings (standard mode). Runs - # deterministically by default; setting panel_model routes the panels through - # the configured HarnessX provider. high_assurance forces fusion on every finding. + # Two-panel fusion adjudication for triggered findings (standard mode). Deterministic + # panels and reconciler; high_assurance forces fusion on every finding. enabled: bool = True - panel_model: str | None = None - decider_model: str | None = None high_assurance: bool = False @@ -198,7 +219,6 @@ class ResolvedConfig: static_analysis: StaticAnalysisConfig = StaticAnalysisConfig() score: ScoreConfig = ScoreConfig() ruleset: RulesetConfig = RulesetConfig() - harnessx: HarnessxConfig = HarnessxConfig() fusion: FusionConfig = FusionConfig() hardening: HardeningConfig = HardeningConfig() complexity: ComplexityConfig = ComplexityConfig() @@ -217,6 +237,7 @@ def load_config(config_path: Path | None = None, *, dotenv: bool = True) -> Reso if config_path is not None and config_path.exists(): with config_path.open("rb") as handle: data = tomllib.load(handle) + _check_retired(data) return ResolvedConfig( models=_load_models(data.get("models", {})), @@ -227,7 +248,6 @@ def load_config(config_path: Path | None = None, *, dotenv: bool = True) -> Reso 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", {})), fusion=_load_fusion(data.get("fusion", {})), hardening=_load_hardening(data.get("hardening", {})), complexity=_load_complexity(data.get("complexity", {})), @@ -263,7 +283,6 @@ def _load_models(value: object) -> ModelConfig: return ModelConfig( ranker=_string(data.get("ranker")), hunter=_string(data.get("hunter")), - verifier=_string(data.get("verifier")), patcher=_string(data.get("patcher")), judge=_string(data.get("judge")), chat_base_url=_string(data.get("chat_base_url")), @@ -342,15 +361,6 @@ 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 _load_hardening(value: object) -> HardeningConfig: data = _section(value) return HardeningConfig( @@ -363,8 +373,6 @@ def _load_fusion(value: object) -> FusionConfig: data = _section(value) return FusionConfig( enabled=bool(data.get("enabled", True)), - panel_model=_string(data.get("panel_model")), - decider_model=_string(data.get("decider_model")), high_assurance=bool(data.get("high_assurance", False)), ) diff --git a/src/openultrasast/fusion.py b/src/openultrasast/fusion.py index 40b4cdd..acd7f7d 100644 --- a/src/openultrasast/fusion.py +++ b/src/openultrasast/fusion.py @@ -7,11 +7,9 @@ disclosure, or an explicit high-assurance request. Two panels independently revi bounded evidence (one steel-manning the vulnerability case, one the false-positive case), critique and revise, vote, and a decider issues the final disposition. -The deterministic engine here is the zero-dependency core (and the fallback). When the -``openultrasast[harnessx]`` extra and panel models are configured, ``fuse_findings_dispatch`` -routes the panels through real LLM reviews; the decider reconciliation stays deterministic -and auditable. Every fused finding receives exactly one of the five dispositions, and the -decision discloses model IDs, panel roles, votes, degradations, and the decision source. +The panels and the decider are deterministic and auditable. Every fused finding receives +exactly one of the five dispositions, and the decision discloses panel roles, votes and the +decision source. Independent LLM agreement is the plane's ``agree`` task, not this module. """ from __future__ import annotations @@ -21,7 +19,6 @@ from dataclasses import asdict, dataclass, field from enum import StrEnum from .findings import StaticFinding -from .harness_ext import has_harnessx from .verification import VerificationResult, VerificationStatus @@ -57,7 +54,7 @@ class FusionDecision: triggered: bool triggers: list[str] disposition: FusionDisposition - decision_source: str # "deterministic-reconciler" | "llm-panels" | "no-fusion" + decision_source: str # "deterministic-reconciler" | "no-fusion" panels: list[PanelVerdict] votes: dict[str, int] model_ids: list[str] @@ -191,155 +188,30 @@ def fuse_finding( ) -def fuse_findings_dispatch( +def fuse_findings( findings: list[StaticFinding], verifications: list[VerificationResult], *, - panel_model: str | None = None, - decider_model: str | None = None, - provider: str = "anthropic", - use_harnessx: bool = False, high_assurance: bool = False, mitigated_ids: frozenset[str] = frozenset(), ) -> list[FusionDecision]: - """Fuse every finding that the trigger policy selects. - - Deterministic by default. When ``use_harnessx`` is set, a panel model is configured, - and the extra is present, the panels are produced by real LLM reviews; otherwise the - deterministic panels run and (if an LLM run was requested) a degradation is recorded. - """ + """Fuse every finding that the trigger policy selects, with the deterministic panels.""" verification_by_id = {result.finding_id: result for result in verifications} - want_llm = use_harnessx and bool(panel_model) decisions: list[FusionDecision] = [] for finding in findings: verification = verification_by_id.get(finding.finding_id) triggers = fusion_triggers(finding, verification, high_assurance=high_assurance) - if not triggers: - continue - if want_llm and has_harnessx(): - decisions.append(_fuse_with_llm_panels(finding, verification, triggers, panel_model, decider_model, provider, mitigated_ids)) - else: - decision = fuse_finding(finding, verification, triggers, mitigated_ids=mitigated_ids) - if want_llm and not has_harnessx(): - decision = _with_degradation(decision, "fusion_llm_unavailable: harnessx extra absent; used deterministic panels") - decisions.append(decision) + if triggers: + decisions.append(fuse_finding(finding, verification, triggers, mitigated_ids=mitigated_ids)) return decisions -def _with_degradation(decision: FusionDecision, note: str) -> FusionDecision: - return FusionDecision( - finding_id=decision.finding_id, - triggered=decision.triggered, - triggers=decision.triggers, - disposition=decision.disposition, - decision_source=decision.decision_source, - panels=decision.panels, - votes=decision.votes, - model_ids=decision.model_ids, - degradations=[*decision.degradations, note], - warnings=decision.warnings, - ) - - -def _fuse_with_llm_panels( - finding: StaticFinding, - verification: VerificationResult | None, - triggers: list[str], - panel_model: str | None, - decider_model: str | None, - provider: str, - mitigated_ids: frozenset[str], -) -> FusionDecision: - """LLM-backed panels with a deterministic, auditable decider (lazy/guarded).""" - import asyncio - - model = panel_model or "" - try: - panels = asyncio.run(_run_llm_panels(finding, verification, model, provider)) - source = "llm-panels" - warnings: list[str] = [] - except Exception as exc: # noqa: BLE001 — panel/provider failure falls back, never crashes the scan - panels = [_panel_a(finding, verification), _panel_b(finding, verification)] - source = "deterministic-reconciler" - warnings = [f"llm_panels_failed: {type(exc).__name__}"] - disposition, _rationale = _reconcile(finding, verification, panels, mitigated_ids) - votes = Counter(str(panel.disposition) for panel in panels) - model_ids = sorted({panel.model_id for panel in panels if panel.model_id}) - if decider_model: - model_ids = sorted({*model_ids, decider_model}) - return FusionDecision( - finding_id=finding.finding_id, - triggered=True, - triggers=triggers, - disposition=disposition, - decision_source=source, - panels=panels, - votes=dict(sorted(votes.items())), - model_ids=model_ids, - degradations=[], - warnings=warnings, - ) - - -async def _run_llm_panels( - finding: StaticFinding, - verification: VerificationResult | None, - model: str, - provider: str, -) -> list[PanelVerdict]: - from harnessx.core.events import Message - from harnessx.processors.evaluation.llm_judge import build_judge_prompt, parse_judge_response - - from .harness_ext import build_provider - - client = build_provider(model, provider) - panels: list[PanelVerdict] = [] - for role, leaning, stance in ( - ("panel_a", "vulnerability", "Argue the vulnerability case as strongly as the evidence allows."), - ("panel_b", "false_positive", "Argue the false-positive case as strongly as the evidence allows."), - ): - summary = f"reachability={finding.reachability_status}; severity={finding.severity}; tags={', '.join(finding.tags) or 'none'}" - prompt = build_judge_prompt( - task_description=f"{stance} Finding: {finding.title} at {finding.path}:{finding.line}.", - trajectory_summary=summary, - extracted_answer=verification.pro_case if verification else finding.rationale, - ) - response = await client.complete(messages=[Message(role="user", content=prompt)], tools=[]) - verdict = parse_judge_response(_text(response.content)) - base = _panel_a(finding, verification) if role == "panel_a" else _panel_b(finding, verification) - plausible = str(verdict.get("verdict", "")) == "plausible" - confidence = ( - float(verdict.get("confidence", base.confidence)) if isinstance(verdict.get("confidence"), int | float) else base.confidence - ) - panels.append( - PanelVerdict( - role=role, - leaning=leaning, - disposition=base.disposition if plausible or role == "panel_a" else FusionDisposition.REJECTED, - confidence=confidence, - vulnerability_case=base.vulnerability_case, - false_positive_case=base.false_positive_case, - rationale=str(verdict.get("cause", base.rationale)), - model_id=model, - ) - ) - return panels - - -def _text(content: object) -> str: - if isinstance(content, str): - return content - if isinstance(content, list): - return "".join(str(block.get("text", "")) if isinstance(block, dict) else str(block) for block in content) - return str(content) - - __all__ = [ "FusionDecision", "FusionDisposition", "PanelVerdict", "fuse_finding", - "fuse_findings_dispatch", + "fuse_findings", "fusion_triggers", "should_fuse", ] diff --git a/src/openultrasast/harness.py b/src/openultrasast/harness.py index aca59e6..742ad48 100644 --- a/src/openultrasast/harness.py +++ b/src/openultrasast/harness.py @@ -180,7 +180,7 @@ def write_harness_config( model_roles={ "ranker": config.models.ranker, "hunter": config.models.hunter, - "verifier": config.models.verifier, + "verifier": None, # schema kept byte for byte; LLM verification runs on the plane "patcher": config.models.patcher, }, processor_versions={processor.spec.name: processor.spec.version for processor in processors}, diff --git a/src/openultrasast/slot_contract.py b/src/openultrasast/slot_contract.py index ca41b9a..ec646aa 100644 --- a/src/openultrasast/slot_contract.py +++ b/src/openultrasast/slot_contract.py @@ -4,14 +4,9 @@ The deterministic scan stages carry their work in named *slots* and run with mod invocation disabled (``skip_model = True``). Each stage declares a read/write slot allow-list that is validated on every lifecycle hook, preserving the ``ProcessorSpec.reads/writes`` provenance discipline of the sync ``HarnessRuntime`` -(the top-level driver) — without forcing the deterministic stages through HarnessX's -agent loop ("bridge, do not map", per the spec design). +(the top-level driver). -This is the zero-dependency substrate. When the optional ``openultrasast[harnessx]`` -extra is present, the same model-disabled processors can be hosted under a HarnessX -``MultiHookProcessor`` (see ``stage_processors.host_under_harnessx``); the -``skip_model`` marker maps onto HarnessX's ``BeforeModelEvent.skip_model`` so no -deterministic stage incurs a model call. +This is the zero-dependency substrate for the deterministic pipeline. """ from __future__ import annotations diff --git a/src/openultrasast/stage_processors.py b/src/openultrasast/stage_processors.py index 744cbb5..298c203 100644 --- a/src/openultrasast/stage_processors.py +++ b/src/openultrasast/stage_processors.py @@ -3,12 +3,8 @@ Each stage is a :class:`slot_contract.SlotContractMixin` with ``skip_model = True`` and a declared read/write slot allow-list. Running them through a :class:`SlotPipeline` reproduces the deterministic quick-scan path byte-for-byte while enforcing the slot -contract on every lifecycle hook. The hunter and verifier stages are *not* here — -they are the agentic plane (``hunter_harness``/``verify_judge``); this module is the -deterministic, model-free remainder. - -``host_under_harnessx`` is the optional, lazy bridge: it wraps a model-disabled stage -as a HarnessX ``MultiHookProcessor`` that disables the model call on ``before_model``. +contract on every lifecycle hook. The hunter and verifier stages are *not* here; this +module is the deterministic, model-free remainder that ``test_gate`` pins. """ from __future__ import annotations @@ -126,36 +122,6 @@ def run_quick_pipeline( return build_deterministic_pipeline().run(initial, runtime=runtime) -def host_under_harnessx(processor: SlotContractMixin) -> Any: - """Host a model-disabled deterministic stage under a HarnessX ``MultiHookProcessor``. - - Lazy + capability-guarded (imports ``harnessx`` only when called). On the - ``before_model`` hook the hosted stage yields a ``BeforeModelEvent`` with - ``skip_model=True``, so no deterministic stage incurs a model call when it runs on - the HarnessX runloop. Raises :class:`HarnessXUnavailableError` when the extra is absent. - """ - import dataclasses - - from .harness_ext import require_harnessx - - require_harnessx() - from harnessx.core.events import BeforeModelEvent - from harnessx.core.processor import MultiHookProcessor - - class _ModelDisabledStage(MultiHookProcessor): # type: ignore[misc, valid-type] - _singleton_group = "ousast_deterministic" - _order = 50 - - def __init__(self, stage: SlotContractMixin) -> None: - self.stage = stage - self.name = stage.name - - async def on_before_model(self, event: BeforeModelEvent): # type: ignore[no-untyped-def] - yield dataclasses.replace(event, skip_model=True, synthetic_output=f"[deterministic stage {self.stage.name}: model disabled]") - - return _ModelDisabledStage(processor) - - __all__ = [ "EntryPointProcessor", "PolicyScoringProcessor", @@ -165,6 +131,5 @@ __all__ = [ "ScoreProcessor", "build_deterministic_pipeline", "deterministic_processors", - "host_under_harnessx", "run_quick_pipeline", ] diff --git a/src/openultrasast/verify_judge.py b/src/openultrasast/verify_judge.py deleted file mode 100644 index 1621425..0000000 --- a/src/openultrasast/verify_judge.py +++ /dev/null @@ -1,130 +0,0 @@ -"""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 build_provider, 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, - verifier_provider: str = "anthropic", - 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, verifier_provider) - - -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, verifier_provider)) - - -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 - - provider = build_provider(verifier_model, verifier_provider) - 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) diff --git a/tests/test_cli_hx_dispatch.py b/tests/test_cli_hx_dispatch.py deleted file mode 100644 index 29e2be5..0000000 --- a/tests/test_cli_hx_dispatch.py +++ /dev/null @@ -1,65 +0,0 @@ -"""CLI capability-gated dispatch + graceful degradation (Phase 3 task 6.5). - -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 -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.setattr(cli, "has_harnessx", lambda: False) # simulate the extra being absent - monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") - monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) - monkeypatch.delenv("OPENULTRASAST_HUNTER_CLIENT", raising=False) - 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: - manifest = _run(tmp_path, monkeypatch, '[models]\nhunter = "anthropic/claude"\nverifier = "anthropic/claude"\n') - - degradations = manifest.get("degradations", []) - stages = {entry["stage"] for entry in degradations} - # `model` is present because the model layer records that no CPG engine was available rather than being - # silently absent (contributor-scan Req 1.3). This test is about the HarnessX extra, so it asserts over - # the HarnessX stages specifically instead of the whole set. - assert stages == {"hunter_pool", "verify", "model"} - assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in degradations if entry["stage"] != "model") - # Fallback still produced findings (deterministic hunter ran). - assert manifest["findings"] - - -def test_manifest_has_no_harnessx_degradations_without_models(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - manifest = _run(tmp_path, monkeypatch, config_body=None) - degradations = manifest.get("degradations", []) - assert all(entry.get("reason") != "harnessx_extra_unavailable" for entry in degradations) - assert any(entry.get("reason") == "hunter_model_unavailable" for entry in degradations) - - -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"] diff --git a/tests/test_config.py b/tests/test_config.py index 6df0ac2..7b2e0fd 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,7 +1,11 @@ +import logging import os +from dataclasses import replace from pathlib import Path -from openultrasast.config import load_config, load_dotenv, write_resolved_config +import pytest + +from openultrasast.config import RetiredConfigError, load_config, load_dotenv, write_resolved_config def test_load_config_reads_toml(tmp_path: Path) -> None: @@ -29,48 +33,64 @@ 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: +def test_retired_agentic_section_loads_with_one_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + # retired 2026-09-30: a budget-only [harnessx] section loads with one warning naming the plane (Req 3.3) config_path = tmp_path / "openultrasast.toml" - config_path.write_text( - "\n".join( - [ - "[harnessx]", - 'provider = "openai"', - "max_cost_usd = 5.0", - "token_threshold = 200000", - ] - ) - ) + config_path.write_text('[harnessx]\nprovider = "openai"\nmax_cost_usd = 5.0\n[score]\nmin_score = 70\n') # retired 2026-09-30 - config = load_config(config_path) + with caplog.at_level(logging.WARNING, logger="openultrasast.config"): + 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 + records = [record for record in caplog.records if record.name == "openultrasast.config"] + assert len(records) == 1 + assert "ax plane" in records[0].getMessage() and "ops/ax/README.md" in records[0].getMessage() + assert config.score.min_score == 70 # the rest of the file still loads + assert config == replace(load_config(None), score=config.score) # and nothing else is read from the section + + +@pytest.mark.parametrize( + ("body", "key", "replacement"), + [ + ('[models]\nverifier = "gpt-4o"\n', "`[models] verifier`", "`verify` + `agree`"), + ('[fusion]\npanel_model = "gpt-4o"\n', "`[fusion] panel_model`", "`agree` task"), + ('[fusion]\ndecider_model = "gpt-4o"\n', "`[fusion] decider_model`", "`agree` task"), + ], +) +def test_retired_llm_keys_fail_naming_the_replacement(tmp_path: Path, body: str, key: str, replacement: str) -> None: + config_path = tmp_path / "openultrasast.toml" + config_path.write_text(body) + + with pytest.raises(RetiredConfigError) as raised: + load_config(config_path) + + message = str(raised.value) + assert key in message and replacement in message and "ousast plane run" in message and "Remove the key" in message + + +def test_the_cli_exits_2_on_a_retired_key(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + from openultrasast.cli import main + + (tmp_path / "repo").mkdir() + config_path = tmp_path / "openultrasast.toml" + config_path.write_text('[models]\nverifier = "gpt-4o"\n') + + assert main(["scan", str(tmp_path / "repo"), "--config", str(config_path)]) == 2 + assert "`[models] verifier`" in capsys.readouterr().err def test_fusion_defaults_when_section_absent() -> None: config = load_config(None) assert config.fusion.enabled is True - assert config.fusion.panel_model is None assert config.fusion.high_assurance is False def test_load_config_reads_fusion_section(tmp_path: Path) -> None: config_path = tmp_path / "openultrasast.toml" - config_path.write_text('[fusion]\nenabled = true\npanel_model = "gpt-4o"\ndecider_model = "gpt-4o"\nhigh_assurance = true\n') + config_path.write_text("[fusion]\nenabled = false\nhigh_assurance = true\n") config = load_config(config_path) - assert config.fusion.panel_model == "gpt-4o" - assert config.fusion.decider_model == "gpt-4o" + assert config.fusion.enabled is False assert config.fusion.high_assurance is True diff --git a/tests/test_fusion.py b/tests/test_fusion.py index 1687a4b..3cbf66b 100644 --- a/tests/test_fusion.py +++ b/tests/test_fusion.py @@ -9,11 +9,10 @@ from openultrasast.findings import StaticFinding from openultrasast.fusion import ( FusionDisposition, fuse_finding, - fuse_findings_dispatch, + fuse_findings, fusion_triggers, should_fuse, ) -from openultrasast.harness_ext import has_harnessx from openultrasast.verification import EvidenceLevel, VerificationResult, VerificationStatus @@ -135,18 +134,10 @@ def test_dispatch_only_fuses_triggered_findings() -> None: _finding(finding_id="hot:app.py:3", severity="critical"), _finding(finding_id="cold:lib.py:9", severity="low", reachability="unknown"), ] - decisions = fuse_findings_dispatch(findings, []) + decisions = fuse_findings(findings, []) assert [d.finding_id for d in decisions] == ["hot:app.py:3"] -def test_dispatch_records_degradation_when_llm_requested_but_extra_absent() -> None: - if has_harnessx(): - pytest.skip("HarnessX extra installed; exercises the extra-absent degradation path") - decisions = fuse_findings_dispatch([_finding()], [], panel_model="gpt-4o", use_harnessx=True) - assert decisions and decisions[0].decision_source == "deterministic-reconciler" - assert any("fusion_llm_unavailable" in note for note in decisions[0].degradations) - - def test_decision_serializes_with_full_disclosure() -> None: decision = fuse_finding(_finding(), _verification(VerificationStatus.ACCEPTED, verified=True), ["severity:critical"]) payload = json.loads(json.dumps(decision.to_dict())) diff --git a/tests/test_scan_deep.py b/tests/test_scan_deep.py index baf2b68..1a8af73 100644 --- a/tests/test_scan_deep.py +++ b/tests/test_scan_deep.py @@ -141,7 +141,6 @@ def test_deep_scan_unsafe_hunter_snippet_is_safety_rejected_without_docker_job(t monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1") monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "unsafe-snippet") - monkeypatch.setattr(cli, "has_harnessx", lambda: False) monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake) docker_argv = _guard_docker(monkeypatch) diff --git a/tests/test_scan_pipeline_integrity.py b/tests/test_scan_pipeline_integrity.py index 45d1238..0b1281b 100644 --- a/tests/test_scan_pipeline_integrity.py +++ b/tests/test_scan_pipeline_integrity.py @@ -1,12 +1,10 @@ """Scan-pipeline integrity & serialization (Phase 4 task 9). -9.1 — with the extra absent, quick and standard scans run through the existing sync -driver with behaviour unchanged from the deterministic baseline, and a -requested-but-unavailable HarnessX stage records a manifest degradation. +9.1 — quick and standard scans run through the sync driver with the deterministic +hunter, verifier and fusion; a configured hunter model never changes the +deterministic findings. 9.2 — every persisted scan-state slot serializes and restores without loss (guards -against a non-serializable slot silently restoring as empty). This covers the -current artifact-persistence model; the HarnessX processor slot-contract variant -(spec task 9.2 over task 6.3) remains deferred with 6.3. +against a non-serializable slot silently restoring as empty). """ import json @@ -15,7 +13,6 @@ from pathlib import Path import pytest -from openultrasast import cli from openultrasast.benchmark import load_findings from openultrasast.cli import main from openultrasast.findings import StaticFinding, write_findings @@ -27,14 +24,13 @@ from openultrasast.verification import EvidenceLevel, VerificationResult, Verifi _VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n" -# ---- 9.1 fallback parity + degradation -------------------------------------- +# ---- 9.1 deterministic parity + degradation -------------------------------------- -def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_body: str | None, harnessx: bool) -> tuple[list, dict]: +def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_body: str | None) -> tuple[list, dict]: repo = tmp_path / "repo" repo.mkdir(parents=True) (repo / "app.py").write_text(_VULN) - monkeypatch.setattr(cli, "has_harnessx", lambda: harnessx) monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) monkeypatch.delenv("OPENULTRASAST_HUNTER_CLIENT", raising=False) @@ -50,40 +46,27 @@ def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_ return findings, manifest -def test_standard_scan_extra_absent_matches_deterministic_baseline(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - # Requested-but-unavailable HarnessX (models set, extra absent) must produce the - # same findings as the plain deterministic path (no models), differing only by - # the recorded degradation. - requested, requested_manifest = _scan( - tmp_path / "a", - monkeypatch, - mode="standard", - config_body='[models]\nhunter = "openai/gpt-4o"\nverifier = "openai/gpt-4o"\n', - harnessx=False, - ) - baseline, baseline_manifest = _scan(tmp_path / "b", monkeypatch, mode="standard", config_body=None, harnessx=False) +def test_standard_scan_with_a_hunter_model_keeps_deterministic_findings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # A configured hunter model (no client reachable here) must leave the deterministic findings byte-identical. + requested, requested_manifest = _scan(tmp_path / "a", monkeypatch, mode="standard", config_body='[models]\nhunter = "openai/gpt-4o"\n') + baseline, baseline_manifest = _scan(tmp_path / "b", monkeypatch, mode="standard", config_body=None) - assert requested == baseline # byte-identical findings -> sync driver behaviour unchanged - stages = {entry["stage"] for entry in requested_manifest.get("degradations", [])} - # `model` joins the set when the model layer runs without a CPG engine: findings are byte-identical - # (asserted above) and the only difference is that the scan now SAYS the arbiter was unavailable - # (contributor-scan Req 1.3) rather than being silently absent. - assert stages == {"hunter_pool", "verify", "model"} - assert {e["reason"] for e in requested_manifest["degradations"] if e["stage"] == "model"} == {"cpg_unavailable"} - assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in requested_manifest["degradations"] if entry["stage"] != "model") - assert all(entry["reason"] != "hunter_model_unavailable" for entry in requested_manifest["degradations"]) - assert [e["reason"] for e in baseline_manifest.get("degradations", []) if e["stage"] != "model"] == ["hunter_model_unavailable"] + assert requested == baseline + # The hunter model runs the tool hunter (its client is absent here, so it contributes nothing) and removes the + # `map` skip; `model` records that no CPG engine was available (contributor-scan Req 1.3). + assert requested_manifest["degradations"] == [{"reason": "cpg_unavailable", "stage": "model"}] + assert {(e["stage"], e["reason"]) for e in baseline_manifest["degradations"]} == { + ("model", "cpg_unavailable"), + ("map", "hunter_model_unavailable"), + } -def test_quick_scan_is_unaffected_by_the_extra(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - # Quick mode never touches the agentic plane: identical findings whether the - # extra is present or absent, and never a degradation. - present, present_manifest = _scan(tmp_path / "a", monkeypatch, mode="quick", config_body=None, harnessx=True) - absent, _ = _scan(tmp_path / "b", monkeypatch, mode="quick", config_body=None, harnessx=False) +def test_quick_scan_records_no_degradation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # Quick mode never touches a model: findings are produced and no degradation is recorded. + findings, manifest = _scan(tmp_path, monkeypatch, mode="quick", config_body=None) - assert present == absent - assert present # findings were produced - assert "degradations" not in present_manifest + assert findings + assert "degradations" not in manifest # ---- 9.2 per-slot serialization round-trip ---------------------------------- diff --git a/tests/test_slot_contract.py b/tests/test_slot_contract.py index a44d4fa..f328bee 100644 --- a/tests/test_slot_contract.py +++ b/tests/test_slot_contract.py @@ -2,8 +2,7 @@ Proves: deterministic stages carry work in declared slots, a slot-contract violation is raised on undeclared slot access, the pipeline is model-disabled, the slot pipeline -reproduces the deterministic quick-scan path byte-for-byte, lifecycle hooks fire, and -(when the extra is present) a stage can run under HarnessX with the model disabled. +reproduces the deterministic quick-scan path byte-for-byte, and lifecycle hooks fire. """ from collections.abc import Mapping @@ -13,13 +12,10 @@ from typing import Any import pytest from openultrasast.gate import _quick_scan -from openultrasast.harness_ext import has_harnessx from openultrasast.slot_contract import SlotContractError, SlotContractMixin, SlotPipeline from openultrasast.stage_processors import ( - PreprocessProcessor, build_deterministic_pipeline, deterministic_processors, - host_under_harnessx, run_quick_pipeline, ) @@ -110,24 +106,3 @@ def test_pipeline_emits_lifecycle_hooks_per_stage() -> None: for name in names: assert ("stage_start", name) in recorder.events assert ("stage_end", name) in recorder.events - - -# ---- optional: run under HarnessX with the model disabled (1.6) -------------- - - -@pytest.mark.skipif(not has_harnessx(), reason="requires the harnessx extra") -def test_host_under_harnessx_disables_the_model() -> None: # pragma: no cover - needs the extra - import asyncio - - from harnessx.core.events import BeforeModelEvent - from harnessx.core.processor import MultiHookProcessor - - hosted = host_under_harnessx(PreprocessProcessor()) - assert isinstance(hosted, MultiHookProcessor) - - async def drive() -> list[Any]: - event = BeforeModelEvent(run_id="test", step_id=0) - return [out async for out in hosted.on_before_model(event)] - - emitted = asyncio.run(drive()) - assert emitted and emitted[-1].skip_model is True diff --git a/tests/test_tool_hunter.py b/tests/test_tool_hunter.py index 773af55..8269cb6 100644 --- a/tests/test_tool_hunter.py +++ b/tests/test_tool_hunter.py @@ -219,14 +219,13 @@ def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]: def test_standard_scripted_hunter_emits_tool_hunter_suspicion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - from openultrasast import cli, tool_hunter + from openultrasast import tool_hunter from openultrasast.cli import main repo = _standard_repo(tmp_path) config = _write_hunter_config(tmp_path) monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "scripted") - monkeypatch.setattr(cli, "has_harnessx", lambda: False) docker_argv = _guard_docker(monkeypatch) hunter_calls: list[int] = [] real_hunter = tool_hunter.run_tool_hunter @@ -256,14 +255,13 @@ def test_standard_scripted_hunter_emits_tool_hunter_suspicion(tmp_path: Path, mo def test_dump_only_hunter_client_emits_no_hunter_findings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - from openultrasast import cli, tool_hunter + from openultrasast import tool_hunter from openultrasast.cli import main repo = _standard_repo(tmp_path) config = _write_hunter_config(tmp_path) monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "dump-only") - monkeypatch.setattr(cli, "has_harnessx", lambda: False) docker_argv = _guard_docker(monkeypatch) hunter_calls: list[int] = [] real_hunter = tool_hunter.run_tool_hunter diff --git a/tests/test_verify_judge.py b/tests/test_verify_judge.py deleted file mode 100644 index 44c65a6..0000000 --- a/tests/test_verify_judge.py +++ /dev/null @@ -1,70 +0,0 @@ -"""Verifier dispatcher: structural fallback (default) + judge verdict mapping (Phase 3 task 6.4).""" - -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(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: - 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