From b9caeeb835b6bf015ccfaeb1b2c521f02f65131c Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 22 Jun 2026 18:39:09 +0200 Subject: [PATCH] =?UTF-8?q?feat(slots):=20Phase=203=20task=206.3=20?= =?UTF-8?q?=E2=80=94=20model-disabled,=20slot-contracted=20deterministic?= =?UTF-8?q?=20stages?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Express the deterministic scan stages as model-disabled processors carrying work in named slots under a read/write allow-list validated on every lifecycle hook. - slot_contract.py (zero-dep): SlotContractError, SlotContractMixin (reads_slots/ writes_slots/skip_model=True, validates reads pre-hook and writes post-hook), SlotPipeline (runs stages in order, enforces the model-disabled invariant, emits stage_start/stage_end lifecycle hooks). Preserves the ProcessorSpec.reads/writes provenance discipline without forcing deterministic stages through HarnessX's agent loop ("bridge, do not map"). - stage_processors.py: Preprocess/EntryPoint/Rank/QuickFindings/PolicyScoring/Score as skip_model slot processors; run_quick_pipeline is byte-identical to the quick path. host_under_harnessx is the lazy, capability-gated bridge that disables the model on HarnessX's BeforeModelEvent.skip_model (proven against the real API when the extra is present). - 8.4 zero-dep guard hardened: now hermetic (subprocess) and covers the two new modules; the module-level import scan flags only column-0 imports so the intended lazy import inside host_under_harnessx is allowed. The sync HarnessRuntime stays the top-level driver (design "stays through Phase 4"); these slot processors are the model-disabled representation, ready for Phase 5. tests/test_slot_contract.py (8 tests). 153 passed, 3 skipped; ruff + mypy clean; detection gate PASS. Co-Authored-By: Claude Opus 4.8 --- .../spec.json | 2 +- .../harnessx-self-improving-rulesets/tasks.md | 5 +- src/openultrasast/slot_contract.py | 93 ++++++++++ src/openultrasast/stage_processors.py | 170 ++++++++++++++++++ tests/test_gate.py | 20 +-- tests/test_slot_contract.py | 133 ++++++++++++++ 6 files changed, 410 insertions(+), 13 deletions(-) create mode 100644 src/openultrasast/slot_contract.py create mode 100644 src/openultrasast/stage_processors.py create mode 100644 tests/test_slot_contract.py diff --git a/.kiro/specs/harnessx-self-improving-rulesets/spec.json b/.kiro/specs/harnessx-self-improving-rulesets/spec.json index 41f2268..38b2cbe 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/spec.json +++ b/.kiro/specs/harnessx-self-improving-rulesets/spec.json @@ -1,7 +1,7 @@ { "feature_name": "harnessx-self-improving-rulesets", "created_at": "2026-06-22T10:07:18Z", - "updated_at": "2026-06-22T16:20:00Z", + "updated_at": "2026-06-22T16:45:00Z", "language": "en", "phase": "implementation", "approvals": { diff --git a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md index d5a74ea..df59b9d 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md +++ b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md @@ -138,7 +138,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: Packaging Configuration_ - _Depends: 5.1_ -- [ ] 6. Adopt HarnessX at the hunter loop and verifier +- [x] 6. Adopt HarnessX at the hunter loop and verifier - [x] 6.1 Run the hunter pool stage on a real HarnessX agent loop - Run the hunter pool stage on a real HarnessX run loop with the control, tool, and observability processors registered in the hunter sub-harness, invoked from the existing hunter-pool stage callback. - Confine all asynchronous execution to this single stage so the deterministic stages remain synchronous. @@ -154,10 +154,11 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: HxScanOrchestrator_ - _Depends: 6.1_ -- [ ] 6.3 (P) Run deterministic stages as model-disabled HarnessX processors with slot contracts +- [x] 6.3 (P) Run deterministic stages as model-disabled HarnessX processors with slot contracts - Run each deterministic stage under HarnessX with model invocation disabled so no deterministic stage incurs a model call. - Preserve each stage's declared read/write slot allow-list and validate slot access on every lifecycle hook. - Observable completion: deterministic stages execute under HarnessX with no model calls and a slot-contract violation is raised on undeclared slot access, proven by a slot-contract test. + - Implemented zero-dep in `slot_contract.py` (`SlotContractMixin`/`SlotPipeline`, validates reads/writes per lifecycle hook, enforces `skip_model`) + `stage_processors.py` (the deterministic stages as `skip_model` slot processors; `run_quick_pipeline` is byte-identical to the quick path). Per "bridge, do not map", the sync `HarnessRuntime` stays the top-level driver and these slot processors are the model-disabled representation; `host_under_harnessx` is the lazy, capability-gated bridge that disables the model on HarnessX's `BeforeModelEvent.skip_model`. Module names differ from the design's `harness/{slot_contract,processors}.py` because `harness` is a module, not a package. Tests: `tests/test_slot_contract.py`. - _Requirements: 1.5, 1.6_ - _Boundary: HxScanOrchestrator_ - _Depends: 6.1_ diff --git a/src/openultrasast/slot_contract.py b/src/openultrasast/slot_contract.py new file mode 100644 index 0000000..ca41b9a --- /dev/null +++ b/src/openultrasast/slot_contract.py @@ -0,0 +1,93 @@ +"""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) — without forcing the deterministic stages through HarnessX's +agent loop ("bridge, do not map", per the spec design). + +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. +""" + +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 new file mode 100644 index 0000000..744cbb5 --- /dev/null +++ b/src/openultrasast/stage_processors.py @@ -0,0 +1,170 @@ +"""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 — +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``. +""" + +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) + + +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", + "PreprocessProcessor", + "QuickFindingsProcessor", + "RankProcessor", + "ScoreProcessor", + "build_deterministic_pipeline", + "deterministic_processors", + "host_under_harnessx", + "run_quick_pipeline", +] diff --git a/tests/test_gate.py b/tests/test_gate.py index b4ea009..5fdba97 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -6,9 +6,7 @@ 8.4 — the governance/scoring/benchmark planes never import the HarnessX extra. """ -import importlib import json -import sys from pathlib import Path import pytest @@ -146,17 +144,20 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = ( "openultrasast.ruleset.store", "openultrasast.improve", "openultrasast.improve.evolve", + "openultrasast.slot_contract", + "openultrasast.stage_processors", ) -def test_governance_scoring_benchmark_planes_never_import_harnessx() -> None: - for module in GOVERNANCE_SCORING_BENCHMARK_PLANES: - importlib.import_module(module) - leaked = sorted(name for name in sys.modules if name == "harnessx" or name.startswith("harnessx.")) - assert leaked == [], f"governance/scoring/benchmark planes leaked HarnessX imports: {leaked}" +def test_governance_scoring_benchmark_planes_never_import_harnessx(assert_cold_of_harnessx) -> None: # type: ignore[no-untyped-def] + # Hermetic: a fresh interpreter imports every plane and must pull no harnessx. + imports = "\n".join(f"import {module}" for module in GOVERNANCE_SCORING_BENCHMARK_PLANES) + assert_cold_of_harnessx(imports) def test_planes_have_no_module_level_harnessx_import() -> None: + # Flags only column-0 (module-level) imports; an indented lazy import inside a + # capability-guarded function (e.g. stage_processors.host_under_harnessx) is fine. src = Path("src/openultrasast") offenders = [] for module in GOVERNANCE_SCORING_BENCHMARK_PLANES: @@ -166,7 +167,6 @@ def test_planes_have_no_module_level_harnessx_import() -> None: if path is None: continue for line in path.read_text().splitlines(): - stripped = line.strip() - if stripped.startswith(("import harnessx", "from harnessx")): - offenders.append(f"{path}: {stripped}") + if line.startswith(("import harnessx", "from harnessx")): + offenders.append(f"{path}: {line}") assert offenders == [], f"plane modules import HarnessX at module scope: {offenders}" diff --git a/tests/test_slot_contract.py b/tests/test_slot_contract.py new file mode 100644 index 0000000..a44d4fa --- /dev/null +++ b/tests/test_slot_contract.py @@ -0,0 +1,133 @@ +"""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, lifecycle hooks fire, and +(when the extra is present) a stage can run under HarnessX with the model disabled. +""" + +from collections.abc import Mapping +from pathlib import Path +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, +) + + +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 + + +# ---- 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