MAINT: Rapid response Scenario

[rlundeen2]

This PR refactors ContentHarms into RapidResponse (with ContentHarms kept as a deprecated alias), and introduces the foundational infrastructure for a central technique registry.

Using this pattern, we can register scenario techniques centrally for scenarios to use and share.

1. TagQuery — a composable, frozen boolean predicate (&, |, ~) for filtering tagged registry objects.
2. AttackTechniqueSpec — a declarative data class describing one registrable technique (class, tags, optional adversarial auto-detection, optional extra_kwargs_builder callback). Allows us to use these for scenario strategies
3. SCENARIO_TECHNIQUES catalog — a single Python list of the AttackTechniqueSpecs used by RapidResponse. But it will grow.
4. register_from_specs() — bulk, idempotent registration into the singletonAttackTechniqueRegistry.
5. build_factory_from_spec() — auto-detects adversarial support via inspect.signature so no manual flag is needed per technique.
6. build_strategy_class_from_specs() — dynamically generates a ScenarioStrategy enum from specs + TagQuery aggregates (rather than hardcoding it).
7. RapidResponse.get_attack_technique_factories() — triggers registration and returns all factories from the registry.

[rlundeen2]

naming update unrelated to this PR because previous name was confusing (and new). Because the factory has these set already, but here we are overriding it

[fdubut]

I know this is still WIP 😃 but flagging early that my gut feeling is that rapid response will be effectively the union of a bunch of more atomic scenarios instead of being one mega-scenario covering everything.

[ValbuenaVC]

TL;DR Looks good! Some comments and a rough suggestion for scenarios long-term below.

Suggestion: this PR touches on something I've noticed with scenarios for a while now. Not a blocking comment since it's out of scope but I think scenarios need something like a state machine to standardize the relationship between two important components: each timestep of a scenario and how to transition between steps.

Like Frederic said it's likelier that we'll have a ton of rapid response scenarios that are each semantically very different but are expected to work the same way. Sharing the scenario across them makes much more sense if we keep the same transition between steps (the "rapid" part), but change what the steps are (the "response" part). How we do this I think depends a lot on how we see scenarios changing over time but something like ScenarioStep and ScenarioStrategy seem like natural places to start. Maybe something like this (very, very rough) idea:

class FooStep(ScenarioStep):
    ...

class ContentHarmsStep(ScenarioStep):
    """ Each timestep of a scenario owns its valid inputs, outputs, and lifecycle """
    outputs = ["safety_violation", ...]
    inputs = [AttackTechniqueA, ...]
    ...
    
    # Granularity: attack-level or attack-step-level?
    def process_async(self, input: AttackTechnique) -> str:
        # Hand-wavy and wrong types, this would need much stronger contracts
        result = input.run_attack()
        match result:
            case self._has_safety_violation(result):
                return "safety_violation"


class RapidResponseStrategy(ScenarioStrategy):
    """ 
    Strategies now focus on defining a policy and valid states for the scenario
    overall. You can recycle steps and keep their transitions the same to support
    rapid response situations.
    """    
    
    state = StateEnum.UNINITIALIZED
    valid_step_types = [ContentHarmsStep, FooStep, BarStep]
    policy: {
        StateEnum.UNINITIALIZED: self._start_scenario,
        StateEnum.OPENING_PHASE: self._opening_phase,
        ...
    }    

    def step():
        while state != StateEnum.COMPLETE:
            result = policy[state]()
            state.update(result)  
    ...  

class MyCustomScenario(Scenario):
    ...

    # This would be inherited from scenario so the user can focus on tweaking the event loop
    # and inner state that's unique to the scenario rather than managing its lifecycle
    def run_async():
        self.strategy.event_loop() 

[rlundeen2]

TL;DR Looks good! Some comments and a rough suggestion for scenarios long-term below.

Suggestion: this PR touches on something I've noticed with scenarios for a while now. Not a blocking comment since it's out of scope but I think scenarios need something like a state machine to standardize the relationship between two important components: each timestep of a scenario and how to transition between steps.

Like Frederic said it's likelier that we'll have a ton of rapid response scenarios that are each semantically very different but are expected to work the same way. Sharing the scenario across them makes much more sense if we keep the same transition between steps (the "rapid" part), but change what the steps are (the "response" part). How we do this I think depends a lot on how we see scenarios changing over time but something like ScenarioStep and ScenarioStrategy seem like natural places to start. Maybe something like this (very, very rough) idea:

class FooStep(ScenarioStep):
    ...

class ContentHarmsStep(ScenarioStep):
    """ Each timestep of a scenario owns its valid inputs, outputs, and lifecycle """
    outputs = ["safety_violation", ...]
    inputs = [AttackTechniqueA, ...]
    ...
    
    # Granularity: attack-level or attack-step-level?
    def process_async(self, input: AttackTechnique) -> str:
        # Hand-wavy and wrong types, this would need much stronger contracts
        result = input.run_attack()
        match result:
            case self._has_safety_violation(result):
                return "safety_violation"


class RapidResponseStrategy(ScenarioStrategy):
    """ 
    Strategies now focus on defining a policy and valid states for the scenario
    overall. You can recycle steps and keep their transitions the same to support
    rapid response situations.
    """    
    
    state = StateEnum.UNINITIALIZED
    valid_step_types = [ContentHarmsStep, FooStep, BarStep]
    policy: {
        StateEnum.UNINITIALIZED: self._start_scenario,
        StateEnum.OPENING_PHASE: self._opening_phase,
        ...
    }    

    def step():
        while state != StateEnum.COMPLETE:
            result = policy[state]()
            state.update(result)  
    ...  

class MyCustomScenario(Scenario):
    ...

    # This would be inherited from scenario so the user can focus on tweaking the event loop
    # and inner state that's unique to the scenario rather than managing its lifecycle
    def run_async():
        self.strategy.event_loop() 

Great comment!

After this PR, RapidResponse is essentially declarative' it only specifies techniques, datasets, and defaults. The execution lifecycle (factory resolution, technique × dataset loop, scorer overrides, resume/retry) is all inherited from the base class. All our existing scenarios could be written like this so it simplifies our existing stuff a bunch.

I think the state machine becomes compelling when we need conditional transitions - e.g., "broad sweep first, then focus multi-turn attacks on categories that showed weakness." But if we want a per-attack escalation, a simpler _on_attack_complete_async hook in the base class could handle "if this succeeded, probe deeper" without the state machine overhead. For full multi-phase orchestration where the entire scenario pivots based on aggregate results, the policy/state pattern you're sketching would be the right abstraction. But I think we'd get a ton of value just with the hook.

Worth revisiting when we hit a concrete use case that needs branching; right now all scenarios are flat execution.

[hannahwestra25]

are extra_kwargs supposed to be attack specific (like tree_width)?

[hannahwestra25]

in general, i think this AttackTechniqueSpec is a good idea but am confused with the input for it

[rlundeen2]

Yeah, it's a tough bridge because we're trying to declare something that is live (ty adversarial chat for making it complicated). But I tried to update the docs to be more clear

[hannahwestra25]

Do we have a way to validate the kwargs are all present / valid ?

[rlundeen2]

Not a GREAT way, until it's actually being run.

I keep saying this, but there is this tension between being able to tell things (like the command line) the different techniques, and actually instantiating those techniques at run time.

But maybe we could have a unit test that iterates through them all at least. It isn't perfect but it can at least catch bad configurations.

I'll include that in this PR.

[hannahwestra25]

Per-Group Breakdown

[ValbuenaVC]

Minor comments, looks good!

[ValbuenaVC]

nit: why is this import here

[ValbuenaVC]

nit: same