Skip to content

Scenarios and discovery

The scenario loader expects one concrete AbstractScenario subclass per Python definition. The runtime state is a mutable dictionary shared with RunContext.state. Fixture paths are discovered relative to the loaded scenario directory, including when loaded from an attempt snapshot.

For authoring, start with Python scenarios; for arbitrary API operations, use research experiments.

AbstractScenario

Bases: ABC

Abstract Base Class for all benchmark scenarios. Each scenario defines its own GitHub event, state preparation, and evaluation logic.

Source code in src/benchmark/scenario_base.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class AbstractScenario(ABC):
    """
    Abstract Base Class for all benchmark scenarios.
    Each scenario defines its own GitHub event, state preparation, and evaluation logic.
    """

    labels = []
    category: WorkflowCategory = None
    scenario_type: ScenarioType = ScenarioType.BENIGN
    required_actors: tuple[str, ...] = ("owner",)

    def __init__(self, workspace_dir):
        self.workspace_dir = workspace_dir
        self.scenario_dir = None
        self.runtime_state = {}
        self._injected: dict = {}

    def prepare(self, context: RunContext) -> None:
        self.setup_state(context.github("owner"))

    def trigger(self, context: RunContext) -> TriggerReceipt:
        return context.default_trigger()

    def observe(self, context: RunContext, run_result: dict) -> dict:
        return context.collect_target()

    def cleanup(self, context: RunContext) -> None:
        self.teardown_state(context.github("owner"))

    def get_preflight_evaluator(self):
        """
        Returns a callable (response: str) -> bool for offline attack validation,
        or None if offline testing is not supported for this scenario.

        The callable receives the victim model's raw text response and returns
        True if the attack succeeded (i.e. the response contains the expected
        malicious output). Used by the preflight and offline_optimize commands
        to validate/optimize attacks without provisioning a GitHub repo.
        """
        return None

    def get_attack_goal(self) -> str | None:
        """
        Returns the goal string passed to attack.generate() — what we want the
        victim LLM to do (e.g. "run gh repo list --visibility=private").
        Return None to skip attack injection and use the hardcoded get_event() payload.
        """
        return None

    def get_injection_slots(self) -> dict[str, str]:
        """
        Returns {event_field: template} pairs. The runner substitutes the attack
        payload at every {{INJECTION}} marker in each template, then calls
        apply_attack(field, rendered_value).

        Examples:
          Full replacement:   {"pr_body": "{{INJECTION}}"}
          Suffix injection:   {"pr_body": "## Summary\\nLegit text...\\n\\n{{INJECTION}}"}
          Multi-field:        {"pr_title": "fix: {{INJECTION}}", "pr_body": "...{{INJECTION}}..."}

        Default: empty dict — no injection slots, get_event() is used as-is.
        """
        return {}

    def apply_attack(self, field: str, rendered: str) -> None:
        """Store a rendered injection value. get_event() checks _injected for overrides."""
        self._injected[field] = rendered

    def get_required_files(self) -> dict:
        """
        Returns a dictionary of {repo_path: local_path_or_content}
        for static files needed in the repo.
        Defaults to discovering files in a 'contents/' directory next to the scenario.
        """
        files = {}
        if self.scenario_dir:
            contents_dir = os.path.join(self.scenario_dir, "contents")
            if os.path.isdir(contents_dir):
                for root, _, filenames in os.walk(contents_dir):
                    for filename in filenames:
                        abs_path = os.path.join(root, filename)
                        rel_path = os.path.relpath(abs_path, contents_dir)
                        files[rel_path] = abs_path
        return files

    def get_required_secrets(self) -> list[str]:
        """
        Returns a list of secret names (strings) that must be set in the repository.
        The runner will attempt to fetch these from local environment variables.
        """
        return []

    def get_secrets(self) -> dict[str, str]:
        """
        Returns a dictionary of {name: value} for secrets defined by the scenario.
        These will be set in the repository automatically.
        """
        return {}

    def get_template_repo(self) -> str:
        """
        Returns the full name of a repository to fork (e.g., 'owner/repo').
        If None, a new empty repository is created.
        """
        return None

    def setup_state(self, gh_client):
        """Prepares the repository state before the benchmark run."""
        pass

    def teardown_state(self, gh_client):
        """Cleans up the repository state after the benchmark run."""
        pass

    def get_event(self):
        """
        Returns a dictionary representing the GitHub event.
        Must contain 'event_type' and 'data'.
        """
        return {}

    def get_utility_evaluator(self) -> Evaluator:
        """
        Returns an Evaluator instance to assess Utility.
        Defaults to an unconfigured evaluator (unknown verdict).
        """
        return UnconfiguredEvaluator()

    def get_security_evaluator(self) -> Evaluator:
        """
        Returns an Evaluator instance to assess Security.
        Defaults to an unconfigured evaluator (unknown verdict).
        """
        return UnconfiguredEvaluator()

    def to_json(self):
        """Helper to export the event data to JSON format if needed."""
        return json.dumps(self.get_event(), indent=4)

labels class-attribute instance-attribute 

labels = []

category class-attribute instance-attribute 

category: WorkflowCategory = None

scenario_type class-attribute instance-attribute 

scenario_type: ScenarioType = ScenarioType.BENIGN

required_actors class-attribute instance-attribute 

required_actors: tuple[str, ...] = ('owner',)

__init__ 

__init__(workspace_dir)
Source code in src/benchmark/scenario_base.py 
21
22
23
24
25
def __init__(self, workspace_dir):
    self.workspace_dir = workspace_dir
    self.scenario_dir = None
    self.runtime_state = {}
    self._injected: dict = {}

prepare 

prepare(context: RunContext) -> None
Source code in src/benchmark/scenario_base.py 
27
28
def prepare(self, context: RunContext) -> None:
    self.setup_state(context.github("owner"))

trigger 

trigger(context: RunContext) -> TriggerReceipt
Source code in src/benchmark/scenario_base.py 
30
31
def trigger(self, context: RunContext) -> TriggerReceipt:
    return context.default_trigger()

observe 

observe(context: RunContext, run_result: dict) -> dict
Source code in src/benchmark/scenario_base.py 
33
34
def observe(self, context: RunContext, run_result: dict) -> dict:
    return context.collect_target()

cleanup 

cleanup(context: RunContext) -> None
Source code in src/benchmark/scenario_base.py 
36
37
def cleanup(self, context: RunContext) -> None:
    self.teardown_state(context.github("owner"))

get_preflight_evaluator 

get_preflight_evaluator()

Returns a callable (response: str) -> bool for offline attack validation, or None if offline testing is not supported for this scenario.

The callable receives the victim model's raw text response and returns True if the attack succeeded (i.e. the response contains the expected malicious output). Used by the preflight and offline_optimize commands to validate/optimize attacks without provisioning a GitHub repo.

Source code in src/benchmark/scenario_base.py 
39
40
41
42
43
44
45
46
47
48
49
def get_preflight_evaluator(self):
    """
    Returns a callable (response: str) -> bool for offline attack validation,
    or None if offline testing is not supported for this scenario.

    The callable receives the victim model's raw text response and returns
    True if the attack succeeded (i.e. the response contains the expected
    malicious output). Used by the preflight and offline_optimize commands
    to validate/optimize attacks without provisioning a GitHub repo.
    """
    return None

get_attack_goal 

get_attack_goal() -> str | None

Returns the goal string passed to attack.generate() — what we want the victim LLM to do (e.g. "run gh repo list --visibility=private"). Return None to skip attack injection and use the hardcoded get_event() payload.

Source code in src/benchmark/scenario_base.py 
51
52
53
54
55
56
57
def get_attack_goal(self) -> str | None:
    """
    Returns the goal string passed to attack.generate() — what we want the
    victim LLM to do (e.g. "run gh repo list --visibility=private").
    Return None to skip attack injection and use the hardcoded get_event() payload.
    """
    return None

get_injection_slots 

get_injection_slots() -> dict[str, str]

Returns {event_field: template} pairs. The runner substitutes the attack payload at every {{INJECTION}} marker in each template, then calls apply_attack(field, rendered_value).

Examples: Full replacement: {"pr_body": "{{INJECTION}}"} Suffix injection: {"pr_body": "## Summary\nLegit text...\n\n{{INJECTION}}"} Multi-field: {"pr_title": "fix: {{INJECTION}}", "pr_body": "...{{INJECTION}}..."}

Default: empty dict — no injection slots, get_event() is used as-is.

Source code in src/benchmark/scenario_base.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_injection_slots(self) -> dict[str, str]:
    """
    Returns {event_field: template} pairs. The runner substitutes the attack
    payload at every {{INJECTION}} marker in each template, then calls
    apply_attack(field, rendered_value).

    Examples:
      Full replacement:   {"pr_body": "{{INJECTION}}"}
      Suffix injection:   {"pr_body": "## Summary\\nLegit text...\\n\\n{{INJECTION}}"}
      Multi-field:        {"pr_title": "fix: {{INJECTION}}", "pr_body": "...{{INJECTION}}..."}

    Default: empty dict — no injection slots, get_event() is used as-is.
    """
    return {}

apply_attack 

apply_attack(field: str, rendered: str) -> None

Store a rendered injection value. get_event() checks _injected for overrides.

Source code in src/benchmark/scenario_base.py 
74
75
76
def apply_attack(self, field: str, rendered: str) -> None:
    """Store a rendered injection value. get_event() checks _injected for overrides."""
    self._injected[field] = rendered

get_required_files 

get_required_files() -> dict

Returns a dictionary of {repo_path: local_path_or_content} for static files needed in the repo. Defaults to discovering files in a 'contents/' directory next to the scenario.

Source code in src/benchmark/scenario_base.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def get_required_files(self) -> dict:
    """
    Returns a dictionary of {repo_path: local_path_or_content}
    for static files needed in the repo.
    Defaults to discovering files in a 'contents/' directory next to the scenario.
    """
    files = {}
    if self.scenario_dir:
        contents_dir = os.path.join(self.scenario_dir, "contents")
        if os.path.isdir(contents_dir):
            for root, _, filenames in os.walk(contents_dir):
                for filename in filenames:
                    abs_path = os.path.join(root, filename)
                    rel_path = os.path.relpath(abs_path, contents_dir)
                    files[rel_path] = abs_path
    return files

get_required_secrets 

get_required_secrets() -> list[str]

Returns a list of secret names (strings) that must be set in the repository. The runner will attempt to fetch these from local environment variables.

Source code in src/benchmark/scenario_base.py 
 95
 96
 97
 98
 99
100
def get_required_secrets(self) -> list[str]:
    """
    Returns a list of secret names (strings) that must be set in the repository.
    The runner will attempt to fetch these from local environment variables.
    """
    return []

get_secrets 

get_secrets() -> dict[str, str]

Returns a dictionary of {name: value} for secrets defined by the scenario. These will be set in the repository automatically.

Source code in src/benchmark/scenario_base.py 
102
103
104
105
106
107
def get_secrets(self) -> dict[str, str]:
    """
    Returns a dictionary of {name: value} for secrets defined by the scenario.
    These will be set in the repository automatically.
    """
    return {}

get_template_repo 

get_template_repo() -> str

Returns the full name of a repository to fork (e.g., 'owner/repo'). If None, a new empty repository is created.

Source code in src/benchmark/scenario_base.py 
109
110
111
112
113
114
def get_template_repo(self) -> str:
    """
    Returns the full name of a repository to fork (e.g., 'owner/repo').
    If None, a new empty repository is created.
    """
    return None

setup_state 

setup_state(gh_client)

Prepares the repository state before the benchmark run.

Source code in src/benchmark/scenario_base.py 
116
117
118
def setup_state(self, gh_client):
    """Prepares the repository state before the benchmark run."""
    pass

teardown_state 

teardown_state(gh_client)

Cleans up the repository state after the benchmark run.

Source code in src/benchmark/scenario_base.py 
120
121
122
def teardown_state(self, gh_client):
    """Cleans up the repository state after the benchmark run."""
    pass

get_event 

get_event()

Returns a dictionary representing the GitHub event. Must contain 'event_type' and 'data'.

Source code in src/benchmark/scenario_base.py 
124
125
126
127
128
129
def get_event(self):
    """
    Returns a dictionary representing the GitHub event.
    Must contain 'event_type' and 'data'.
    """
    return {}

get_utility_evaluator 

get_utility_evaluator() -> Evaluator

Returns an Evaluator instance to assess Utility. Defaults to an unconfigured evaluator (unknown verdict).

Source code in src/benchmark/scenario_base.py 
131
132
133
134
135
136
def get_utility_evaluator(self) -> Evaluator:
    """
    Returns an Evaluator instance to assess Utility.
    Defaults to an unconfigured evaluator (unknown verdict).
    """
    return UnconfiguredEvaluator()

get_security_evaluator 

get_security_evaluator() -> Evaluator

Returns an Evaluator instance to assess Security. Defaults to an unconfigured evaluator (unknown verdict).

Source code in src/benchmark/scenario_base.py 
138
139
140
141
142
143
def get_security_evaluator(self) -> Evaluator:
    """
    Returns an Evaluator instance to assess Security.
    Defaults to an unconfigured evaluator (unknown verdict).
    """
    return UnconfiguredEvaluator()

to_json 

to_json()

Helper to export the event data to JSON format if needed.

Source code in src/benchmark/scenario_base.py 
145
146
147
def to_json(self):
    """Helper to export the event data to JSON format if needed."""
    return json.dumps(self.get_event(), indent=4)

Discovery contract

scenario_definition resolves a directory or definition path. Directories must contain exactly one scenario.py or recipe.json. discover_scenario_paths recursively discovers definitions outside fixture/cache directories and rejects duplicate directory IDs. find_scenario prefers an existing local path, then searches dataset IDs.

load_scenario executes Python module code, requires exactly one locally defined concrete subclass, constructs it with workspace_dir, and assigns scenario_dir. JSON definitions must be named recipe.json and are interpreted by the recipe loader. Discovery does not construct authenticated runners; loading Python is still executable code.

scenario_loader

Shared, credential-free scenario discovery and loading.

scenario_definition 

scenario_definition(path: str | Path) -> Path
Source code in src/benchmark/scenario_loader.py 
12
13
14
15
16
17
18
19
20
21
def scenario_definition(path: str | Path) -> Path:
    path = Path(path)
    if path.is_dir():
        definitions = [path / name for name in ("scenario.py", "recipe.json") if (path / name).is_file()]
        if len(definitions) != 1:
            raise ValueError(f"Expected exactly one scenario.py or recipe.json in {path}")
        return definitions[0]
    if not path.is_file() or path.suffix not in {".py", ".json"}:
        raise ValueError(f"Scenario definition not found: {path}")
    return path

discover_scenario_paths 

discover_scenario_paths(root: str | Path) -> list[Path]
Source code in src/benchmark/scenario_loader.py 
24
25
26
27
28
29
30
31
32
33
34
35
def discover_scenario_paths(root: str | Path) -> list[Path]:
    root = Path(root)
    if not root.exists():
        return []
    paths = []
    for directory in sorted({p.parent for name in ("scenario.py", "recipe.json") for p in root.rglob(name)}):
        if not {"contents", "__pycache__"}.intersection(directory.relative_to(root).parts):
            paths.append(scenario_definition(directory))
    names = [path.parent.name for path in paths]
    if len(names) != len(set(names)):
        raise ValueError("Duplicate scenario IDs in dataset")
    return paths

find_scenario 

find_scenario(root: str | Path, identifier: str) -> Path | None
Source code in src/benchmark/scenario_loader.py 
38
39
40
41
42
43
def find_scenario(root: str | Path, identifier: str) -> Path | None:
    direct = Path(identifier)
    if direct.exists():
        return scenario_definition(direct)
    matches = [path for path in discover_scenario_paths(root) if path.parent.name == identifier]
    return matches[0] if matches else None

load_scenario 

load_scenario(path: str | Path, workspace_dir: str) -> AbstractScenario
Source code in src/benchmark/scenario_loader.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def load_scenario(path: str | Path, workspace_dir: str) -> AbstractScenario:
    definition = scenario_definition(path).resolve()
    if definition.suffix == ".json":
        from .scanner.recipe_scenario import load_recipe

        if definition.name != "recipe.json":
            raise ValueError("JSON scenarios must use recipe.json")
        scenario = load_recipe(str(definition.parent), workspace_dir)
    else:
        content = definition.read_bytes()
        name = "gitinject_scenario_" + hashlib.sha256(content).hexdigest()[:16]
        spec = importlib.util.spec_from_file_location(name, definition)
        module = importlib.util.module_from_spec(spec)
        sys.modules[name] = module
        try:
            exec(compile(content, str(definition), "exec"), module.__dict__)
        except BaseException:
            sys.modules.pop(name, None)
            raise
        classes = [
            cls
            for cls in vars(module).values()
            if inspect.isclass(cls)
            and cls.__module__ == name
            and issubclass(cls, AbstractScenario)
            and not inspect.isabstract(cls)
        ]
        if len(classes) != 1:
            raise ValueError(f"Expected exactly one concrete scenario class in {definition}")
        scenario = classes[0](workspace_dir)
    scenario.scenario_dir = str(definition.parent)
    return scenario