FEAT text adaptive scenario

[hannahwestra25]

Add Adaptive Scenario Framework with TextAdaptive

Summary

Introduces an adaptive scenario framework that picks attack techniques per-objective using an epsilon-greedy bandit informed by observed success rates, rather than running every selected technique against every objective. Concentrates spend on techniques that actually work against the target, and stops early on first success.

Adds:

• AdaptiveScenario — modality-agnostic base class
• TextAdaptive — concrete text-attack subclass
• TechniqueSelector Protocol + EpsilonGreedyTechniqueSelector — pluggable selector with Laplace-smoothed estimates and pooled cross-context backoff
• AdaptiveDispatchAttack — per-dataset dispatch strategy with per-call seed-group routing
• Walkthrough notebook + .py doc
• Unit tests (64 tests across selector, protocol, dispatcher, and scenario)

Motivation

Static scenarios are O(techniques × objectives): every technique runs against every objective regardless of whether earlier attempts already succeeded or whether the technique is known to be ineffective against the target. For evaluation runs with many techniques and many objectives, this wastes spend on combinations that aren't informative.

Adaptive scenarios reduce this to O(max_attempts × objectives) by:

• learning from observed outcomes,
• exploiting techniques that work on the target,
• still exploring (with probability epsilon) so the table doesn't collapse onto a single technique prematurely,
• stopping per-objective on first success.

How it works

For each objective the dispatcher loops up to max_attempts_per_objective times:

1. Select — with probability epsilon pick a random technique, otherwise pick the one with the highest Laplace-smoothed success estimate (s + 1) / (n + 1). Cells with fewer than pool_threshold local observations fall back to the technique's pooled rate across all contexts (cold-start handling). Each decision derives a per-decision RNG from SHA-256(random_seed|context|decision_key) for resume-safe reproducibility.
2. Execute — run the chosen technique against the seed group (read from AdaptiveDispatchParams.seed_group), merging the technique's seed_technique if it declares one. Techniques incompatible with the current seed group are filtered per-call.
3. Record — update the selector's (context, technique) → (successes, attempts) table and stop early on success.

The selector is shared by reference across all dispatchers in a scenario run, so learning accumulates globally. The per-call context key is derived by a ContextExtractor; global_context (default) shares one table across all objectives, harm_category_context partitions by harm category.

Public API

from pyrit.scenario.scenarios.adaptive import (
    TextAdaptive,
    EpsilonGreedyTechniqueSelector,
    harm_category_context,
)

# Basic — uses default epsilon-greedy selector
scenario = TextAdaptive()
await scenario.initialize_async(objective_target=target)
result = await scenario.run_async()

# Tuned — custom selector + per-category learning
scenario = TextAdaptive(
    selector=EpsilonGreedyTechniqueSelector(epsilon=0.3, random_seed=42),
    context_extractor=harm_category_context,
)
scenario.set_params_from_args(args={"max_attempts_per_objective": 5})
await scenario.initialize_async(objective_target=target)
result = await scenario.run_async()

Adaptive scenarios are also resumable — pass scenario_result_id="..." to the constructor and prior dispatch trails are replayed into the selector before the remaining objectives run.

Notes

• BASELINE_ATTACK_POLICY = Enabled — prompt_sending is excluded from the adaptive technique pool and runs as the baseline comparison instead. This separates "what does the target do unprovoked" (baseline) from "what adversarial moves help" (adaptive techniques).
• Per-dataset atomic attacks — one AtomicAttack per dataset carrying all seed groups, with per-call seed-group routing via AdaptiveDispatchParams. Per-call compatibility filtering happens inside the dispatcher.
• Selector as constructor kwarg — selector: TechniqueSelector | None on the scenario. When None (default), an EpsilonGreedyTechniqueSelector is created with default settings. Selector-specific params (epsilon, pool_threshold, random_seed) live on the selector, not the scenario. max_attempts_per_objective is a scenario parameter via supported_parameters().
• Resume rehydration — queries get_attack_results(scenario_result_id=...) and filters by attribution_data["parent_collection"] to replay prior dispatch trails via record_outcome. Already-completed atomics are skipped by the base Scenario resume path.
• Two-row persistence per success — the inner technique persists its raw AttackResult via its own post-execute hook; the dispatcher returns a replace-based copy with a fresh attack_result_id/timestamp and the adaptive trail stamped onto metadata. Both rows share conversation_id.
• Thread safety — EpsilonGreedyTechniqueSelector guards its counts table with a threading.Lock so individual select / record_outcome operations are atomic.

Files

Area	File
Base scenario	pyrit/scenario/scenarios/adaptive/adaptive_scenario.py
Text subclass	pyrit/scenario/scenarios/adaptive/text_adaptive.py
Selector protocol + context extractors	pyrit/scenario/scenarios/adaptive/selectors/protocol.py
Epsilon-greedy selector	pyrit/scenario/scenarios/adaptive/selectors/epsilon_greedy.py
Per-dataset dispatcher	pyrit/scenario/scenarios/adaptive/dispatcher.py
Package wiring	pyrit/scenario/scenarios/adaptive/__init__.py, pyrit/scenario/scenarios/adaptive/selectors/__init__.py
Walkthrough	doc/code/scenarios/3_adaptive_scenarios.py, doc/code/scenarios/3_adaptive_scenarios.ipynb
Tests	tests/unit/scenario/scenarios/adaptive/test_epsilon_greedy.py, tests/unit/scenario/scenarios/adaptive/test_protocol.py, tests/unit/scenario/scenarios/adaptive/test_dispatcher.py, tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py

Testing

pytest tests/unit/scenario/scenarios/adaptive/ — 64 tests pass. Coverage includes:

• selector exploration / exploitation / cold-start / pooled backoff / concurrent record_outcome
• protocol conformance, context extractor coverage
• dispatcher early-stop, max-attempts retry, label propagation, context routing, fresh-result invariant, per-call compatibility filtering
• scenario per-dataset atomics, shared selector, harm-category partitioning, seed-technique filtering, resume rehydration, baseline-policy enforcement, params via supported_parameters()

[rlundeen2]

Looking great!

[rlundeen2]

I might reformat this can call it "TechniqueSelector" or technique_selector.py

# returns the technique_id; calling select_async with the same SelectorContext twice should yield the same answer
async def select_async(
    *,
    technique_identifiers: Sequence[str],
    context: SelectorContext,
    number_to_get: int = 1,
) -> Sequence[str]:
    """Returns techniques in priority order (try first, try second).
    Returns up to number_to_get; fewer if not enough techniques are available.
    """

@dataclass(frozen=True)
class SelectorContext:
    scenario_result_id: str
    scenario_class_name: str
    objective: str
    attempt_index: int
    harm_category: str | None = None
    memory_labels # in case the selector wants to make use 

I don't think it needs to record results within the class; they're already in memory. We probably don't want to enforce "one at a time" logic or have custom resume logic. WE want to use all memory available for many of these, and that can vary. If we wanted to scope to a single scenario run, we could use the SelectorContext to limit it.

[rlundeen2]

This algorithm is sound for in-memory bandit-style selection, but the in-memory counts are a weakness in two ways: (a) state is lost between runs, and (b) we only learn from this scenario's own attempts. With thousands of historical AttackResults in memory, an analytics-backed selector can be much more powerful. I would

1. Move ASR aggregation to pyrit/analytics/result_analysis.py add compute_technique_success_rates(memory, *, technique_hashes, scope) -> dict[str, AttackStats] next to the existing analyze_results. Default scope is all-history filtered by technique identity; pass an AnalyticsScope to restrict (e.g. to this scenario_class, this scenario_result_id, a harm category, or to exclude in-flight runs).
2. Identify techniques using AttackTechnique eval hashes; but in the future we may want to identify parts of techniques
3. See my comment on "protocol" about how I think AttackTechnique should look. For Epsilon greedy I think pool threshold and epsilon make sense, but we likely don't need count, lock, etc

---

An example of what result_analytics could look like

Python
# In pyrit/analytics/result_analysis.py (or technique_analytics.py)

async def compute_technique_success_rates(
    memory: MemoryInterface,
    *,
    technique_hashes: Sequence[str],
    scope: AnalyticsScope | None = None,
) -> dict[str, AttackStats]:
    """ASR per technique, filtered by AnalyticsScope (scenario_class,
    scenario_result_id, harm_categories, since, labels)."""

[rlundeen2]

I like how you group these by dataset initially.

If you take the TechniqueSelector nudge (stateless, queries memory), I think a lot of this simplifies. At construction time per objective, pre-select all K techniques to try:

chosen = await selector.select_async(
    technique_identifiers=all_techniques,
    context=context,
    number_to_get=max_attempts_per_objective,
)
# chosen is a Sequence[str] in priority order
for technique in chosen:
    result = await run(technique)
    if result.outcome == SUCCESS:
        break

No per-attempt re-selection — within one objective the analytics signal barely moves, so K queries collapse to 1. Resume: just call select_async again with number_to_get = K - already_executed; the selector is deterministic-given-memory so the new picks are sensible relative to whatever's already in the DB.

[rlundeen2]

Once the TechniqueSelector redesign lands (stateless, queries memory) and selection moves to batch-at-construction on AdaptiveScenario, this file shrinks and is super simplified — probably to ~50 lines. Sketching what could work

We could have a new attack that executes attacks sequentially like this

# pyrit/executor/attack/compound/strategy_sequence.py

class StopPolicy(StrEnum):
    ON_SUCCESS = "on_success"
    ON_SUCCESS_OR_ERROR = "on_success_or_error"
    RUN_ALL = "run_all"


class StrategySequenceAttack(AttackStrategy):
    """Runs a sequence of attack strategies against one objective,
    stopping per ``stop_policy``. Each strategy runs as a real
    AtomicAttack so scenario attribution lands correctly."""

    def __init__(
        self,
        *,
        strategies: Sequence[AttackStrategy],
        stop_policy: StopPolicy = StopPolicy.ON_SUCCESS,
    ) -> None:
        ...

Then you can call that with the selected techniques