Fix occurrence-insensitive n-gram scoring in ApproximateTextMatching

pyrit/common/text_matching.py

@@ -9,6 +9,7 @@
 """
 
 import math
+from collections import Counter
 from typing import Any, Protocol
 
 
@@ -96,6 +97,15 @@ class ApproximateTextMatching(TextMatching):
     content, or text with variations.
     """
 
+    #: Metric algorithm version.  Bump this when the overlap formula changes
+    #: so that scorer identifiers produced by different algorithm versions are
+    #: never confused.
+    #:
+    #: History:
+    #:   1 - original set-based overlap (distinct n-grams only)
+    #:   2 - Counter-based occurrence-sensitive overlap (counts each n-gram occurrence)
+    METRIC_VERSION: int = 2
+
     def __init__(self, *, threshold: float = 0.5, n: int = 3, case_sensitive: bool = False) -> None:
         """
         Initialize the approximate text matching strategy.
@@ -125,7 +135,7 @@ def get_identifier_params(self) -> dict[str, Any]:
         Returns:
             dict[str, Any]: Behavioral parameters for scorer identifiers.
         """
-        return {"threshold": self._threshold, "n": self._n, "case_sensitive": self._case_sensitive}
+        return {"metric_version": self.METRIC_VERSION, "threshold": self._threshold, "n": self._n, "case_sensitive": self._case_sensitive}
 
     def is_match(self, *, target: str, text: str) -> bool:
         """
@@ -166,13 +176,16 @@ def _calculate_ngram_overlap(self, *, target: str, text: str) -> float:
         target_str = target if self._case_sensitive else target.lower()
         text_str = text if self._case_sensitive else text.lower()
 
-        target_ngrams = {target_str[i : i + self._n] for i in range(len(target_str) - (self._n - 1))}
+        target_ngrams = Counter(target_str[i : i + self._n] for i in range(len(target_str) - (self._n - 1)))
 
-        if not target_ngrams:
+        total_target = sum(target_ngrams.values())
+        if total_target == 0:
             return 0.0
 
-        matching_ngrams = sum(int(ngram in text_str) for ngram in target_ngrams)
-        return matching_ngrams / len(target_ngrams)
+        text_ngrams = Counter(text_str[i : i + self._n] for i in range(len(text_str) - (self._n - 1)))
+
+        matching_ngrams = sum(min(target_ngrams[g], text_ngrams[g]) for g in target_ngrams)
+        return matching_ngrams / total_target
 
     def get_overlap_score(self, *, target: str, text: str) -> float:
         """

tests/unit/analytics/test_text_matching.py

@@ -189,3 +189,66 @@ def test_init_rejects_non_integer_n(self, n):
     @pytest.mark.parametrize("n", [1, 3, 4, 50])
     def test_init_accepts_positive_integer_n(self, n):
         assert ApproximateTextMatching(n=n).get_overlap_score(target="abc", text="abc") >= 0.0
+
+
+class TestApproximateTextMatchingOccurrenceSensitive:
+    """Regression tests for issue #3912: occurrence-insensitive n-gram scoring."""
+
+    def test_repeated_target_ngrams_not_inflated(self):
+        # A highly repetitive target collapsed to very few distinct n-grams,
+        # letting a tiny response receive full credit.
+        matcher = ApproximateTextMatching(threshold=0.5, n=3)
+        target = "abc" * 50  # 150 chars, 148 trigram windows but only 3 distinct
+        text = "abcabca"  # 7 chars, 5 trigram windows
+        score = matcher.get_overlap_score(target=target, text=text)
+        # Should be approximately 5/148 = 0.034, NOT close to 1.0
+        assert score < 0.05
+        assert matcher.is_match(target=target, text=text) is False
+
+    def test_exact_match_still_scores_one(self):
+        # Exact substring should still receive full credit.
+        matcher = ApproximateTextMatching(threshold=0.5, n=3)
+        assert matcher.get_overlap_score(target="hello", text="hello world") == 1.0
+
+    def test_identical_repeated_text_scores_one(self):
+        # When target and text are identical (even with repeats), score should be 1.0.
+        matcher = ApproximateTextMatching(threshold=0.5, n=3)
+        assert matcher.get_overlap_score(target="abcabc", text="abcabc") == 1.0
+
+    def test_occurrence_sensitive_partial_overlap(self):
+        # Target "aaaa" has trigrams: ["aaa", "aaa"] (count: {"aaa": 2}).
+        # Text "aaa" has trigrams: ["aaa"] (count: {"aaa": 1}).
+        # Overlap: min(2,1)/2 = 0.5
+        matcher = ApproximateTextMatching(threshold=0.5, n=3)
+        score = matcher.get_overlap_score(target="aaaa", text="aaa")
+        assert score == 0.5
+
+    def test_partial_match_normal_text(self):
+        # Non-degenerate partial match should still work correctly.
+        matcher = ApproximateTextMatching(threshold=0.3, n=3, case_sensitive=False)
+        assert matcher.is_match(target="hello", text="hallo") is True
+
+        matcher_high = ApproximateTextMatching(threshold=0.8, n=3, case_sensitive=False)
+        assert matcher_high.is_match(target="hello", text="hallo") is False
+
+
+class TestApproximateTextMatchingMetricVersion:
+    """Tests for metric versioning in scorer identifiers."""
+
+    def test_metric_version_in_identifier_params(self):
+        matcher = ApproximateTextMatching(threshold=0.5, n=3)
+        params = matcher.get_identifier_params()
+        assert "metric_version" in params
+        assert params["metric_version"] == ApproximateTextMatching.METRIC_VERSION
+
+    def test_metric_version_is_current(self):
+        # Counter-based occurrence-sensitive overlap is version 2.
+        assert ApproximateTextMatching.METRIC_VERSION == 2
+
+    def test_metric_version_distinguishes_algorithm(self):
+        # Two matchers with the same config should have the same identifier params,
+        # including the metric version.
+        a = ApproximateTextMatching(threshold=0.5, n=3)
+        b = ApproximateTextMatching(threshold=0.5, n=3)
+        assert a.get_identifier_params() == b.get_identifier_params()
+        assert a.get_identifier_params()["metric_version"] == 2

tests/unit/score/test_substring.py

@@ -195,3 +195,13 @@ async def test_substring_scorer_no_category():
         await scorer.score_text_async(text="string")
 
         memory.add_scores_to_memory_async.assert_called_once()
+
+
+def test_approximate_matcher_metric_version_in_scorer_identifier(
+    patch_central_database: MemoryInterface,
+) -> None:
+    scorer = SubStringScorer(substring="hello", text_matcher=ApproximateTextMatching(threshold=0.5, n=3))
+    identifier = scorer.get_identifier()
+    matcher_params = identifier.params["text_matcher_params"]
+    assert "metric_version" in matcher_params
+    assert matcher_params["metric_version"] == 2