From f574b5535b0cf9f95cea3a08fe85ab6d7cf242b0 Mon Sep 17 00:00:00 2001
From: viktorbeck98 <viktor.beck98@gmail.com>
Date: Tue, 31 Mar 2026 15:38:11 +0200
Subject: [PATCH] feat(config): support named wildcards and named event IDs in
 template matcher
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Users can now write <label> instead of <*> in template files to give
wildcards meaningful names, and use an EventId column in CSV template
files to assign stable string identifiers to events. Both formats are
compiled to the existing positional representation at load time, so all
runtime code (ParserSchema, get_configured_variables) is unchanged.

- _compile_templates(): converts <label> → <*> and records label order
  and per-template event ID labels (from CSV EventId column)
- TemplatesManager / TemplateMatcher: accept metadata; expose
  compile_detector_config() to resolve string pos labels and string
  event keys to positional ints before pipeline startup
- Variable.pos: str | int (string = named label, compiled to int)
- Variable.name: optional (defaults to empty; label serves as name)
- load_templates(): returns (templates, event_id_labels); CSV files
  support an optional EventId column for named event IDs (.txt files
  remain positional-only)
- Tests and test fixtures for all new paths; roundtrip tests confirm
  string pos and string event keys survive YAML → Pydantic → YAML

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../common/_config/_formats.py                |  11 +-
 src/detectmatelibrary/common/detector.py      |   2 +-
 .../parsers/template_matcher/_matcher_op.py   |  78 +++++++++-
 .../parsers/template_matcher/_parser.py       |  91 +++++++++--
 tests/test_common/test_config_roundtrip.py    |  87 +++++++++++
 tests/test_folder/test_config.yaml            |  32 ++++
 tests/test_folder/test_named_templates.csv    |   2 +
 tests/test_folder/test_named_templates.txt    |   1 +
 tests/test_parsers/test_template_matcher.py   | 146 ++++++++++++++++++
 9 files changed, 430 insertions(+), 20 deletions(-)
 create mode 100644 tests/test_folder/test_named_templates.csv
 create mode 100644 tests/test_folder/test_named_templates.txt

diff --git a/src/detectmatelibrary/common/_config/_formats.py b/src/detectmatelibrary/common/_config/_formats.py
index 564d5e5..dcee7e2 100644
--- a/src/detectmatelibrary/common/_config/_formats.py
+++ b/src/detectmatelibrary/common/_config/_formats.py
@@ -6,16 +6,17 @@
 
 # Sub-formats ********************************************************+
 class Variable(BaseModel):
-    pos: int
-    name: str
+    pos: str | int
+    name: str = ""
     params: Dict[str, Any] = {}
 
     def to_dict(self) -> Dict[str, Any]:
         """Convert Variable to YAML-compatible dictionary."""
         result: Dict[str, Any] = {
             "pos": self.pos,
-            "name": self.name,
         }
+        if self.name:
+            result["name"] = self.name
         if self.params:
             result["params"] = self.params
         return result
@@ -38,7 +39,7 @@ def to_dict(self) -> Dict[str, Any]:
 class _EventInstance(BaseModel):
     """Configuration for a specific instance within an event."""
     params: Dict[str, Any] = {}
-    variables: Dict[int, Variable] = {}
+    variables: Dict[str | int, Variable] = {}
     header_variables: Dict[str, Header] = {}
 
     @classmethod
@@ -79,7 +80,7 @@ def _init(cls, instances_dict: Dict[str, Dict[str, Any]]) -> "_EventConfig":
         return cls(instances=instances)
 
     @property
-    def variables(self) -> Dict[int, Variable]:
+    def variables(self) -> Dict[str | int, Variable]:
         """Pass-through to first instance for compatibility."""
         if self.instances:
             return next(iter(self.instances.values())).variables
diff --git a/src/detectmatelibrary/common/detector.py b/src/detectmatelibrary/common/detector.py
index e224f68..e3fad55 100644
--- a/src/detectmatelibrary/common/detector.py
+++ b/src/detectmatelibrary/common/detector.py
@@ -56,7 +56,7 @@ def get_configured_variables(
     # Extract template variables by position
     if hasattr(event_config, "variables"):
         for pos, var in event_config.variables.items():
-            if pos < len(input_["variables"]):
+            if isinstance(pos, int) and pos < len(input_["variables"]):
                 result[var.name] = input_["variables"][pos]
 
     # Extract header/log format variables by name
diff --git a/src/detectmatelibrary/parsers/template_matcher/_matcher_op.py b/src/detectmatelibrary/parsers/template_matcher/_matcher_op.py
index 8760ab2..399edfd 100644
--- a/src/detectmatelibrary/parsers/template_matcher/_matcher_op.py
+++ b/src/detectmatelibrary/parsers/template_matcher/_matcher_op.py
@@ -1,8 +1,17 @@
 from collections import defaultdict
-from typing import Dict, List, Any, Tuple
+from typing import Dict, List, Any, Tuple, TypedDict
 import regex
 import re
 
+from detectmatelibrary.common._config._formats import (
+    EventsConfig, _EventConfig, _EventInstance, Variable
+)
+
+
+class TemplateMetadata(TypedDict):
+    event_id_label: str | None
+    labels: list[str]
+
 
 def safe_search(pattern: str, string: str, timeout: int = 1) -> regex.Match[str] | None:
     """Perform regex search with a timeout to prevent catastrophic
@@ -64,6 +73,7 @@ class TemplatesManager:
     def __init__(
         self,
         template_list: list[str],
+        metadata: dict[int, TemplateMetadata] | None = None,
         remove_spaces: bool = True,
         remove_punctuation: bool = True,
         lowercase: bool = True
@@ -96,6 +106,61 @@ def __init__(
             first = tokens[0] if tokens else ""
             self._prefix_index[first].append(idx)
 
+        _metadata: dict[int, TemplateMetadata] = metadata or {}
+        self._event_label_to_idx: dict[str, int] = {
+            m["event_id_label"]: i
+            for i, m in _metadata.items()
+            if m["event_id_label"]
+        }
+        self._idx_to_var_map: dict[int, dict[str, int]] = {
+            i: {label: pos for pos, label in enumerate(m["labels"])}
+            for i, m in _metadata.items()
+            if m["labels"]
+        }
+
+    def compile_events_config(self, events_config: EventsConfig) -> EventsConfig:
+        """Resolve named event IDs and named variable labels to positional
+        ints.
+
+        Translates user-friendly named format to the internal positional
+        representation. Returns a new EventsConfig with only int keys
+        and int positions.
+        """
+        new_events: Dict[Any, _EventConfig] = {}
+
+        for event_key, event_config in events_config.events.items():
+            if isinstance(event_key, str) and event_key in self._event_label_to_idx:
+                resolved_key: str | int = self._event_label_to_idx[event_key]
+            else:
+                resolved_key = event_key
+
+            var_map = self._idx_to_var_map.get(resolved_key if isinstance(resolved_key, int) else -1, {})
+
+            new_instances: Dict[str, _EventInstance] = {}
+            for instance_id, instance in event_config.instances.items():
+                new_vars: Dict[str | int, Variable] = {}
+                for pos, var in instance.variables.items():
+                    if isinstance(pos, str):
+                        if pos not in var_map:
+                            raise ValueError(
+                                f"Label '{pos}' not found in template for event '{event_key}'. "
+                                f"Available labels: {list(var_map)}"
+                            )
+                        resolved_pos = var_map[pos]
+                        new_vars[resolved_pos] = Variable(
+                            pos=resolved_pos, name=pos, params=var.params
+                        )
+                    else:
+                        new_vars[pos] = var
+                new_instances[instance_id] = _EventInstance(
+                    params=instance.params,
+                    variables=new_vars,
+                    header_variables=instance.header_variables,
+                )
+            new_events[resolved_key] = _EventConfig(instances=new_instances)
+
+        return EventsConfig(events=new_events)
+
     def candidate_indices(self, s: str) -> Tuple[str, List[int]]:
         pre_s = self.preprocess(s)
         candidates = []
@@ -110,17 +175,28 @@ class TemplateMatcher:
     def __init__(
         self,
         template_list: list[str],
+        metadata: dict[int, TemplateMetadata] | None = None,
         remove_spaces: bool = True,
         remove_punctuation: bool = True,
         lowercase: bool = True
     ) -> None:
         self.manager = TemplatesManager(
             template_list=template_list,
+            metadata=metadata,
             remove_spaces=remove_spaces,
             remove_punctuation=remove_punctuation,
             lowercase=lowercase
         )
 
+    def compile_detector_config(self, events_config: EventsConfig) -> EventsConfig:
+        """Resolve named event IDs and variable labels to positional ints.
+
+        Call once at setup time. Returns a new EventsConfig using the
+        internal positional representation, compatible with
+        get_configured_variables().
+        """
+        return self.manager.compile_events_config(events_config)
+
     @staticmethod
     def extract_parameters(log: str, template: str) -> tuple[str, ...] | None:
         """Extract parameters from the log based on the template."""
diff --git a/src/detectmatelibrary/parsers/template_matcher/_parser.py b/src/detectmatelibrary/parsers/template_matcher/_parser.py
index 67fb5a8..edcae99 100644
--- a/src/detectmatelibrary/parsers/template_matcher/_parser.py
+++ b/src/detectmatelibrary/parsers/template_matcher/_parser.py
@@ -1,10 +1,13 @@
-from detectmatelibrary.parsers.template_matcher._matcher_op import TemplateMatcher
+from detectmatelibrary.parsers.template_matcher._matcher_op import TemplateMatcher, TemplateMetadata
 from detectmatelibrary.common.parser import CoreParser, CoreParserConfig
 from detectmatelibrary import schemas
 
 from typing import Any
 import csv
 import os
+import re
+
+_NAMED_WC_RE = re.compile(r'<([A-Za-z_]\w*)>')
 
 
 class TemplatesNotFoundError(Exception):
@@ -15,34 +18,93 @@ class TemplateNoPermissionError(Exception):
     pass
 
 
-def load_templates(path: str) -> list[str]:
+def _compile_templates(
+    raw_templates: list[str],
+    event_id_labels: list[str | None] | None = None,
+) -> tuple[list[str], dict[int, TemplateMetadata]]:
+    """Convert named wildcards to <*> and record label order and event ID
+    labels.
+
+    Args:
+        raw_templates: Raw template strings, possibly containing named wildcards.
+        event_id_labels: Optional per-template event ID labels (from CSV EventId column).
+                         If provided, must have the same length as raw_templates.
+
+    Returns:
+        compiled: Template strings with only <*> wildcards, ready for TemplatesManager.
+        metadata: Mapping of template index to TemplateMetadata.
+
+    Raises:
+        ValueError: If a template mixes <*> and named wildcards.
+    """
+    compiled: list[str] = []
+    metadata: dict[int, TemplateMetadata] = {}
+
+    for i, raw in enumerate(raw_templates):
+        has_anon = "<*>" in raw
+        labels = _NAMED_WC_RE.findall(raw)
+        has_named = bool(labels)
+
+        if has_anon and has_named:
+            raise ValueError(
+                f"Template mixes <*> and named wildcards: {raw!r}. "
+                "Use either <*> (positional) or <label> (named) exclusively."
+            )
+
+        compiled_tpl = _NAMED_WC_RE.sub("<*>", raw) if has_named else raw
+        idx = len(compiled)
+        compiled.append(compiled_tpl)
+        eid_label = event_id_labels[i] if event_id_labels else None
+        metadata[idx] = TemplateMetadata(event_id_label=eid_label, labels=labels)
+
+    return compiled, metadata
+
+
+def load_templates(path: str) -> tuple[list[str], list[str | None]]:
+    """Load templates from a .txt or .csv file.
+
+    Returns:
+        A tuple of (template_strings, event_id_labels). For .txt files, all
+        event_id_labels are None (positional IDs only). For .csv files, an
+        optional EventId column provides named event ID labels.
+    """
     if not os.path.exists(path):
         raise TemplatesNotFoundError(f"Templates file not found at: {path}")
     if not os.access(path, os.R_OK):
         raise TemplateNoPermissionError(
             f"You do not have the permission to access the templates file: {path}"
         )
+    templates: list[str] = []
+    eid_labels: list[str | None] = []
     if path.endswith(".txt"):
         with open(path, "r") as f:
-            templates = [line.strip() for line in f if line.strip()]
+            for line in f:
+                s = line.strip()
+                if s:
+                    templates.append(s)
+                    eid_labels.append(None)
     elif path.endswith(".csv"):
-        templates = []
-        # Use the lightweight built-in csv module instead of pandas
-        # Expect a header with a 'template' column
         with open(path, newline="", encoding="utf-8") as f:
             reader = csv.DictReader(f)
             if reader.fieldnames is None or "EventTemplate" not in reader.fieldnames:
                 raise ValueError("CSV file must contain a 'EventTemplate' column.")
+            has_event_id_col = "EventId" in (reader.fieldnames or [])
             for row in reader:
                 val = row.get("EventTemplate")
                 if val is None:
                     continue
                 s = str(val).strip()
-                if s:
-                    templates.append(s)
+                if not s:
+                    continue
+                templates.append(s)
+                if has_event_id_col:
+                    eid = str(row.get("EventId", "")).strip()
+                    eid_labels.append(eid or None)
+                else:
+                    eid_labels.append(None)
     else:
         raise ValueError("Unsupported template file format. Use .txt or .csv files.")
-    return templates
+    return templates, eid_labels
 
 
 class MatcherParserConfig(CoreParserConfig):
@@ -67,11 +129,14 @@ def __init__(
         super().__init__(name=name, config=config)
         self.config: MatcherParserConfig
 
-        templates = load_templates(
-            self.config.path_templates
-        ) if self.config.path_templates is not None else []
+        if self.config.path_templates is not None:
+            raw_templates, eid_labels = load_templates(self.config.path_templates)
+        else:
+            raw_templates, eid_labels = [], []
+        compiled_templates, metadata = _compile_templates(raw_templates, eid_labels)
         self.template_matcher = TemplateMatcher(
-            template_list=templates,
+            template_list=compiled_templates,
+            metadata=metadata,
             remove_spaces=self.config.remove_spaces,
             remove_punctuation=self.config.remove_punctuation,
             lowercase=self.config.lowercase,
diff --git a/tests/test_common/test_config_roundtrip.py b/tests/test_common/test_config_roundtrip.py
index 24359ff..24fee61 100644
--- a/tests/test_common/test_config_roundtrip.py
+++ b/tests/test_common/test_config_roundtrip.py
@@ -290,3 +290,90 @@ def test_parser_true_roundtrip(self):
 
         # Dicts should be identical
         assert dict1 == dict2
+
+
+class TestNamedVariablesRoundtrip:
+    """Test that named wildcard positions and named event IDs round-trip
+    correctly."""
+
+    def test_named_pos_preserved_as_string(self):
+        """String pos values must survive yaml -> pydantic -> yaml
+        unchanged."""
+        config_yaml = load_test_config()
+        method_id = "detector_named_pos"
+
+        config = MockupDetectorConfig.from_dict(config_yaml, method_id)
+        result_dict = config.to_dict(method_id)
+
+        original = config_yaml["detectors"][method_id]
+        result = result_dict["detectors"][method_id]
+
+        orig_vars = original["events"][1]["example_detector_1"]["variables"]
+        result_vars = result["events"][1]["example_detector_1"]["variables"]
+
+        assert len(result_vars) == len(orig_vars)
+        for orig_var, result_var in zip(orig_vars, result_vars):
+            assert result_var["pos"] == orig_var["pos"]
+            assert isinstance(result_var["pos"], str)  # must stay a string, not coerced to int
+
+    def test_named_pos_no_name_field_when_omitted(self):
+        """Variables with only a label pos must not emit a 'name' key."""
+        config_yaml = load_test_config()
+        config = MockupDetectorConfig.from_dict(config_yaml, "detector_named_pos")
+        result_dict = config.to_dict("detector_named_pos")
+
+        instance = result_dict["detectors"]["detector_named_pos"]["events"][1]["example_detector_1"]
+        no_params_var = instance["variables"][0]  # pos: pid, no params
+
+        assert "pos" in no_params_var
+        assert no_params_var["pos"] == "pid"
+        assert "name" not in no_params_var
+
+    def test_named_pos_params_preserved(self):
+        """Named pos variable with params must preserve those params."""
+        config_yaml = load_test_config()
+        config = MockupDetectorConfig.from_dict(config_yaml, "detector_named_pos")
+        result_dict = config.to_dict("detector_named_pos")
+
+        instance = result_dict["detectors"]["detector_named_pos"]["events"][1]["example_detector_1"]
+        with_params_var = instance["variables"][1]  # pos: op, params: {threshold: 0.5}
+
+        assert with_params_var["pos"] == "op"
+        assert with_params_var["params"] == {"threshold": 0.5}
+
+    def test_named_event_id_preserved_as_string(self):
+        """String event ID keys must survive yaml -> pydantic -> yaml
+        unchanged."""
+        config_yaml = load_test_config()
+        method_id = "detector_named_event_id"
+
+        config = MockupDetectorConfig.from_dict(config_yaml, method_id)
+        result_dict = config.to_dict(method_id)
+
+        result_events = result_dict["detectors"][method_id]["events"]
+        assert "login_failure" in result_events
+        assert isinstance(list(result_events.keys())[0], str)
+
+    def test_named_pos_true_roundtrip(self):
+        """Yaml -> pydantic -> yaml -> pydantic produces identical objects."""
+        config_yaml = load_test_config()
+        method_id = "detector_named_pos"
+
+        config1 = MockupDetectorConfig.from_dict(config_yaml, method_id)
+        dict1 = config1.to_dict(method_id)
+        config2 = MockupDetectorConfig.from_dict(dict1, method_id)
+        dict2 = config2.to_dict(method_id)
+
+        assert dict1 == dict2
+
+    def test_named_event_id_true_roundtrip(self):
+        """Yaml -> pydantic -> yaml -> pydantic produces identical objects."""
+        config_yaml = load_test_config()
+        method_id = "detector_named_event_id"
+
+        config1 = MockupDetectorConfig.from_dict(config_yaml, method_id)
+        dict1 = config1.to_dict(method_id)
+        config2 = MockupDetectorConfig.from_dict(dict1, method_id)
+        dict2 = config2.to_dict(method_id)
+
+        assert dict1 == dict2
diff --git a/tests/test_folder/test_config.yaml b/tests/test_folder/test_config.yaml
index ff679b1..0c38f27 100644
--- a/tests/test_folder/test_config.yaml
+++ b/tests/test_folder/test_config.yaml
@@ -165,3 +165,35 @@ detectors:
     method_type: new_value_combo_detector
     parser: example_parser_1
     auto_config: true
+
+  detector_named_pos:
+    method_type: ExampleDetector
+    parser: example_parser_1
+    auto_config: false
+    params: {}
+    events:
+      1:  # positional event_id; named wildcards in .txt templates use int event keys
+        example_detector_1:
+          params: {}
+          variables:
+            - pos: pid  # named label; no 'name' field needed
+            - pos: op
+              params:
+                threshold: 0.5
+          header_variables:
+            - pos: Level
+              params:
+                threshold: 0.2
+
+  detector_named_event_id:
+    method_type: ExampleDetector
+    parser: example_parser_1
+    auto_config: false
+    params: {}
+    events:
+      login_failure:  # named event_id; only valid for CSV templates with an EventId column
+        example_detector_1:
+          params: {}
+          variables:
+            - pos: pid
+            - pos: uid
diff --git a/tests/test_folder/test_named_templates.csv b/tests/test_folder/test_named_templates.csv
new file mode 100644
index 0000000..3ce1622
--- /dev/null
+++ b/tests/test_folder/test_named_templates.csv
@@ -0,0 +1,2 @@
+EventId,EventTemplate
+login_failure,"pid=<pid> uid=<uid> auid=<auid> ses=<ses> msg='op=PAM:<op> acct=<acct>"
diff --git a/tests/test_folder/test_named_templates.txt b/tests/test_folder/test_named_templates.txt
new file mode 100644
index 0000000..1ef130f
--- /dev/null
+++ b/tests/test_folder/test_named_templates.txt
@@ -0,0 +1 @@
+pid=<pid> uid=<uid> auid=<auid> ses=<ses> msg='op=PAM:<op> acct=<acct>
diff --git a/tests/test_parsers/test_template_matcher.py b/tests/test_parsers/test_template_matcher.py
index 04e4bdc..9e5cf5c 100644
--- a/tests/test_parsers/test_template_matcher.py
+++ b/tests/test_parsers/test_template_matcher.py
@@ -3,6 +3,8 @@
 from detectmatelibrary.parsers.template_matcher import (
     MatcherParser, MatcherParserConfig, TemplatesNotFoundError
 )
+from detectmatelibrary.parsers.template_matcher._parser import _compile_templates
+from detectmatelibrary.common._config._formats import EventsConfig
 from detectmatelibrary import schemas
 
 test_template = [
@@ -95,3 +97,147 @@ def test_custom_preprocessing_flags(self):
         parser.parse(input_log, output_data)
         # Still matches template even with preprocessing disabled
         assert output_data.template == test_template[0]
+
+
+class TestCompileTemplates:
+    def test_named_wildcards_compiled_to_anon(self):
+        raw = ["pid=<pid> uid=<uid> auid=<auid>"]
+        compiled, metadata = _compile_templates(raw)
+        assert compiled == ["pid=<*> uid=<*> auid=<*>"]
+        assert metadata[0]["labels"] == ["pid", "uid", "auid"]
+        assert metadata[0]["event_id_label"] is None
+
+    def test_anonymous_wildcards_unchanged(self):
+        raw = ["pid=<*> uid=<*>"]
+        compiled, metadata = _compile_templates(raw)
+        assert compiled == ["pid=<*> uid=<*>"]
+        assert metadata[0]["labels"] == []
+        assert metadata[0]["event_id_label"] is None
+
+    def test_event_id_label_stored(self):
+        raw = ["pid=<pid> uid=<uid>"]
+        eid_labels: list[str | None] = ["login_failure"]
+        compiled, metadata = _compile_templates(raw, eid_labels)
+        assert compiled == ["pid=<*> uid=<*>"]
+        assert metadata[0]["event_id_label"] == "login_failure"
+        assert metadata[0]["labels"] == ["pid", "uid"]
+
+    def test_mixing_raises_value_error(self):
+        raw = ["pid=<*> uid=<uid>"]
+        with pytest.raises(ValueError, match="mixes"):
+            _compile_templates(raw)
+
+    def test_multiple_templates(self):
+        raw = ["pid=<pid> uid=<uid>", "arch=<*> syscall=<*>"]
+        compiled, metadata = _compile_templates(raw)
+        assert compiled[0] == "pid=<*> uid=<*>"
+        assert compiled[1] == "arch=<*> syscall=<*>"
+        assert metadata[0]["labels"] == ["pid", "uid"]
+        assert metadata[1]["labels"] == []
+
+
+class TestNamedWildcardsTxt:
+    """Named wildcards in .txt template files (event IDs remain positional)."""
+
+    named_template_compiled = "pid=<*> uid=<*> auid=<*> ses=<*> msg='op=PAM:<*> acct=<*>"
+
+    def test_match_produces_correct_variables(self):
+        config = MatcherParserConfig(path_templates="tests/test_folder/test_named_templates.txt")
+        parser = MatcherParser(name="MatcherParser", config=config)
+        input_log = schemas.LogSchema({"log": test_log_match})
+        output_data = schemas.ParserSchema()
+        parser.parse(input_log, output_data)
+
+        assert output_data.template == self.named_template_compiled
+        assert output_data.EventID == 0
+        assert len(output_data.variables) == 6
+
+    def test_compile_resolves_named_positions(self):
+        config = MatcherParserConfig(path_templates="tests/test_folder/test_named_templates.txt")
+        parser = MatcherParser(name="MatcherParser", config=config)
+
+        raw_events: dict = {
+            0: {
+                "my_detector": {
+                    "params": {},
+                    "variables": [
+                        {"pos": "pid"},
+                        {"pos": "op"},
+                    ]
+                }
+            }
+        }
+        events_config = EventsConfig._init(raw_events)
+        compiled = parser.template_matcher.compile_detector_config(events_config)
+
+        # pid is label 0 → pos 0; op is label 4 → pos 4
+        resolved_vars = compiled.events[0].variables
+        assert 0 in resolved_vars
+        assert resolved_vars[0].name == "pid"
+        assert 4 in resolved_vars
+        assert resolved_vars[4].name == "op"
+
+    def test_compile_unknown_label_raises(self):
+        config = MatcherParserConfig(path_templates="tests/test_folder/test_named_templates.txt")
+        parser = MatcherParser(name="MatcherParser", config=config)
+
+        raw_events: dict = {
+            0: {"d": {"params": {}, "variables": [{"pos": "nonexistent"}]}}
+        }
+        events_config = EventsConfig._init(raw_events)
+        with pytest.raises(ValueError, match="nonexistent"):
+            parser.template_matcher.compile_detector_config(events_config)
+
+
+class TestNamedEventIdCsv:
+    """Named event IDs via CSV EventId column."""
+
+    named_template_compiled = "pid=<*> uid=<*> auid=<*> ses=<*> msg='op=PAM:<*> acct=<*>"
+
+    def test_match_produces_correct_variables(self):
+        config = MatcherParserConfig(path_templates="tests/test_folder/test_named_templates.csv")
+        parser = MatcherParser(name="MatcherParser", config=config)
+        input_log = schemas.LogSchema({"log": test_log_match})
+        output_data = schemas.ParserSchema()
+        parser.parse(input_log, output_data)
+
+        assert output_data.template == self.named_template_compiled
+        assert output_data.EventID == 0
+        assert len(output_data.variables) == 6
+
+    def test_compile_resolves_named_event_key(self):
+        config = MatcherParserConfig(path_templates="tests/test_folder/test_named_templates.csv")
+        parser = MatcherParser(name="MatcherParser", config=config)
+
+        raw_events: dict = {
+            "login_failure": {
+                "my_detector": {
+                    "params": {},
+                    "variables": [
+                        {"pos": "pid"},
+                        {"pos": "uid"},
+                    ]
+                }
+            }
+        }
+        events_config = EventsConfig._init(raw_events)
+        compiled = parser.template_matcher.compile_detector_config(events_config)
+
+        # "login_failure" → int key 0
+        assert 0 in compiled.events
+        resolved_vars = compiled.events[0].variables
+        assert 0 in resolved_vars and resolved_vars[0].name == "pid"
+        assert 1 in resolved_vars and resolved_vars[1].name == "uid"
+
+    def test_positional_int_event_key_still_works(self):
+        config = MatcherParserConfig(path_templates="tests/test_folder/test_named_templates.csv")
+        parser = MatcherParser(name="MatcherParser", config=config)
+
+        raw_events: dict = {
+            0: {"d": {"params": {}, "variables": [{"pos": 0, "name": "process_id"}]}}
+        }
+        events_config = EventsConfig._init(raw_events)
+        compiled = parser.template_matcher.compile_detector_config(events_config)
+
+        assert 0 in compiled.events
+        assert compiled.events[0].variables[0].name == "process_id"