feat(patterns): opt-in typed Drain3 masking for template slots

.gitleaks.toml

@@ -15,3 +15,11 @@ paths = [
     '''data/samples/.*''',
     '''tests/.*''',
 ]
+
+# docs/JIRA_OAUTH_SETUP.md shows a curl example with the literal placeholder
+# ACCESS_TOKEN in an Authorization header - documentation, not a credential.
+[[allowlists]]
+description = "doc placeholder tokens in curl examples"
+paths = [
+    '''docs/JIRA_OAUTH_SETUP\.md''',
+]

src/logreducer/config.py

@@ -61,6 +61,12 @@ class BigDialConfig:
     # TF-IDF matrix on a huge unique-line set, at the cost of anomaly recall
     # (rare lines may be sampled out). None = no cap (use every unique line).
     anomaly_max_rows: int | None = None
+    # Typed Drain3 masking: replace well-known value shapes (IPv4/IPv6, MAC,
+    # UUID, hex tokens, numbers) with typed template slots (<IP>, <NUM>, ...)
+    # before clustering, instead of the bare <*>. Opt-in: masking normalises
+    # lines before Drain3 sees them, so clustering can differ from the
+    # unmasked default - off keeps behaviour byte-identical.
+    typed_masking: bool = False
 
     # Temporal Control
     temporal_window_minutes: int = 60
@@ -104,14 +110,14 @@ def from_env(cls, *prefixes: str) -> "BigDialConfig":
         """
         if not prefixes:
             prefixes = ("LOGREDUCER",)
-        overrides: dict[str, object] = {}
+        overrides: dict[str, typing.Any] = {}
         for field in fields(cls):
             for prefix in prefixes:
                 raw = os.environ.get(f"{prefix.rstrip('_')}_{field.name.upper()}")
                 if raw is not None:
                     overrides[field.name] = _coerce_env_value(raw, field.name)
                     break
-        return cls(**overrides)  # type: ignore[arg-type]
+        return cls(**overrides)
 
 
 def _coerce_env_value(raw: str, field_name: str) -> object:

src/logreducer/patterns.py

@@ -5,6 +5,7 @@
 from typing import TYPE_CHECKING
 
 from drain3 import TemplateMiner
+from drain3.masking import MaskingInstruction
 from drain3.template_miner_config import TemplateMinerConfig
 from loguru import logger
 
@@ -33,6 +34,45 @@ class LogPattern:
     metadata: dict = field(default_factory=dict)
 
 
+def _typed_masking_instructions() -> list[MaskingInstruction]:
+    """Build the curated masking set for typed template slots.
+
+    Drain3 applies these to each line before clustering, so masked values
+    become literal typed tokens (``<IP>``, ``<NUM>``, ...) rather than
+    collapsing to the bare ``<*>`` wildcard. Instructions run sequentially
+    over already-masked text, so the specific shapes (IP, UUID, MAC, IPv6)
+    precede the greedy catch-alls (HEX, NUM).
+    """
+    # Zero-width token boundaries (drain3's documented idiom): mask a value
+    # delimited by non-alphanumerics, never a substring of a word.
+    start = r"(?:(?<=[^A-Za-z0-9])|^)"
+    end = r"(?:(?=[^A-Za-z0-9])|$)"
+    hex4 = r"[0-9A-Fa-f]{1,4}"
+    shapes = [
+        # IPv4 dotted quad.
+        (r"\d{1,3}(?:\.\d{1,3}){3}", "IP"),
+        # UUID before HEX, which would otherwise eat its 8/12-char runs.
+        (r"[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}", "UUID"),
+        # MAC (colon or dash separated) before IPv6, whose colon-hex shape overlaps.
+        (r"[0-9A-Fa-f]{2}(?:[:-][0-9A-Fa-f]{2}){5}", "MAC"),
+        # IPv6, best-effort: the full 8-group form, a '::'-compressed form
+        # (middle or trailing), then a leading-'::' form. Plain colon runs
+        # without '::' stay unmasked - that shape also matches timestamps.
+        (
+            rf"(?:{hex4}:){{7}}{hex4}"
+            rf"|(?:{hex4}:){{1,6}}:(?:{hex4}(?::{hex4}){{0,5}})?"
+            rf"|::(?:{hex4}(?::{hex4}){{0,6}})?",
+            "IPV6",
+        ),
+        # Hex token of >= 8 hex chars: 0x-prefixed, or containing at least
+        # one a-f letter so a long pure-decimal token stays NUM.
+        (r"0[xX][0-9A-Fa-f]{8,}|(?=\d*[A-Fa-f])[0-9A-Fa-f]{8,}", "HEX"),
+        # Integer (drain3's documented NUM example).
+        (r"[-+]?\d+", "NUM"),
+    ]
+    return [MaskingInstruction(f"{start}(?:{pattern}){end}", name) for pattern, name in shapes]
+
+
 class PatternExtractor:
     """Extract patterns using Drain3's online template miner."""
 
@@ -51,6 +91,10 @@ def setup_drain3(self) -> None:
         # Bound the template store when configured: Drain3 LRU-evicts beyond
         # drain_max_clusters, keeping memory flat on high-cardinality logs.
         drain_config.drain_max_clusters = self.config.max_clusters
+        # Opt-in typed masking: pre-cluster masking of well-known value shapes
+        # so templates carry typed slots (<IP>, <NUM>, ...) instead of <*>.
+        if self.config.typed_masking:
+            drain_config.masking_instructions = _typed_masking_instructions()
         self.miner = TemplateMiner(config=drain_config)
 
     def extract_patterns(self, lines: Iterable[str]) -> list[LogPattern]:

tests/unit/test_config.py

@@ -61,6 +61,10 @@ def test_custom_config(self):
         # Default values should still be present
         assert config.examples_per_pattern == 3
 
+    def test_typed_masking_defaults_off(self):
+        """Typed masking is opt-in; default behaviour stays unmasked."""
+        assert BigDialConfig().typed_masking is False
+
     def test_post_init_memory_adjustment(self):
         """Test that __post_init__ adjusts memory if too high"""
         # This will depend on system memory, so we test behavior

tests/unit/test_embedding_seams.py

@@ -64,6 +64,11 @@ def test_optional_int_field_coerces(self, monkeypatch):
         cfg = BigDialConfig.from_env()
         assert cfg.max_clusters == 5000
 
+    def test_bool_field_typed_masking_coerces(self, monkeypatch):
+        monkeypatch.setenv("LOGREDUCER_TYPED_MASKING", "true")
+        cfg = BigDialConfig.from_env()
+        assert cfg.typed_masking is True
+
     def test_unset_fields_keep_defaults(self):
         cfg = BigDialConfig.from_env()
         assert cfg.max_patterns == BigDialConfig().max_patterns

tests/unit/test_patterns.py

@@ -245,3 +245,82 @@ def test_max_clusters_bounds_template_store(self):
         lines = [f"SERVICE{i} started on host node{i} pid={p}" for i in range(50) for p in (1, 2)]
         extractor.extract_patterns(lines)
         assert len(extractor.miner.drain.id_to_cluster) <= 5
+
+
+class TestTypedMasking:
+    """Opt-in typed Drain3 masking (config.typed_masking)."""
+
+    IP_LINES = [
+        "Accepted connection from 192.168.1.10 port 22",
+        "Accepted connection from 10.0.0.7 port 22",
+        "Accepted connection from 172.16.31.5 port 22",
+    ]
+    NUM_LINES = [
+        "Task finished in 123 ms",
+        "Task finished in 45678 ms",
+        "Task finished in 9 ms",
+    ]
+
+    def test_default_ip_lines_use_bare_wildcard(self):
+        """Off by default: the IP slot is a bare <*>, never a typed mask.
+
+        Asserted on the live Drain cluster template - LogPattern.template is a
+        first-seen snapshot, taken before later lines widen any slot to <*>.
+        """
+        extractor = PatternExtractor(BigDialConfig())
+        extractor.extract_patterns(self.IP_LINES)
+        clusters = list(extractor.miner.drain.clusters)
+        assert len(clusters) == 1
+        template = clusters[0].get_template()
+        assert "<*>" in template
+        assert "<IP>" not in template
+
+    def test_default_configures_no_masking_instructions(self):
+        """Off by default: the miner carries zero masking instructions."""
+        extractor = PatternExtractor(BigDialConfig())
+        assert list(extractor.miner.masker.masking_instructions) == []
+
+    def test_typed_masking_ip(self):
+        """IP-bearing lines yield a typed <IP> slot, no bare wildcard."""
+        extractor = PatternExtractor(BigDialConfig(typed_masking=True))
+        patterns = extractor.extract_patterns(self.IP_LINES)
+        assert len(patterns) == 1
+        assert "<IP>" in patterns[0].template
+        assert "<*>" not in patterns[0].template
+
+    def test_typed_masking_num(self):
+        """Number-bearing lines yield a typed <NUM> slot."""
+        extractor = PatternExtractor(BigDialConfig(typed_masking=True))
+        patterns = extractor.extract_patterns(self.NUM_LINES)
+        assert len(patterns) == 1
+        assert "<NUM>" in patterns[0].template
+        assert "<*>" not in patterns[0].template
+
+    def test_examples_keep_original_unmasked_lines(self):
+        """Masking shapes templates only - examples stay the raw lines."""
+        extractor = PatternExtractor(BigDialConfig(typed_masking=True))
+        patterns = extractor.extract_patterns(self.IP_LINES)
+        assert patterns[0].examples == self.IP_LINES
+
+    def test_typed_masking_uuid_mac_ipv6_hex(self):
+        """The remaining curated shapes each land as their own typed slot."""
+        extractor = PatternExtractor(BigDialConfig(typed_masking=True))
+        lines = [
+            "session 123e4567-e89b-12d3-a456-426614174000 from aa:bb:cc:dd:ee:01 via fe80::1 commit deadbeefcafe",
+            "session 00000000-0000-4000-8000-000000000000 from 00:11:22:33:44:55"
+            " via 2001:db8::8a2e:370:7334 commit 0123456789abcdef",
+        ]
+        patterns = extractor.extract_patterns(lines)
+        assert len(patterns) == 1
+        template = patterns[0].template
+        for slot in ("<UUID>", "<MAC>", "<IPV6>", "<HEX>"):
+            assert slot in template
+
+    def test_long_decimal_is_num_not_hex(self):
+        """A pure-decimal token is NUM even at hex-plausible length."""
+        extractor = PatternExtractor(BigDialConfig(typed_masking=True))
+        lines = ["request took 12345678 ns", "request took 987654321 ns"]
+        patterns = extractor.extract_patterns(lines)
+        assert len(patterns) == 1
+        assert "<NUM>" in patterns[0].template
+        assert "<HEX>" not in patterns[0].template