From 3091ed42eaf71a17b6d0565413aeeb540d13a586 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Fri, 2 Oct 2026 07:22:35 +0200 Subject: [PATCH] chore: remove dead modules (judge, execution, calibrate, stage_processors, slot_contract) The reachability-first audit (previous commit) found these five with no import path from any entry point: - model/judge.py: replaced by model/pipeline.py, which never imported it. - model/execution.py: the MGD Req 10 execution seam, never built or called. - model/calibrate.py: no CLI, no importer; learn/calibrate.py is the live one. - stage_processors.py + slot_contract.py: kept only so test_gate could pin them; the scan runs findings.quick_scan_findings directly. Their tests go with them, test_gate's zero-dependency list drops the two entries, the audit lists drop their rows, and the manifest is regenerated: 150 modules, 145 load_bearing, 5 standalone, 0 orphaned. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- .../measurements/2026-09-08-module-audit.json | 76 +------- src/openultrasast/model/audit.py | 7 +- src/openultrasast/model/calibrate.py | 180 ------------------ src/openultrasast/model/execution.py | 118 ------------ src/openultrasast/model/judge.py | 136 ------------- src/openultrasast/slot_contract.py | 88 --------- src/openultrasast/stage_processors.py | 135 ------------- tests/test_gate.py | 2 - tests/test_model_calibrate.py | 97 ---------- tests/test_model_execution.py | 116 ----------- tests/test_model_judge.py | 163 ---------------- tests/test_module_audit.py | 5 +- tests/test_slot_contract.py | 108 ----------- 13 files changed, 14 insertions(+), 1217 deletions(-) delete mode 100644 src/openultrasast/model/calibrate.py delete mode 100644 src/openultrasast/model/execution.py delete mode 100644 src/openultrasast/model/judge.py delete mode 100644 src/openultrasast/slot_contract.py delete mode 100644 src/openultrasast/stage_processors.py delete mode 100644 tests/test_model_calibrate.py delete mode 100644 tests/test_model_execution.py delete mode 100644 tests/test_model_judge.py delete mode 100644 tests/test_slot_contract.py diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 116f2a4..182a817 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,17 +1,10 @@ { - "modules": 155, + "modules": 150, "counts": { "load_bearing": 145, - "orphaned": 5, "standalone_capability": 5 }, - "orphaned": [ - "model.calibrate", - "model.execution", - "model.judge", - "slot_contract", - "stage_processors" - ], + "orphaned": [], "standalone": { "fusion": "two-panel adjudication engine; reached from the scan's report stage and used standalone", "mcp": "narrow MCP server over stdio; its own `ousast mcp` entry point and OpenCode integration", @@ -156,7 +149,6 @@ "cpg.backend", "model.config_value", "model.dominance", - "model.judge", "model.partitions", "model.pipeline", "model.scan", @@ -190,7 +182,6 @@ "cpg.artifact", "model.config_value", "model.dominance", - "model.judge", "model.partitions", "model.pipeline", "model.scan", @@ -245,7 +236,6 @@ "semantic.obligations.check", "semantic.overlay", "semantic.prove_filter", - "stage_processors", "tool_hunter", "verification" ] @@ -272,8 +262,7 @@ "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate, cli", "importers": [ - "cli", - "slot_contract" + "cli" ] }, { @@ -557,8 +546,7 @@ "pairs", "plane.tasks.alerts", "preprocess", - "push.runner", - "stage_processors" + "push.runner" ] }, { @@ -583,8 +571,6 @@ "model.candidates", "model.config_value", "model.dominance", - "model.execution", - "model.judge", "model.layout", "model.partitions", "model.pipeline", @@ -616,12 +602,6 @@ "reason": "model-grounded-detection Req 3: this classifier; run by the maintainer, not by the scan", "importers": [] }, - { - "module": "model.calibrate", - "classification": "orphaned", - "reason": "no import path from an entry point or a python -m module; listed as: model-grounded-detection Req 9: the corpus as the model's calibration set", - "importers": [] - }, { "module": "model.candidates", "classification": "load_bearing", @@ -674,7 +654,7 @@ { "module": "model.endpoint", "classification": "load_bearing", - "reason": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3", + "reason": "model-grounded-detection Req 8: the bounded model client the hunter, the plane and the CLI share", "importers": [ "cli", "learn.cli", @@ -692,18 +672,6 @@ "model.scan" ] }, - { - "module": "model.execution", - "classification": "orphaned", - "reason": "no import path from an entry point or a python -m module; listed as: model-grounded-detection Req 10: the deferred execution tier seam, adopted not built", - "importers": [] - }, - { - "module": "model.judge", - "classification": "orphaned", - "reason": "no import path from an entry point or a python -m module; listed as: model-grounded-detection Req 8: one bounded question, checked against the model", - "importers": [] - }, { "module": "model.ladder", "classification": "load_bearing", @@ -711,8 +679,6 @@ "importers": [ "model.config_value", "model.dominance", - "model.execution", - "model.judge", "model.pipeline", "model.scan", "model.taint", @@ -797,7 +763,6 @@ "importers": [ "model.config_value", "model.dominance", - "model.judge", "model.pipeline", "model.regions", "model.report", @@ -811,7 +776,6 @@ "classification": "load_bearing", "reason": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter", "importers": [ - "model.judge", "model.pipeline", "model.scan" ] @@ -1113,8 +1077,7 @@ "improve.validator", "plane.tasks.loop", "scoring.project_score", - "semantic.overlay", - "stage_processors" + "semantic.overlay" ] }, { @@ -1157,8 +1120,7 @@ "rank", "regress.candidate", "semantic.overlay", - "skills", - "stage_processors" + "skills" ] }, { @@ -1337,8 +1299,7 @@ "improve.evolve", "map_gate", "pairs", - "plane.tasks.alerts", - "stage_processors" + "plane.tasks.alerts" ] }, { @@ -1347,7 +1308,6 @@ "reason": "reachable from gate, map_gate, pair_gate, cli", "importers": [ "harness", - "model.judge", "model.pipeline", "push.assistance", "push.report", @@ -1435,8 +1395,7 @@ "plane.tasks.features", "plane.tasks.loop", "ruleset.store", - "semantic.facts", - "stage_processors" + "semantic.facts" ] }, { @@ -1505,8 +1464,7 @@ "reason": "reachable from gate, map_gate, pair_gate, cli", "importers": [ "cli", - "improve.evolve", - "stage_processors" + "improve.evolve" ] }, { @@ -1760,20 +1718,6 @@ "hunter" ] }, - { - "module": "slot_contract", - "classification": "orphaned", - "reason": "no import path from an entry point or a python -m module (imported only by unreachable stage_processors); listed as: the slot/contract protocol stage_processors implements", - "importers": [ - "stage_processors" - ] - }, - { - "module": "stage_processors", - "classification": "orphaned", - "reason": "no import path from an entry point or a python -m module; listed as: deterministic slot-contracted scan stages; pinned by the zero-dependency guard in test_gate", - "importers": [] - }, { "module": "stages", "classification": "load_bearing", diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 3449d68..85cbf22 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -41,16 +41,13 @@ SPEC_OWNED = { "push": "pre-push-safety-net Req 5.5/7.3: experimental push contract package; no hook installed", "push.contracts": "pre-push-safety-net Req 5.5/7.3: immutable comparisons and independent result statuses; runner pending", "model.specs": "model-grounded-detection Req 7.2-7.4: the security vocabularies the CPG queries are parameterised by", - "model.endpoint": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3", + "model.endpoint": "model-grounded-detection Req 8: the bounded model client the hunter, the plane and the CLI share", "model.audit": "model-grounded-detection Req 3: this classifier; run by the maintainer, not by the scan", "model.ladder": "model-grounded-detection Req 5: the evidence ladder every finding carries", "model.taint": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter", - "model.judge": "model-grounded-detection Req 8: one bounded question, checked against the model", "model.dominance": "model-grounded-detection Req 6.2: guard dominance, the arbiter for bugs that never crash", "model.config_value": "model-grounded-detection Req 6.3: constant abstraction for the configuration families", - "model.calibrate": "model-grounded-detection Req 9: the corpus as the model's calibration set", "model.report": "model-grounded-detection Req 9.3/11.1: per-slice reporting with the overfitting gap", - "model.execution": "model-grounded-detection Req 10: the deferred execution tier seam, adopted not built", "model.pipeline": "model-grounded-detection Req 8: the enumerator proposes, the LLM answers, the model disposes", "model.regions": "contributor-scan Req 2: regions from entry points, with the families each admits", "model.scan": "contributor-scan Req 3: the repository driver -- one CPG, one budget, ranked spend", @@ -91,8 +88,6 @@ 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", - "stage_processors": "deterministic slot-contracted scan stages; pinned by the zero-dependency guard in test_gate", - "slot_contract": "the slot/contract protocol stage_processors implements", } diff --git a/src/openultrasast/model/calibrate.py b/src/openultrasast/model/calibrate.py deleted file mode 100644 index fedda9c..0000000 --- a/src/openultrasast/model/calibrate.py +++ /dev/null @@ -1,180 +0,0 @@ -"""The corpus calibrates the model, not an LLM vote (model-grounded-detection, Req 9). - -This is the inversion the whole feature rests on. In the prior architecture a pair the detector missed was a -*detector* failure, to be averaged away over K runs and chased with a prompt change. Here the arbiter is -deterministic: it will miss that pair the same way every time, and the only thing that moves it is better -modelling. So a pair the model cannot split is a **named gap in the model**, recorded with its reason, and -the report's job is to make those gaps legible by family and by slice rather than to produce a score. - -`covered` means the model reached a strictly higher rung on the vulnerable side than on its fix. Flagging -both sides equally is not coverage — it distinguishes nothing, which is exactly the failure the group-2 -measurement found taint reachability making on every parameterised SQL pair. -""" - -from __future__ import annotations - -from collections.abc import Iterable, Mapping, Sequence -from dataclasses import dataclass, field -from typing import TypedDict - - -class FamilyGaps(TypedDict): - """Gaps for one family: how many pairs, and the tally of reasons behind them.""" - - pairs: int - reasons: dict[str, int] - - -# Ascending, matching `ladder.Rung`. A pair is split when the vulnerable side outranks its fix. -_ORDER = {"none": 0, "": 0, "suspicion": 0, "model_corroborated": 1, "model_entailed": 2, "execution_confirmed": 3} - - -@dataclass(frozen=True) -class PairOutcome: - """What the model made of one vulnerable/fixed pair.""" - - pair: str - family: str - slice: str - vuln_rung: str - fixed_rung: str - reason: str = "" - - -@dataclass(frozen=True) -class ModelGap: - """A pair the model could not split, and why. Not a miss to average — a gap to close.""" - - pair: str - family: str - slice: str - reason: str - - -@dataclass(frozen=True) -class HarvestRecipe: - """Where a pair comes from. The fix commit's parent is the vulnerable side, by construction.""" - - repo: str - fix_commit: str - relpath: str - function: str = "" - - @property - def parent_ref(self) -> str: - return f"{self.fix_commit}^" - - -@dataclass(frozen=True) -class CalibrationReport: - covered: tuple[str, ...] - gaps: tuple[ModelGap, ...] - per_slice: Mapping[str, Mapping[str, int]] - per_family: Mapping[str, Mapping[str, int]] = field(default_factory=dict) - - def gaps_by_family(self) -> dict[str, FamilyGaps]: - """Gaps grouped by family, each with a tally of the reasons behind them.""" - out: dict[str, FamilyGaps] = {} - for gap in self.gaps: - block = out.setdefault(gap.family, {"pairs": 0, "reasons": {}}) - block["pairs"] += 1 - block["reasons"][gap.reason] = block["reasons"].get(gap.reason, 0) + 1 - return out - - def to_markdown(self) -> str: - """Per-slice rows first; there is no headline here that is not backed by them (Req 9.3, 11.1).""" - lines = [ - "## Model calibration", - "", - "Coverage is a pair the model *splits*: a strictly higher rung on", - "the vulnerable side than on its fix. Everything else is a named gap.", - "", - "| slice | pairs | covered | gaps | coverage |", - "|---|---|---|---|---|", - ] - for slice_name, block in sorted(self.per_slice.items()): - pairs = block["pairs"] or 1 - lines.append(f"| {slice_name} | {block['pairs']} | {block['covered']} | {block['gaps']} | {block['covered'] / pairs:.0%} |") - by_family = self.gaps_by_family() - if by_family: - lines += ["", "### Gaps by family", ""] - for family, gaps in sorted(by_family.items()): - detail = ", ".join(f"{k} x{v}" for k, v in sorted(gaps["reasons"].items())) - lines.append(f"- `{family}`: {gaps['pairs']} pair(s) — {detail}") - return "\n".join(lines) + "\n" - - -def calibrate(outcomes: Iterable[PairOutcome]) -> CalibrationReport: - """Split the corpus into what the model covers and what it cannot, with reasons.""" - covered: list[str] = [] - gaps: list[ModelGap] = [] - per_slice: dict[str, dict[str, int]] = {} - per_family: dict[str, dict[str, int]] = {} - - for outcome in outcomes: - split = _ORDER.get(outcome.vuln_rung, 0) > _ORDER.get(outcome.fixed_rung, 0) - for key, store in ((outcome.slice, per_slice), (outcome.family, per_family)): - block = store.setdefault(key, {"pairs": 0, "covered": 0, "gaps": 0}) - block["pairs"] += 1 - block["covered" if split else "gaps"] += 1 - if split: - covered.append(outcome.pair) - else: - gaps.append( - ModelGap(pair=outcome.pair, family=outcome.family, slice=outcome.slice, reason=outcome.reason or _default_reason(outcome)) - ) - return CalibrationReport( - covered=tuple(sorted(covered)), - gaps=tuple(sorted(gaps, key=lambda g: (g.family, g.pair))), - per_slice=per_slice, - per_family=per_family, - ) - - -def _default_reason(outcome: PairOutcome) -> str: - """Every gap carries a reason, even when the caller supplied none — a bare gap teaches nothing.""" - if _ORDER.get(outcome.vuln_rung, 0) == 0: - return "no_verdict_on_the_vulnerable_side" - if outcome.vuln_rung == outcome.fixed_rung: - return f"both_sides_reached_{outcome.vuln_rung}: the model distinguished nothing" - return "fixed_side_outranked_the_vulnerable_one" - - -def harvest_recipe(entry: Mapping[str, object]) -> HarvestRecipe: - """A pair from CVE history: the fix commit and its parent, with the fix diff as the oracle (Req 9.4). - - Nothing here is hand-labelled. The recipe names where to fetch both sides; what makes the pair *true* is - the fix diff itself, read by the model rather than annotated by us. - """ - return HarvestRecipe( - repo=str(entry.get("repo", "")), - fix_commit=str(entry.get("fix_commit", "")), - relpath=str(entry.get("relpath", "")), - function=str(entry.get("function", "") or ""), - ) - - -def outcomes_from_rows(rows: Sequence[Mapping[str, object]]) -> list[PairOutcome]: - """Adapt the ceiling artifact's rows, so calibration reads the measurement rather than re-running it.""" - return [ - PairOutcome( - pair=str(row.get("pair", "")), - family=str(row.get("family", "")), - slice=str(row.get("slice", "")), - vuln_rung=str(row.get("vuln_rung", "none")), - fixed_rung=str(row.get("fixed_rung", "none")), - reason=str(row.get("miss_cause", "") or ""), - ) - for row in rows - ] - - -__all__ = [ - "CalibrationReport", - "HarvestRecipe", - "ModelGap", - "PairOutcome", - "calibrate", - "harvest_recipe", - "outcomes_from_rows", -] diff --git a/src/openultrasast/model/execution.py b/src/openultrasast/model/execution.py deleted file mode 100644 index e10aa0d..0000000 --- a/src/openultrasast/model/execution.py +++ /dev/null @@ -1,118 +0,0 @@ -"""The execution tier: deferred, adopted, and deliberately narrow (model-grounded-detection, Req 10). - -This is the rung this project does **not** build. Clearwing already runs sandbox lifecycles, container pools, -sanitizer images, PoC replay and stability classification, and the corrective brief that started this feature -concluded — from this project's own defect history — that reimplementing that class of machinery is exactly -where we go wrong. So the module is a seam: it decides *eligibility*, calls an adopted tool, and records what -happened. A test asserts structurally that it imports no container or subprocess machinery and defines no -build/run/compile/patch function, because the failure mode to prevent is this file quietly growing a sandbox. - -Eligibility is narrow on purpose, and both halves matter: - -* **Only a family a static model cannot arbitrate.** Memory safety in C turns on runtime layout, so a crash - is the right oracle. Every web/logic family has taint reachability, guard dominance or constant - abstraction, and Req 10.3 forbids any of them depending on this tier. -* **Only a finding the model could not decide.** The tier escalates uncertainty; it never re-checks a verdict - the model already reached. - -And a failure to reproduce is *not* a refutation. A sandbox that does not trigger a bug has not shown the bug -is absent — it has shown this attempt did not trigger it. The candidate stays exactly where the model left it. -""" - -from __future__ import annotations - -import os -from dataclasses import dataclass -from typing import Protocol - -from .ladder import Rung - -# Families whose bugs a static model genuinely cannot arbitrate. Everything else is served by taint -# reachability, guard dominance or constant abstraction, and must not wait on a sandbox. -EXECUTION_FAMILIES = frozenset({"memory"}) - -# Rungs that mean "the model could not decide". Anything above is already arbitrated. -UNDECIDED = frozenset({"suspicion", ""}) - -CLEARWING_ENV = "OPENULTRASAST_CLEARWING" -_OFF = frozenset({"0", "off", "false", "no"}) -_ON = frozenset({"1", "on", "true", "yes"}) - - -@dataclass(frozen=True) -class Candidate: - """What the execution tier is asked about: a finding the static model left undecided.""" - - finding_id: str - family: str - rung: str - path: str - line: int | None = None - - -@dataclass(frozen=True) -class Outcome: - """What the tier concluded. ``rung`` is ``None`` whenever nothing was established.""" - - rung: Rung | None - reason: str - - -class ExecutionBackend(Protocol): - def available(self) -> bool: ... - - def reproduce(self, candidate: Candidate) -> bool: ... - - -@dataclass -class NullExecutionBackend: - """No adopted tool present. The normal case, and not an error.""" - - def available(self) -> bool: - return False - - def reproduce(self, candidate: Candidate) -> bool: - del candidate - return False - - -def eligible(candidate: Candidate) -> bool: - """Is this candidate one the execution tier should even look at?""" - return candidate.family in EXECUTION_FAMILIES and candidate.rung in UNDECIDED - - -def confirm(candidate: Candidate, *, backend: ExecutionBackend | None = None) -> Outcome: - """Escalate one undecided candidate to the adopted execution tier, if there is one.""" - if not eligible(candidate): - return Outcome( - rung=None, - reason=f"not_eligible: family {candidate.family!r} at rung {candidate.rung!r} is arbitrated statically", - ) - engine = backend if backend is not None else resolve_execution_backend() - if not engine.available(): - return Outcome(rung=None, reason="clearwing_unavailable: no adopted execution tier on this machine") - if not engine.reproduce(candidate): - # Deliberate wording: this is not a refutation. Failing to trigger a bug is not showing it is absent. - return Outcome(rung=None, reason="not_reproduced: the attempt did not trigger it; the finding is unchanged") - return Outcome(rung=Rung.EXECUTION_CONFIRMED, reason="reproduced in the adopted sandbox") - - -def resolve_execution_backend() -> ExecutionBackend: - """The adopted tier for this machine. Absent by default; the environment flag is what the matrix drives.""" - flag = os.environ.get(CLEARWING_ENV, "").strip().lower() - if flag in _OFF or flag not in _ON: - return NullExecutionBackend() - return NullExecutionBackend() # the adopted tool is wired here when a class needs it (Req 10.1) - - -__all__ = [ - "CLEARWING_ENV", - "EXECUTION_FAMILIES", - "Candidate", - "ExecutionBackend", - "NullExecutionBackend", - "Outcome", - "confirm", - "eligible", - "resolve_execution_backend", -] diff --git a/src/openultrasast/model/judge.py b/src/openultrasast/model/judge.py deleted file mode 100644 index 5c1ffe4..0000000 --- a/src/openultrasast/model/judge.py +++ /dev/null @@ -1,136 +0,0 @@ -"""The LLM proposes; the model disposes (model-grounded-detection, Req 8). - -This module is where the prior architecture's premise is inverted. There, the LLM was an oracle whose noise -had to be averaged over K runs, floored per family, and gated by a sign test. Here it answers one bounded, -typed question about a candidate it was *handed*, and its answer is checked against the CPG. Three outcomes, -and they are exhaustive: - - the model entails it report `model_entailed`, and never call the model at all - the model contradicts it drop the claim, recording the contradiction - the model cannot decide report `suspicion`, honestly - -Nothing here averages, votes, retries for agreement, or maintains a budget: a deterministic arbiter removes -the need for every one of those. The LLM is consulted only for the residual the graph cannot settle — whether -a sanitizer on a real flow is actually sufficient — and even that answer only ever *confirms* a flow the CPG -already found. It can never conjure one. -""" - -from __future__ import annotations - -import json -import logging -from collections.abc import Mapping -from typing import Any - -from ..cpg.backend import CpgResult -from ..redaction import redact_secrets -from .ladder import Rung, Verdict -from .specs import TaintSpec -from .taint import verdict as taint_verdict - -logger = logging.getLogger(__name__) - -QUESTION = """You are judging ONE candidate a static model already found. Do not look for other issues. - -A dataflow analysis resolved this path in {language} code: - - source: {source} - sink: {sink} (line {line}, function {function}) - a sanitizer appears on the path: {sanitized} - -The graph has established that the value reaches the sink. The only open question is whether the sanitizer on -that path actually makes it safe for the {family} weakness class. - -Answer ONLY with JSON: {{"vulnerable": true|false, "why": ""}}""" - - -def judge( - cpg: CpgResult, - spec: TaintSpec, - *, - function: str = "", - client: Any | None = None, - model: str = "", - parameter_sources: bool = False, -) -> Verdict: - """Arbitrate one candidate. Always returns a verdict; ``suspicion`` is a real answer, not a failure.""" - answer = taint_verdict(cpg, spec, function=function, parameter_sources=parameter_sources) - - if answer is not None and answer.rung is Rung.ENTAILED: - return answer # the graph decided; the model is not consulted - - if answer is None: - # No resolved flow. Whatever anyone claims about this site, the model saw nothing to support it. - return Verdict( - rung=Rung.SUSPICION, - family=spec.family, - contradiction="the model resolved no flow: no source reaches this sink", - ) - - # A real flow through a sanitizer. Whether that sanitizer suffices is a semantic judgement the graph - # cannot make, so it is the one thing the LLM is asked -- about evidence the model itself produced. - if client is None: - return Verdict( - rung=Rung.SUSPICION, - family=spec.family, - witness=answer.witness, - contradiction="no chat endpoint: the sanitizer's sufficiency was never judged", - ) - - said_vulnerable = _ask(client, model, spec, answer.witness, function) - if said_vulnerable is None: - return Verdict(rung=Rung.SUSPICION, family=spec.family, witness=answer.witness, contradiction="the judge returned no usable answer") - if not said_vulnerable: - return Verdict( - rung=Rung.SUSPICION, - family=spec.family, - witness=answer.witness, - contradiction="the judge held the sanitizer on this path sufficient", - ) - return answer # CORROBORATED: a real flow the judge agrees is not neutralised - - -def _ask(client: Any, model: str, spec: TaintSpec, witness: str, function: str) -> bool | None: - """One question, one answer. No tools, no retries for agreement, no second opinion.""" - source, _, sink = witness.partition(" -> ") - line = "" - if "(line " in sink: - sink, _, tail = sink.partition(" (line ") - line = tail.split(",")[0].strip(") ") - prompt = redact_secrets( - QUESTION.format( - language=spec.language, - source=source.strip(), - sink=sink.strip(), - line=line or "?", - function=function or "?", - sanitized="yes", - family=spec.family, - ) - ) - try: - response = client.complete( - model=model, - messages=[{"role": "user", "content": prompt}], - tools=[], # the candidate is given; the model never searches for it - json_object=True, - ) - except Exception as exc: # noqa: BLE001 -- an endpoint failure is a degradation, never a verdict - logger.warning("judge call failed: %s", exc) - return None - return _vulnerable(getattr(response, "content", None)) - - -def _vulnerable(content: object) -> bool | None: - if not isinstance(content, str) or not content.strip(): - return None - try: - payload = json.loads(content) - except json.JSONDecodeError: - return None - if isinstance(payload, Mapping) and isinstance(payload.get("vulnerable"), bool): - return bool(payload["vulnerable"]) - return None - - -__all__ = ["QUESTION", "judge"] diff --git a/src/openultrasast/slot_contract.py b/src/openultrasast/slot_contract.py deleted file mode 100644 index ec646aa..0000000 --- a/src/openultrasast/slot_contract.py +++ /dev/null @@ -1,88 +0,0 @@ -"""Slot-contract discipline for model-disabled deterministic stages (task 6.3). - -The deterministic scan stages carry their work in named *slots* and run with model -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). - -This is the zero-dependency substrate for the deterministic pipeline. -""" - -from __future__ import annotations - -from collections.abc import Mapping -from typing import Any, Protocol - -from .harness import HarnessContractError - - -class SlotContractError(HarnessContractError): - """Raised when a stage reads or writes a slot outside its declared allow-list.""" - - -class SlotContractMixin: - """A model-disabled stage that carries work in slots under a read/write allow-list.""" - - name: str = "slot-stage" - version: str = "1" - reads_slots: tuple[str, ...] = () - writes_slots: tuple[str, ...] = () - skip_model: bool = True # deterministic: never invokes a model - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - raise NotImplementedError - - def validate_reads(self, slots: Mapping[str, Any]) -> None: - missing = sorted(slot for slot in self.reads_slots if slot not in slots) - if missing: - raise SlotContractError(f"{self.name}: missing read slots: {', '.join(missing)}") - - def validate_writes(self, updates: Mapping[str, Any]) -> None: - undeclared = sorted(set(updates) - set(self.writes_slots)) - if undeclared: - raise SlotContractError(f"{self.name}: undeclared write slots: {', '.join(undeclared)}") - - def execute(self, slots: Mapping[str, Any]) -> dict[str, Any]: - """Validate the read allow-list, run the stage, validate the write allow-list.""" - self.validate_reads(slots) # pre-hook - updates = self.run(slots) or {} - if not isinstance(updates, dict): - raise SlotContractError(f"{self.name}: stage returned a non-dict update") - self.validate_writes(updates) # post-hook - return updates - - -class _Emitter(Protocol): - def emit(self, event_type: Any, stage: str | None, payload: dict[str, Any]) -> None: ... - - -class SlotPipeline: - """Runs slot processors in order, validating contracts and enforcing model-disable.""" - - def __init__(self, processors: list[SlotContractMixin]) -> None: - for proc in processors: - if not proc.skip_model: - raise SlotContractError(f"{proc.name}: deterministic stage must be model-disabled (skip_model=True)") - self._processors = processors - - @property - def processors(self) -> list[SlotContractMixin]: - return list(self._processors) - - def run(self, initial_slots: Mapping[str, Any], *, runtime: _Emitter | None = None) -> dict[str, Any]: - """Execute every stage; with ``runtime`` set, emit stage_start/stage_end lifecycle hooks.""" - slots: dict[str, Any] = dict(initial_slots) - for proc in self._processors: - if runtime is not None: - runtime.emit( - "stage_start", proc.name, {"reads": list(proc.reads_slots), "writes": list(proc.writes_slots), "skip_model": True} - ) - updates = proc.execute(slots) - slots.update(updates) - if runtime is not None: - runtime.emit("stage_end", proc.name, {"status": "succeeded", "wrote": sorted(updates)}) - return slots - - -__all__ = ["SlotContractError", "SlotContractMixin", "SlotPipeline"] diff --git a/src/openultrasast/stage_processors.py b/src/openultrasast/stage_processors.py deleted file mode 100644 index 298c203..0000000 --- a/src/openultrasast/stage_processors.py +++ /dev/null @@ -1,135 +0,0 @@ -"""Deterministic scan stages as model-disabled, slot-contracted processors (task 6.3). - -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; this -module is the deterministic, model-free remainder that ``test_gate`` pins. -""" - -from __future__ import annotations - -from collections.abc import Mapping -from pathlib import Path -from typing import Any - -from .findings import quick_scan_findings -from .mapping import analyze_entry_points, attach_reachability_hints -from .policy import load_policy -from .preprocess import preprocess_repository -from .rank import rank_targets -from .ruleset import DEFAULT_RULESET_DIR, load_ruleset -from .scoring import build_score_artifact -from .slot_contract import SlotContractMixin, SlotPipeline - - -class PreprocessProcessor(SlotContractMixin): - name = "preprocess" - reads_slots = ("root",) - writes_slots = ("snapshot", "targets") - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - snapshot, targets = preprocess_repository(slots["root"]) - return {"snapshot": snapshot, "targets": targets} - - -class EntryPointProcessor(SlotContractMixin): - name = "entry_point" - reads_slots = ("root", "targets") - writes_slots = ("targets", "entry_points") - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - entry_points = analyze_entry_points(slots["root"], slots["targets"]) - targets = attach_reachability_hints(slots["targets"], entry_points) - return {"targets": targets, "entry_points": entry_points} - - -class RankProcessor(SlotContractMixin): - name = "rank" - reads_slots = ("targets",) - writes_slots = ("rankings",) - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - return {"rankings": rank_targets(slots["targets"])} - - -class QuickFindingsProcessor(SlotContractMixin): - name = "quick_findings" - reads_slots = ("root", "targets", "rankings", "ruleset", "policy") - writes_slots = ("findings",) - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - findings = quick_scan_findings(slots["root"], slots["targets"], slots["rankings"], slots["ruleset"], slots["policy"]) - return {"findings": findings} - - -class PolicyScoringProcessor(SlotContractMixin): - name = "policy_scoring" - reads_slots = ("findings",) - writes_slots = ("reported_findings", "shadow_findings") - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - findings = slots["findings"] - shadow = [finding for finding in findings if finding.status == "shadow"] - reported = [finding for finding in findings if finding.status != "shadow"] - return {"reported_findings": reported, "shadow_findings": shadow} - - -class ScoreProcessor(SlotContractMixin): - name = "score" - reads_slots = ("reported_findings", "ruleset", "policy") - writes_slots = ("score",) - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - cwe_by_rule = {rule.rule_id: rule.cwe for rule in slots["ruleset"]} - reported = slots["reported_findings"] - rule_cwe_by_id = {f.finding_id: cwe_by_rule.get(f.finding_id.split(":", 1)[0], "") for f in reported} - return {"score": build_score_artifact(reported, rule_cwe_by_id, slots["policy"])} - - -def deterministic_processors() -> list[SlotContractMixin]: - """The ordered deterministic stages (preprocess → entry-point → rank → findings → policy → score).""" - return [ - PreprocessProcessor(), - EntryPointProcessor(), - RankProcessor(), - QuickFindingsProcessor(), - PolicyScoringProcessor(), - ScoreProcessor(), - ] - - -def build_deterministic_pipeline() -> SlotPipeline: - return SlotPipeline(deterministic_processors()) - - -def run_quick_pipeline( - root: Path | str, - *, - ruleset: Any = None, - policy: Any = None, - runtime: Any = None, -) -> dict[str, Any]: - """Run the deterministic quick-scan pipeline and return the final slot map. - - Defaults to the bundled ruleset and the vendored policy, which makes the - ``reported_findings`` slot byte-identical to ``findings.quick_scan_findings`` on - the same target. - """ - resolved_ruleset = ruleset if ruleset is not None else load_ruleset(DEFAULT_RULESET_DIR) - resolved_policy = policy if policy is not None else load_policy() - initial = {"root": Path(root), "ruleset": resolved_ruleset, "policy": resolved_policy} - return build_deterministic_pipeline().run(initial, runtime=runtime) - - -__all__ = [ - "EntryPointProcessor", - "PolicyScoringProcessor", - "PreprocessProcessor", - "QuickFindingsProcessor", - "RankProcessor", - "ScoreProcessor", - "build_deterministic_pipeline", - "deterministic_processors", - "run_quick_pipeline", -] diff --git a/tests/test_gate.py b/tests/test_gate.py index 6a7591b..1d9427a 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -134,8 +134,6 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = ( "openultrasast.ruleset.store", "openultrasast.improve", "openultrasast.improve.evolve", - "openultrasast.slot_contract", - "openultrasast.stage_processors", "openultrasast.fusion", "openultrasast.skills", "openultrasast.redaction", diff --git a/tests/test_model_calibrate.py b/tests/test_model_calibrate.py deleted file mode 100644 index fa666c6..0000000 --- a/tests/test_model_calibrate.py +++ /dev/null @@ -1,97 +0,0 @@ -"""model-grounded-detection task 5.1: the corpus calibrates the MODEL, not an LLM vote (Req 9). - -The inversion this feature rests on. Previously a pair the detector missed was a detector failure to be -averaged away over K runs. Here a pair the model cannot split is a *named gap in the model*, with the reason -recorded, because the model is deterministic: it will miss that pair the same way every time, and the only -thing that changes it is better modelling. - -`covered` therefore means the model reached a strictly higher rung on the vulnerable side than on its fix -- -not that it flagged something. Flagging both sides equally distinguishes nothing. -""" - -from __future__ import annotations - - -def _outcome(name, family="injection", slice_name="vibe-py", vuln="model_entailed", fixed="none", reason=""): # type: ignore[no-untyped-def] - from openultrasast.model.calibrate import PairOutcome - - return PairOutcome(pair=name, family=family, slice=slice_name, vuln_rung=vuln, fixed_rung=fixed, reason=reason) - - -def test_a_pair_the_model_splits_is_covered() -> None: - from openultrasast.model.calibrate import calibrate - - report = calibrate([_outcome("a")]) - assert report.covered == ("a",) - assert report.gaps == () - - -def test_a_pair_the_model_cannot_split_is_a_named_gap_not_a_miss() -> None: - """Req 9.2: the reason travels with the gap, because a deterministic model fails the same way every time.""" - from openultrasast.model.calibrate import calibrate - - report = calibrate([_outcome("b", vuln="none", fixed="none", reason="sink_not_in_table")]) - assert report.gaps and report.gaps[0].pair == "b" - assert report.gaps[0].reason == "sink_not_in_table" - assert report.covered == () - - -def test_flagging_both_sides_equally_is_a_gap_not_a_success() -> None: - """The model saw something on the fix too, so it distinguished nothing — that is the whole metric.""" - from openultrasast.model.calibrate import calibrate - - report = calibrate([_outcome("c", vuln="model_entailed", fixed="model_entailed")]) - assert report.covered == () - assert report.gaps[0].reason, "an unsplit pair must carry a reason even when none was supplied" - - -def test_gaps_are_grouped_by_family_with_their_reasons() -> None: - from openultrasast.model.calibrate import calibrate - - report = calibrate( - [ - _outcome("a", family="path", vuln="none", reason="sink_not_in_table"), - _outcome("b", family="path", vuln="none", reason="sink_not_in_table"), - _outcome("c", family="injection", vuln="none", reason="source_unmodelled"), - _outcome("d", family="injection"), - ] - ) - by_family = report.gaps_by_family() - assert set(by_family) == {"path", "injection"} - assert by_family["path"]["pairs"] == 2 - assert by_family["path"]["reasons"] == {"sink_not_in_table": 2} - assert by_family["injection"]["pairs"] == 1 - - -def test_the_report_is_per_slice_and_names_the_deciding_one() -> None: - """Req 9.3/11.1: the real-world slice decides; a development slice must not stand in for the corpus.""" - from openultrasast.model.calibrate import calibrate - - report = calibrate( - [ - _outcome("a", slice_name="vibe-py"), - _outcome("b", slice_name="vfc-js", vuln="none", reason="sink_not_in_table"), - ] - ) - assert set(report.per_slice) == {"vibe-py", "vfc-js"} - assert report.per_slice["vibe-py"]["covered"] == 1 - assert report.per_slice["vfc-js"]["gaps"] == 1 - - -def test_the_report_refuses_a_bare_aggregate() -> None: - from openultrasast.model.calibrate import calibrate - - text = calibrate([_outcome("a"), _outcome("b", slice_name="vfc-js", vuln="none")]).to_markdown() - assert "vibe-py" in text and "vfc-js" in text, "no headline without its per-slice rows" - assert "coverage" in text.lower() - - -def test_a_cve_recipe_yields_the_parent_and_fix_commits() -> None: - """Req 9.4: the harvest seam. The fix diff is the oracle; nothing is hand-labelled.""" - from openultrasast.model.calibrate import harvest_recipe - - recipe = harvest_recipe({"repo": "org/proj", "fix_commit": "abc123", "relpath": "app.py", "function": "run"}) - assert recipe.repo == "org/proj" - assert recipe.fix_commit == "abc123" - assert recipe.parent_ref == "abc123^", "the vulnerable side is the fix commit's parent, by construction" - assert recipe.relpath == "app.py" and recipe.function == "run" diff --git a/tests/test_model_execution.py b/tests/test_model_execution.py deleted file mode 100644 index e327f04..0000000 --- a/tests/test_model_execution.py +++ /dev/null @@ -1,116 +0,0 @@ -"""model-grounded-detection task 6.1: the execution tier is deferred, adopted, and narrow (Req 10). - -The rung this project does NOT build. Clearwing already runs sandboxes, container pools, sanitizer images and -PoC stability, and this session's own history says reimplementing that class of machinery is where we go -wrong. So `execution_confirmed` exists as a rung, the seam calls an adopted tool, and the eligibility rule is -deliberately narrow: only a finding the static model could not decide, in a family a static model genuinely -cannot arbitrate. - -Everything here runs with Clearwing absent, which is the normal case. -""" - -from __future__ import annotations - -import pytest - - -def _finding(family="memory", rung="suspicion"): # type: ignore[no-untyped-def] - from openultrasast.model.execution import Candidate - - return Candidate(finding_id="f1", family=family, rung=rung, path="a.c", line=10) - - -def test_only_a_family_a_static_model_cannot_arbitrate_is_eligible() -> None: - """Req 10.3: no web/logic class depends on this tier. They have taint, dominance and constant abstraction.""" - from openultrasast.model.execution import eligible - - assert eligible(_finding(family="memory")) is True - for family in ("injection", "access_control", "path", "config_secrets", "output_encoding"): - assert eligible(_finding(family=family)) is False, f"{family} must be arbitrated statically" - - -def test_a_finding_the_model_already_decided_is_not_escalated() -> None: - """Req 10.1: the tier is reached only where the model could not decide -- never to double-check itself.""" - from openultrasast.model.execution import eligible - - assert eligible(_finding(rung="suspicion")) is True - assert eligible(_finding(rung="model_entailed")) is False - assert eligible(_finding(rung="model_corroborated")) is False - - -def test_without_clearwing_the_rung_is_never_reached_and_the_reason_is_recorded(monkeypatch: pytest.MonkeyPatch) -> None: - """Req 11.3: an absent capability degrades to a recorded reason, never a traceback and never a verdict.""" - from openultrasast.model.execution import confirm, resolve_execution_backend - - monkeypatch.setenv("OPENULTRASAST_CLEARWING", "off") - backend = resolve_execution_backend() - assert backend.available() is False - outcome = confirm(_finding(), backend=backend) - assert outcome.rung is None, "no engine, no rung" - assert "clearwing_unavailable" in outcome.reason - - -def test_an_ineligible_finding_is_not_sent_to_the_sandbox_even_when_one_exists() -> None: - from openultrasast.model.execution import confirm - - class _Backend: - def __init__(self) -> None: - self.calls = 0 - - def available(self) -> bool: - return True - - def reproduce(self, candidate: object) -> bool: - self.calls += 1 - return True - - backend = _Backend() - outcome = confirm(_finding(family="injection"), backend=backend) - assert outcome.rung is None and backend.calls == 0 - assert "not_eligible" in outcome.reason - - -def test_a_reproduced_candidate_reaches_execution_confirmed() -> None: - from openultrasast.model.execution import confirm - from openultrasast.model.ladder import Rung - - class _Backend: - def available(self) -> bool: - return True - - def reproduce(self, candidate: object) -> bool: - return True - - assert confirm(_finding(), backend=_Backend()).rung is Rung.EXECUTION_CONFIRMED - - -def test_a_candidate_that_does_not_reproduce_stays_where_it_was() -> None: - """Not reproducing is not a refutation: the sandbox failing to trigger a bug does not prove its absence.""" - from openultrasast.model.execution import confirm - - class _Backend: - def available(self) -> bool: - return True - - def reproduce(self, candidate: object) -> bool: - return False - - outcome = confirm(_finding(), backend=_Backend()) - assert outcome.rung is None - assert "not_reproduced" in outcome.reason and "refut" not in outcome.reason.lower() - - -def test_the_seam_reimplements_no_sandbox_machinery() -> None: - """Req 10.2, asserted structurally. The one thing this module must not grow is a sandbox.""" - import ast - from pathlib import Path - - source = Path("src/openultrasast/model/execution.py").read_text() - tree = ast.parse(source) - imported = {n.module.split(".")[0] for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module} - imported |= {a.name.split(".")[0] for n in ast.walk(tree) if isinstance(n, ast.Import) for a in n.names} - for banned in ("docker", "podman", "subprocess", "container", "sandbox"): - assert banned not in imported, f"the execution tier must adopt, not build: {banned}" - names = {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)} - for banned in ("build_image", "run_container", "compile_target", "apply_patch"): - assert banned not in names, f"{banned} belongs to the adopted tool, not here" diff --git a/tests/test_model_judge.py b/tests/test_model_judge.py deleted file mode 100644 index 5b28c97..0000000 --- a/tests/test_model_judge.py +++ /dev/null @@ -1,163 +0,0 @@ -"""model-grounded-detection task 3.2: the LLM proposes, the model disposes (Req 8). - -The judge is where the prior architecture's whole premise is inverted. There is no K-run averaging, no noise -floor and no acceptance rule; there is one bounded, typed question per candidate, and the answer is CHECKED -against the CPG rather than voted on. The three outcomes are exhaustive: - - the model entails it -> report it, and never call the model at all - the model contradicts it -> drop the claim, and record the contradiction - the model cannot decide -> report `suspicion`, honestly - -Every test here runs with no network and a scripted client. -""" - -from __future__ import annotations - -from pathlib import Path - - -def _spec(): # type: ignore[no-untyped-def] - from openultrasast.model.specs import TaintSpec - - return TaintSpec(family="injection", language="python", sources=("request.args",), sinks=("os.system",), sanitizers=("escape",)) - - -def _cpg(rows): # type: ignore[no-untyped-def] - from openultrasast.cpg.backend import CpgResult - - return CpgResult(cpg_path=Path("cpg.bin"), run=lambda q, p: rows) - - -ENTAILING_ROW = [ - {"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": False, "length": 3} -] - - -class _Client: - """A scripted chat client that records whether it was called at all.""" - - def __init__(self, answer: str = '{"vulnerable": true, "family": "injection"}') -> None: - self.calls: list[object] = [] - self._answer = answer - - def complete(self, **kwargs: object): # type: ignore[no-untyped-def] - from openultrasast.tool_hunter import ChatResponse - - self.calls.append(kwargs) - return ChatResponse(content=self._answer, tool_calls=[], reasoning=None) - - -def test_an_entailed_candidate_costs_no_model_call() -> None: - """The point of the arbiter: where the graph decides, the LLM is not consulted at all.""" - from openultrasast.model.judge import judge - from openultrasast.model.ladder import Rung - - client = _Client() - answer = judge(_cpg(ENTAILING_ROW), _spec(), function="run", client=client, model="m") - assert answer.rung is Rung.ENTAILED - assert client.calls == [], "an entailed candidate must not reach the model" - - -def test_a_claim_the_model_contradicts_is_dropped_with_its_reason() -> None: - """Req 8.2. The LLM says vulnerable; the CPG shows no flow reaches the sink. The graph wins.""" - from openultrasast.model.judge import judge - from openultrasast.model.ladder import Rung - - client = _Client('{"vulnerable": true, "family": "injection"}') - answer = judge(_cpg([]), _spec(), function="run", client=client, model="m") - assert answer.rung is Rung.SUSPICION - assert answer.contradiction, "a dropped claim must say why it was dropped" - assert "no flow" in answer.contradiction.lower() or "not reach" in answer.contradiction.lower() - - -def test_a_claim_the_model_supports_advances_to_corroborated() -> None: - """Req 8.3: a sanitized path is a real flow whose sufficiency only the LLM can judge.""" - from openultrasast.model.judge import judge - from openultrasast.model.ladder import Rung - - rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}] - client = _Client('{"vulnerable": true, "family": "injection"}') - answer = judge(_cpg(rows), _spec(), function="run", client=client, model="m") - assert answer.rung is Rung.CORROBORATED - assert len(client.calls) == 1, "exactly one bounded question, never a loop" - - -def test_the_llm_declining_leaves_a_corroborated_flow_at_suspicion() -> None: - """The model saw a flow but the LLM says the sanitizer suffices; the claim does not advance.""" - from openultrasast.model.judge import judge - from openultrasast.model.ladder import Rung - - rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}] - client = _Client('{"vulnerable": false, "family": "injection"}') - answer = judge(_cpg(rows), _spec(), function="run", client=client, model="m") - assert answer.rung is Rung.SUSPICION - - -def test_without_a_client_the_verdict_still_stands_on_its_own() -> None: - """Req 11.3: the endpoint is optional. The model layer does not need it to entail.""" - from openultrasast.model.judge import judge - from openultrasast.model.ladder import Rung - - assert judge(_cpg(ENTAILING_ROW), _spec(), function="run", client=None, model="").rung is Rung.ENTAILED - sanitized = [ - {"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3} - ] - answer = judge(_cpg(sanitized), _spec(), function="run", client=None, model="") - assert answer.rung is Rung.SUSPICION and "endpoint" in answer.contradiction.lower() - - -def test_the_question_is_bounded_typed_and_offers_no_tools() -> None: - """Req 8.1: the LLM answers one question about a site it was handed; it never searches for the site.""" - from openultrasast.model.judge import judge - - rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}] - client = _Client() - judge(_cpg(rows), _spec(), function="run", client=client, model="m") - call = client.calls[0] - assert call["tools"] == [], "no tools: the candidate is given, not searched for" - assert call["json_object"] is True, "a typed answer, not prose" - text = " ".join(str(m.get("content", "")) for m in call["messages"]) # type: ignore[union-attr] - assert "os.system(cmd)" in text and "request.args['c']" in text, "the question carries the model's own evidence" - - -def test_the_prompt_is_redacted() -> None: - """Req 11.4: every prompt passes through redaction.""" - from openultrasast.model.judge import judge - - rows = [ - { - "sink": "connect(pw)", - "sinkLine": "4", - "sinkMethod": "run", - "source": "AKIA" + "IOSFODNN7" + "EXAMPLE", - "sanitized": True, - "length": 2, - } - ] - client = _Client() - judge(_cpg(rows), _spec(), function="run", client=client, model="m") - text = " ".join(str(m.get("content", "")) for m in client.calls[0]["messages"]) # type: ignore[index,union-attr] - assert "AKIAIOSFODNN7EXAMPLE" not in text - - -def test_there_is_no_k_run_averaging_anywhere_in_the_judge() -> None: - """Req 8.5, asserted structurally: the noise architecture must not grow back.""" - import ast - - source = Path("src/openultrasast/model/judge.py").read_text() - tree = ast.parse(source) - - # Check CODE, not prose: the docstring is allowed to name the machinery it replaced. - names = ( - {node.id for node in ast.walk(tree) if isinstance(node, ast.Name)} - | {node.attr for node in ast.walk(tree) if isinstance(node, ast.Attribute)} - | {node.name for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)} - ) - for banned in ("k_runs", "majority", "vote", "votes", "average", "mean", "acceptance", "floor", "budget"): - assert banned not in names, f"{banned} has no place in the judge" - - # No loop over runs anywhere in the module. - assert not [n for n in ast.walk(tree) if isinstance(n, ast.For | ast.While)], "the judge asks once" - - imported = {n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module} - assert not any("rounds" in m or "acceptance" in m for m in imported) diff --git a/tests/test_module_audit.py b/tests/test_module_audit.py index 0b2ab73..a860ce1 100644 --- a/tests/test_module_audit.py +++ b/tests/test_module_audit.py @@ -11,7 +11,6 @@ SRC = Path("src/openultrasast") MANIFEST = Path("benchmarks/measurements/2026-09-08-module-audit.json") -@pytest.mark.xfail(strict=True, reason="five dead modules the fixed audit found; removed in the next commit") def test_no_module_in_the_tree_is_orphaned() -> None: from openultrasast.model.audit import audit @@ -37,6 +36,8 @@ def test_the_named_subsystems_each_carry_an_explicit_decision() -> None: 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) + # The five the reachability-first audit found orphaned are gone too (legacy cleanup 2026-10-02). + assert not {"model.judge", "model.execution", "model.calibrate", "stage_processors", "slot_contract"} & set(rows) def test_the_manifest_matches_the_tree() -> None: @@ -47,7 +48,7 @@ def test_the_manifest_matches_the_tree() -> None: committed = json.loads(MANIFEST.read_text()) current = to_dict(audit(SRC)) assert committed["counts"] == current["counts"], "the tree drifted from its committed audit" - assert committed["orphaned"] == current["orphaned"] + assert committed["orphaned"] == current["orphaned"] == [] def _tree(root: Path, files: dict[str, str]) -> Path: diff --git a/tests/test_slot_contract.py b/tests/test_slot_contract.py deleted file mode 100644 index f328bee..0000000 --- a/tests/test_slot_contract.py +++ /dev/null @@ -1,108 +0,0 @@ -"""Model-disabled, slot-contracted deterministic stages (Phase 3 task 6.3). - -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, and lifecycle hooks fire. -""" - -from collections.abc import Mapping -from pathlib import Path -from typing import Any - -import pytest - -from openultrasast.gate import _quick_scan -from openultrasast.slot_contract import SlotContractError, SlotContractMixin, SlotPipeline -from openultrasast.stage_processors import ( - build_deterministic_pipeline, - deterministic_processors, - run_quick_pipeline, -) - - -class _GoodStage(SlotContractMixin): - name = "good" - reads_slots = ("a",) - writes_slots = ("b",) - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - return {"b": slots["a"] + 1} - - -class _UndeclaredWriteStage(SlotContractMixin): - name = "bad_write" - reads_slots = () - writes_slots = ("declared",) - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - return {"undeclared": 1} - - -class _MissingReadStage(SlotContractMixin): - name = "bad_read" - reads_slots = ("required",) - writes_slots = () - - def run(self, slots: Mapping[str, Any]) -> dict[str, Any]: - return {} - - -# ---- slot contract enforcement (1.5) ---------------------------------------- - - -def test_declared_slots_pass() -> None: - assert _GoodStage().execute({"a": 1}) == {"b": 2} - - -def test_undeclared_write_is_rejected() -> None: - with pytest.raises(SlotContractError, match="undeclared write slots: undeclared"): - _UndeclaredWriteStage().execute({}) - - -def test_missing_read_is_rejected() -> None: - with pytest.raises(SlotContractError, match="missing read slots: required"): - _MissingReadStage().execute({}) - - -# ---- model-disabled invariant (1.6) ----------------------------------------- - - -def test_all_deterministic_stages_are_model_disabled() -> None: - assert all(proc.skip_model for proc in deterministic_processors()) - - -def test_pipeline_rejects_a_model_enabled_stage() -> None: - enabled = _GoodStage() - enabled.skip_model = False - with pytest.raises(SlotContractError, match="model-disabled"): - SlotPipeline([enabled]) - - -# ---- parity with the deterministic quick-scan path -------------------------- - - -def test_pipeline_findings_match_quick_scan_baseline() -> None: - target = Path("benchmarks/fixtures/python-vulnerable") - slots = run_quick_pipeline(target) - assert slots["reported_findings"] == _quick_scan(target) - assert "score" in slots # the score slot is produced too - - -# ---- lifecycle hooks fire per stage ----------------------------------------- - - -class _Recorder: - def __init__(self) -> None: - self.events: list[tuple[str, str | None]] = [] - - def emit(self, event_type: Any, stage: str | None, payload: dict[str, Any]) -> None: - self.events.append((event_type, stage)) - - -def test_pipeline_emits_lifecycle_hooks_per_stage() -> None: - recorder = _Recorder() - run_quick_pipeline(Path("benchmarks/fixtures/python-vulnerable"), runtime=recorder) - names = [proc.name for proc in build_deterministic_pipeline().processors] - for name in names: - assert ("stage_start", name) in recorder.events - assert ("stage_end", name) in recorder.events