feat(scan): stream or subsample datasets; avoid duplicate HarmBench loads

[davidberenstein1957]

Feature Request

Stop fully parsing large HF/local JSONL datasets only to keep a small random subset, and avoid loading HarmBench twice on the default vulnerability path (plain dataset + GCG, and framing generators that also pull HarmBench).

Motivation

Default generators materialize hundreds of lines (DNA ~939, HarmBench ~400) then keep ~20. HarmBench is parsed repeatedly under different generators.

Implementation plan

1. Loader API (internal): e.g. iter_jsonl(path) -> Iterator[dict] + reservoir_sample(iterable, k, rng) -> list.
2. Replace “read all → rng.choice” in HF / local dataset base classes with sample-while-streaming (language filter applied per row or via cheap prefilter).
3. Shared HarmBench cache at process/suite scope: load once, pass behaviors into GCG / framing transforms.
4. Keep public ScenarioGenerator.generate(...) signatures stable.
5. Tests: fixture JSONL with 500 lines, max_scenarios=20 → 20 scenarios; spy/counter proves full materialization did not occur (or peak list length ≤ k + ε); two generators sharing HarmBench → one parse.

Public API (unchanged usage)

from giskard.scan import vulnerability_scan, generate_suite

# Call sites stay the same — only I/O cost changes
suite = await generate_suite(
    description="...",
    languages=["en"],
    max_scenarios=20,
    seed=42,
)

Internal sketch

def reservoir_sample[T](rows: Iterable[T], k: int, rng: random.Random) -> list[T]:
    """Algorithm R — O(k) memory for a full stream."""
    ...

# before
rows = list(parse_all_jsonl(path))          # len == 939
picked = rng.sample(rows, k=max_scenarios)

# after
picked = reservoir_sample(parse_jsonl(path), k=max_scenarios, rng=rng)

Shared HarmBench sketch

# once per generate_suite / process
behaviors = load_harmbench_behaviors(languages=["en"])  # cached

GCGInjectionScenarioGenerator(behaviors=behaviors)
# Framing generators reuse the same list instead of re-downloading

Test assertion idea

def test_reservoir_does_not_materialize_all(monkeypatch, tmp_path):
    path = write_jsonl(tmp_path, n=500)
    calls = {"rows_held": 0}

    def fake_list(xs):
        materialised = list(xs)
        calls["rows_held"] = max(calls["rows_held"], len(materialised))
        return materialised

    # assert peak buffer ~ max_scenarios, not 500
    scenarios = await gen.generate(..., max_scenarios=20)
    assert len(scenarios) == 20
    assert calls["rows_held"] <= 20  # or whatever the implementation guarantees

Acceptance criteria

• Large JSONL + small max_scenarios does not require a full in-memory corpus only to discard rows
• Default vulnerability path does not fully parse HarmBench twice for independent generators
• Fixture tests for sample size + load sharing; seed behavior documented

Out of scope

• Changing default registry membership
• Scan presets / concurrency (separate)

[linear-code]

OSS-106