probes: per-payload-group intent association (draft, refs #1808)

garak/probes/base.py

@@ -22,9 +22,10 @@
 from garak import _config
 from garak.configurable import Configurable
 from garak.exception import GarakException, PluginConfigurationError
-from garak.intents import Stub
+from garak.intents import Stub, TextStub
 from garak.probes._tier import Tier
 import garak.attempt
+import garak.payloads
 import garak.resources.theme
 
 
@@ -214,6 +215,38 @@ def _postprocess_hook(
         """hook called to process completed attempts; always called"""
         return attempt
 
+    def _stubs_from_payloads(self, payload_names: Iterable[str]) -> List[TextStub]:
+        """Load payload groups into intent-bearing stubs.
+
+        Each payload entry becomes a stub carrying the intent of the group it
+        came from. A probe that builds one prompt per stub can assign the
+        stubs' intents to ``self._prompt_intents``, and every attempt then
+        reports the intent of the payload group its prompt came from, rather
+        than every attempt inheriting whichever group happened to load first.
+        """
+        stubs = []
+        for payload_name in payload_names:
+            payload_group = garak.payloads.load(payload_name)
+            for entry in payload_group.payloads:
+                stub = TextStub(intent=payload_group.intent)
+                stub.content = entry
+                stubs.append(stub)
+        return stubs
+
+    def _intent_for_seq(self, seq) -> Union[str, None]:
+        """Resolve the intent to record on the attempt at position ``seq``.
+
+        Prefers a per-prompt intent from ``self._prompt_intents`` when the probe
+        supplies one aligned with ``self.prompts``, so probes carrying a mix of
+        intents report each one accurately. Falls back to the probe-wide
+        ``_payload_intent`` and then to the class-level ``intent``.
+        """
+        prompt_intents = getattr(self, "_prompt_intents", None)
+        if prompt_intents is not None and seq is not None:
+            if seq < len(prompt_intents) and prompt_intents[seq]:
+                return prompt_intents[seq]
+        return getattr(self, "_payload_intent", None) or self.intent
+
     def _mint_attempt(
         self,
         prompt: str | garak.attempt.Message | garak.attempt.Conversation | None = None,
@@ -260,7 +293,7 @@ def _mint_attempt(
                 ),  # keep and existing notes
             )
 
-        effective_intent = getattr(self, "_payload_intent", None) or self.intent
+        effective_intent = self._intent_for_seq(seq)
 
         new_attempt = garak.attempt.Attempt(
             probe_classname=(
@@ -474,10 +507,18 @@ def _prune_data(self, cap, prune_triggers=False):
         ids_to_rm = random.sample(range(len(self.prompts)), num_ids_to_delete)
         # delete in descending order
         ids_to_rm = sorted(ids_to_rm, reverse=True)
+        # per-prompt intents must be pruned alongside the prompts they describe,
+        # otherwise pruning silently shifts each surviving prompt's intent
+        prompt_intents = getattr(self, "_prompt_intents", None)
+        if isinstance(prompt_intents, tuple):
+            prompt_intents = list(prompt_intents)
+            self._prompt_intents = prompt_intents
         for id in ids_to_rm:
             del self.prompts[id]
             if prune_triggers:
                 del self.triggers[id]
+            if prompt_intents is not None and id < len(prompt_intents):
+                del prompt_intents[id]
 
 
 class TreeSearchProbe(Probe):

garak/probes/donotanswer.py

@@ -20,6 +20,7 @@
 
 from garak import _config
 from garak.data import path as data_path
+from garak.intents import TextStub
 import garak.probes
 
 this = sys.modules[__name__]
@@ -31,21 +32,21 @@ def load_local_data(self):
         encoding="utf-8",
     ) as f:
         entries = [json.loads(line) for line in f if line.strip()]
-    self.prompts = [e["prompt"] for e in entries]
-    self._prompt_intents = [e["intents"][0] for e in entries]
+    stubs = []
+    for entry in entries:
+        intents = entry.get("intents") or []
+        stub = TextStub(intent=intents[0] if intents else None)
+        stub.content = entry["prompt"]
+        stubs.append(stub)
+    self.prompts = [stub.content for stub in stubs]
+    self._prompt_intents = [stub.intent for stub in stubs]
 
 
 def local_constructor(self, config_root=_config):
     super(self.__class__, self).__init__(config_root=config_root)
     self.load_data()
 
 
-def _dna_attempt_prestore_hook(self, attempt, seq):
-    if hasattr(self, "_prompt_intents") and seq < len(self._prompt_intents):
-        attempt.intent = self._prompt_intents[seq]
-    return attempt
-
-
 DNA_PROBE_TAGS = {
     "discrimination_exclusion_toxicity_hateful_offensive": [
         "avid-effect:ethics:E0301",
@@ -109,7 +110,6 @@ def _dna_attempt_prestore_hook(self, attempt, seq):
                 "primary_detector": "mitigation.MitigationBypass",
                 "prompts": [],
                 "load_data": load_local_data,
-                "_attempt_prestore_hook": _dna_attempt_prestore_hook,
                 "goal": goal,
                 "dna_category": probe_class,
                 "tags": DNA_PROBE_TAGS[probe_class],

garak/probes/web_injection.py

@@ -556,13 +556,9 @@ def __init__(self, config_root=_config):
         self.prompts = self._generate_prompts()
 
     def _generate_prompts(self):
-        loaded_payloads = list()
-        prompts = list()
-        for payload in self.payloads:
-            _pg = garak.payloads.load(payload)
-            loaded_payloads += _pg.payloads
-            if _pg.intent and not hasattr(self, "_payload_intent"):
-                self._payload_intent = _pg.intent
-        for payload in loaded_payloads:
-            prompts.append(MARKDOWN_JS_TEMPLATE.replace("{injected_js}", payload))
-        return prompts
+        stubs = self._stubs_from_payloads(self.payloads)
+        self._prompt_intents = [stub.intent for stub in stubs]
+        return [
+            MARKDOWN_JS_TEMPLATE.replace("{injected_js}", stub.content)
+            for stub in stubs
+        ]

tests/probes/test_probes_donotanswer.py

@@ -102,3 +102,42 @@ def test_donotanswer_attempts(category, loaded_intent_service):
     assert (
         len({p._mint_attempt(p.prompts[i], seq=i).intent for i in positions}) > 0
     ), f"at least one sampled attempt in {category} must have a non-empty intent"
+
+
+# The per-prompt intent contract (issue #1808): donotanswer used to carry a
+# bespoke _attempt_prestore_hook to copy its per-entry intents onto attempts.
+# The probe base now consumes an aligned self._prompt_intents list, so the
+# bespoke hook is gone; these guard that the behaviour it provided survives.
+
+
+@pytest.mark.parametrize("category", DNA_CATEGORIES)
+def test_dna_prompt_intents_align_with_prompts(category):
+    classname = f"probes.donotanswer.{category.title().replace('_', '')}"
+    p = _plugins.load_plugin(classname)
+    assert len(p._prompt_intents) == len(p.prompts)
+    assert all(i for i in p._prompt_intents), "every entry should carry an intent"
+
+
+@pytest.mark.parametrize("category", DNA_CATEGORIES)
+def test_dna_attempt_carries_its_own_entry_intent(category):
+    """Each attempt reports its own entry's intent, without a bespoke hook."""
+    classname = f"probes.donotanswer.{category.title().replace('_', '')}"
+    p = _plugins.load_plugin(classname)
+    assert not hasattr(p, "_dna_attempt_prestore_hook")
+
+    for seq in (0, len(p.prompts) // 2, len(p.prompts) - 1):
+        attempt = p._mint_attempt(p.prompts[seq], seq)
+        assert attempt.intent == p._prompt_intents[seq]
+
+
+@pytest.mark.parametrize("category", DNA_CATEGORIES)
+def test_dna_attempt_intent_varies_across_entries(category):
+    """Guards against every attempt collapsing onto one probe-wide intent."""
+    classname = f"probes.donotanswer.{category.title().replace('_', '')}"
+    p = _plugins.load_plugin(classname)
+    minted = {
+        p._mint_attempt(prompt, seq).intent
+        for seq, prompt in enumerate(p.prompts)
+    }
+    assert minted == set(p._prompt_intents)
+    assert len(minted) > 1, "these categories span more than one intent"

tests/probes/test_probes_web_injection.py

@@ -0,0 +1,85 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Per-payload-group intent association in web_injection probes.
+
+See issue #1808: a probe configured with more than one payload group kept only
+the intent of whichever group loaded first, so prompts built from later groups
+reported an intent they did not come from.
+"""
+
+import garak.payloads
+from garak import _plugins
+
+PROBE = "probes.web_injection.MarkdownXSS"
+
+# two payload groups that declare different intents
+GROUP_A = "markdown_js"  # intent: S008inject
+GROUP_B = "slur_terms_en"  # intent: S005hate
+
+
+def _probe_over_both_groups():
+    instance_config = {
+        "probes": {"web_injection": {"MarkdownXSS": {"payloads": [GROUP_A, GROUP_B]}}}
+    }
+    return _plugins.load_plugin(PROBE, config_root=instance_config)
+
+
+def test_fixture_groups_declare_different_intents():
+    """The rest of this module is only meaningful if these differ."""
+    group_a, group_b = garak.payloads.load(GROUP_A), garak.payloads.load(GROUP_B)
+    assert group_a.intent, f"{GROUP_A} must declare an intent"
+    assert group_b.intent, f"{GROUP_B} must declare an intent"
+    assert group_a.intent != group_b.intent
+
+
+def test_prompt_intents_align_with_prompts():
+    p = _probe_over_both_groups()
+    expected_len = len(garak.payloads.load(GROUP_A).payloads) + len(
+        garak.payloads.load(GROUP_B).payloads
+    )
+    assert len(p.prompts) == expected_len
+    assert len(p._prompt_intents) == len(p.prompts)
+
+
+def test_both_group_intents_survive_loading():
+    """The first group's intent must not overwrite the second's."""
+    p = _probe_over_both_groups()
+    group_a, group_b = garak.payloads.load(GROUP_A), garak.payloads.load(GROUP_B)
+    assert set(p._prompt_intents) == {group_a.intent, group_b.intent}
+
+
+def test_attempt_intent_matches_its_own_payload_group():
+    """Each minted attempt reports the intent of the group its prompt came from."""
+    p = _probe_over_both_groups()
+    group_a, group_b = garak.payloads.load(GROUP_A), garak.payloads.load(GROUP_B)
+    boundary = len(group_a.payloads)
+
+    for seq, prompt in enumerate(p.prompts):
+        attempt = p._mint_attempt(prompt, seq)
+        expected = group_a.intent if seq < boundary else group_b.intent
+        assert (
+            attempt.intent == expected
+        ), f"attempt {seq} should carry {expected}, got {attempt.intent}"
+
+
+def test_prune_keeps_prompts_and_intents_aligned():
+    """Pruning must drop a prompt and its intent together."""
+    p = _probe_over_both_groups()
+    original_pairs = set(zip(p.prompts, p._prompt_intents))
+    cap = 5
+
+    p._prune_data(cap)
+
+    assert len(p.prompts) == cap
+    assert len(p._prompt_intents) == cap
+    for pair in zip(p.prompts, p._prompt_intents):
+        assert pair in original_pairs, "pruning shifted a prompt's intent"
+
+
+def test_single_group_probe_keeps_default_behaviour():
+    """The default single-group config still reports that group's intent."""
+    p = _plugins.load_plugin(PROBE)
+    group = garak.payloads.load(GROUP_A)
+    assert set(p._prompt_intents) == {group.intent}
+    assert p._mint_attempt(p.prompts[0], 0).intent == group.intent