fix: keep validation time linear on deeply nested or repeated input

.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://anthropic.com/claude-code/plugin.schema.json",
   "name": "plc-programming",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "PLC programming across IEC 61131-3 (ST, LD, FBD, SFC, IL), PLCopen XML / IEC 61131-10, Siemens TIA (SCL, SimaticML, SIMATIC SD), Schneider (Machine Expert, Control Expert), CODESYS, TwinCAT, ABB Freelance, Rockwell L5X/L5K, IEC 61499, AutomationML, DEXPI, PackML/ISA-88 and OPC UA: design, write, convert, review and debug programs, with a stdlib-only structural validator.",
   "author": {
     "name": "SmailG",

CONTRIBUTING.md

@@ -14,7 +14,7 @@ privately, as described in [SECURITY.md](SECURITY.md). Pull requests use the tem
 | `scripts/plc_validate.py` | Validator CLI and hook entry point |
 | `scripts/codesys_project.py` | Offline reader for CODESYS 3.5 / Machine Expert `.project` files (`list`, `st`, `dump-st`, `graph`, `info`, `diff`, `xref`); `scripts/codesysproj/` holds `binary.py` (varint, string table), `project.py` (objects, kinds, text), `graph.py` (LD/FBD records) and `graphview.py` (their text and JSON), `export.py` (`dump-st`), `buildinfo.py` (`info`), `diff.py` (`diff`), `xref.py` and `stscan.py` (`xref`) |
 | `scripts/plccheck/` | One module per format (`st.py`, `plcopen.py`, `simaticml.py`, `twincat.py`, `logix.py`, `schneider.py`, `iec61499.py`, `aml.py`, `dexpi.py`, `tabular.py`); `findings.py` holds the finding type, `xmlutil.py` the line-numbered XML parser |
-| `tests/` | `test_st.py`, `test_formats.py` (validator rules), `test_codesys_project.py` and `test_codesys_tools.py` (the `.project` reader and its `info`, `diff` and `xref`, on synthetic projects only), `test_skills.py` (frontmatter, templates, every `iecst` block lints clean) |
+| `tests/` | `test_st.py`, `test_formats.py` (validator rules), `test_codesys_project.py` and `test_codesys_tools.py` (the `.project` reader and its `info`, `diff` and `xref`, on synthetic projects only), `test_codesys_hardening.py` (hostile and damaged `.project` input and the reader's CLI contracts, synthetic projects only), `test_linear_xml.py`, `test_linear_aml.py` and `test_linear_text.py` (CPU-time scaling and ordering of the format checkers on crafted input; `linear_support.py` holds their ratio helper and XML builders), `test_skills.py` (frontmatter, templates, every `iecst` block lints clean) |
 | `evals/` | `claude plugin eval` cases comparing the plugin against a no-plugin baseline |
 
 ## Rules

plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "plc-programming",
   "displayName": "PLC Programming",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "PLC programming across IEC 61131-3 (ST, LD, FBD, SFC, IL), PLCopen XML / IEC 61131-10, Siemens TIA, Schneider, CODESYS, TwinCAT, ABB Freelance, Rockwell L5X/L5K, IEC 61499, AutomationML, DEXPI, PackML/ISA-88 and OPC UA with a stdlib-only structural validator.",
   "author": {
     "name": "SmailG",

scripts/plccheck/aml.py

@@ -21,6 +21,23 @@ def _resolve(ref: str, base: Path) -> tuple[Path | None, str]:
     return (base / p) if p else None, u.fragment
 
 
+def _global_ids(target: Path, cache: dict[Path, set[str] | None]) -> set[str] | None:
+    """The globalId and uuid values of a referenced PLCopen file, or None if it is not well-formed XML.
+
+    Parsed once per file: every interface that links to the same file would otherwise parse and walk
+    it again, so a file of n links to itself costs n parses of an n-link document.
+    """
+    key = target.resolve()  # `x.xml`, `./x.xml` and `a/../x.xml` are one file
+    if key not in cache:
+        try:
+            doc = parse_file(target)
+        except XmlError:
+            cache[key] = None
+        else:
+            cache[key] = {v for n in doc.iter() for v in (n.get("globalId"), n.get("uuid")) if v is not None}
+    return cache[key]
+
+
 def check_aml(root: Node, path: Path | None = None) -> list[Finding]:
     out: list[Finding] = []
     if root.tag != "CAEXFile":
@@ -56,6 +73,7 @@ def check_aml(root: Node, path: Path | None = None) -> list[Finding]:
                 out.append(warning("AM005", f"InternalLink {side} '{ref}' matches no interface ID in this file", link.line))
 
     base = path.parent if path is not None else None
+    plcopen_ids: dict[Path, set[str] | None] = {}
     for iface in root.iter("ExternalInterface"):
         cls = iface.get("RefBaseClassPath") or ""
         if not any(cls.endswith(c) for c in DATA_CONNECTORS):
@@ -74,11 +92,8 @@ def check_aml(root: Node, path: Path | None = None) -> list[Finding]:
             out.append(warning("AM011", f"refURI target {target.name} not found next to this file", value.line))
         elif fragment and cls.endswith("PLCopenXMLInterface") and target.is_file():
             # An absolute refURI can name a directory, FIFO or device; only a regular file is read.
-            try:
-                doc = parse_file(target)
-            except XmlError:
-                continue
-            if not any(n.get("globalId") == fragment or n.get("uuid") == fragment for n in doc.iter()):
+            ids = _global_ids(target, plcopen_ids)
+            if ids is not None and fragment not in ids:
                 out.append(warning("AM012", f"refURI fragment #{fragment} matches no globalId in {target.name}", value.line))
     for ext in root.findall("ExternalReference"):
         if base is not None and ext.get("Path") and not (base / ext.get("Path")).exists():

scripts/plccheck/logix.py

@@ -3,6 +3,9 @@
 from __future__ import annotations
 
 import re
+from bisect import bisect_left
+from collections import Counter
+from itertools import tee, zip_longest
 
 from .findings import Finding, error, warning, with_context
 from .st import lint_st
@@ -23,23 +26,66 @@
     "JMP": 1, "LBL": 1, "AFI": 0, "NOP": 0, "TND": 0, "MCR": 0, "UID": 0, "UIE": 0,
     "GSV": 4, "SSV": 4, "MSG": 1, "PSC": 0, "PCMD": 3,
 }
-_INSTR = re.compile(r"([A-Za-z_][A-Za-z0-9_]*)\(")
-
-
-def _split_operands(s: str) -> list[str]:
-    parts, depth, cur = [], 0, ""
-    for ch in s:
-        if ch in "([":
+# `NAME(`, found at the start of an identifier run (leading digits skipped). Without the lookbehind every
+# position of a long run would rescan it for a `(`.
+_INSTR = re.compile(r"(?<![A-Za-z0-9_])[0-9]*([A-Za-z_][A-Za-z0-9_]*)\(")
+
+
+def _instructions(t: str) -> list[tuple[str, int, str | None]]:
+    """(mnemonic, operand count, first operand) for each `NAME(` in the neutral text, in order.
+
+    The operands are the comma-separated parts, at bracket depth 0, between the instruction's
+    parentheses; a missing `)` runs to the end of the text less its last character. The count is 0
+    when every part is blank. The first operand is "" when there is none, and None when it holds
+    another instruction (a `NAME(`): that is a call, not a tag, and keeping a copy of it for every
+    instruction around it would grow with the square of the nesting depth. A bare `(` does not
+    count, so `Arr[(i+1)]` stays an operand.
+
+    One pass over the text: parentheses matched with a stack, commas kept by bracket depth, and the
+    operands of each instruction read from those, so no instruction rescans its own operand text.
+    """
+    n = len(t)
+    depth_before = [0] * (n + 1)  # bracket depth ( [ ) ] before each character
+    text_before = [0] * (n + 1)  # non-whitespace characters before each character
+    close: dict[int, int] = {}  # index of each `(` -> index of its `)`
+    commas: dict[int, list[int]] = {}  # bracket depth -> indexes of the commas at that depth
+    opened: list[int] = []
+    depth = chars = 0
+    for k, ch in enumerate(t):
+        depth_before[k], text_before[k] = depth, chars
+        if ch == ",":
+            commas.setdefault(depth, []).append(k)
+        elif ch == "(":
+            opened.append(k)
+            depth += 1
+        elif ch == ")":
+            if opened:
+                close[opened.pop()] = k
+            depth -= 1
+        elif ch == "[":
             depth += 1
-        elif ch in ")]":
+        elif ch == "]":
             depth -= 1
-        if ch == "," and depth == 0:
-            parts.append(cur)
-            cur = ""
-        else:
-            cur += ch
-    parts.append(cur)
-    return [p.strip() for p in parts] if any(p.strip() for p in parts) else []
+        if not ch.isspace():
+            chars += 1
+    depth_before[n], text_before[n] = depth, chars
+    found: list[tuple[str, int, str | None]] = []
+    # Each match is paired with the next. Matches do not overlap, so only the next one can lie inside this one's operands.
+    current, following = tee(_INSTR.finditer(t))
+    next(following, None)
+    for m, after in zip_longest(current, following):
+        start = m.end()
+        end = close.get(start - 1, n - 1)
+        top = commas.get(depth_before[start], [])  # a comma at the depth the operands start at is a separator
+        first, last = bisect_left(top, start), bisect_left(top, end)
+        separators = max(last - first, 0)
+        has_operand = end > start and text_before[end] - text_before[start] > separators
+        first_operand: str | None = ""
+        if has_operand:
+            stop = top[first] if separators else end
+            first_operand = None if after is not None and after.end() <= stop else t[start:stop].strip()
+        found.append((m.group(1).upper(), separators + 1 if has_operand else 0, first_operand))
+    return found
 
 
 def check_rung(text: str, line: int, ctx: str) -> tuple[list[Finding], list[str]]:
@@ -69,20 +115,29 @@ def check_rung(text: str, line: int, ctx: str) -> tuple[list[Finding], list[str]
     if bracket:
         out.append(error("LX021", "unbalanced branch brackets", line, ctx))
     otes: list[str] = []
-    for m in _INSTR.finditer(t):
-        start, depth, i = m.end(), 1, m.end()
-        while i < len(t) and depth:
-            depth += {"(": 1, ")": -1}.get(t[i], 0)
-            i += 1
-        ops = _split_operands(t[start:i - 1])
-        mnem = m.group(1).upper()
-        if mnem in ARITY and len(ops) != ARITY[mnem]:
-            out.append(warning("LX022", f"{mnem} takes {ARITY[mnem]} operand(s), found {len(ops)}", line, ctx))
-        if mnem == "OTE" and ops:
-            otes.append(ops[0])
+    for mnem, count, first_operand in _instructions(t):
+        if mnem in ARITY and count != ARITY[mnem]:
+            out.append(warning("LX022", f"{mnem} takes {ARITY[mnem]} operand(s), found {count}", line, ctx))
+        if mnem == "OTE" and count and first_operand is not None:
+            otes.append(first_operand)
     return out, otes
 
 
+def _own_rungs(routine: Node):
+    """Rung elements of this routine in document order, not descending into a nested <Routines>.
+
+    A nested <Routines> is visited by the caller as a scope of its own. Iterating the whole subtree
+    per routine would check a nested rung once for every routine around it.
+    """
+    stack = list(reversed(routine.children))
+    while stack:
+        node = stack.pop()
+        if node.tag == "Rung":
+            yield node
+        if node.tag != "Routines":
+            stack.extend(reversed(node.children))
+
+
 def check_l5x(root: Node) -> list[Finding]:
     out: list[Finding] = []
     if root.tag != "RSLogix5000Content":
@@ -115,7 +170,7 @@ def check_l5x(root: Node) -> list[Finding]:
             if rtype not in ROUTINE_TYPES:
                 out.append(warning("LX010", f"unknown routine Type '{rtype}'", r.line, ctx))
             if rtype == "RLL":
-                for rung in r.iter("Rung"):
+                for rung in _own_rungs(r):
                     text_node = rung.find("Text")
                     if text_node is None:
                         continue
@@ -147,6 +202,7 @@ def check_l5k(text: str) -> list[Finding]:
     rungs are `N: neutral text;`; ST routine lines start with a single quote."""
     out: list[Finding] = []
     stack: list[tuple[str, int]] = []
+    open_words: Counter[str] = Counter()  # open frames per keyword, so "is X open" is not a stack scan
     ends = set(re.findall(r"\bEND_([A-Z][A-Z0-9_]*)\b", text))
     st_lines: list[tuple[int, str]] = []
     for lineno, raw in enumerate(text.splitlines(), 1):
@@ -165,18 +221,20 @@ def check_l5k(text: str) -> list[Finding]:
             if not end:
                 if word in L5K_BLOCKS or word in ends:
                     stack.append((word, lineno))
+                    open_words[word] += 1
                 continue
-            pos = next((k for k in range(len(stack) - 1, -1, -1) if stack[k][0] == word), None)
-            if pos is None:
+            if not open_words[word]:  # else the scan would walk the whole stack to find nothing
                 out.append(error("LX050", f"END_{word} without an open {word}", lineno))
                 continue
+            pos = next(k for k in range(len(stack) - 1, -1, -1) if stack[k][0] == word)
             for kw, ln in stack[pos + 1:]:
                 out.append(error("LX050", f"{kw} opened at line {ln} is not closed before END_{word}", ln))
             closed = stack[pos]
+            for kw, _ in stack[pos:]:
+                open_words[kw] -= 1
             del stack[pos:]
             if closed[0] == "ST_ROUTINE" and st_lines:
-                src = "\n" * (st_lines[0][0] - 1) + "\n".join(t for _, t in st_lines)
-                f, _ = lint_st(src, "body")
+                f, _ = lint_st("\n".join(t for _, t in st_lines), "body", st_lines[0][0])
                 out.extend(with_context(f, "L5K ST routine"))
                 st_lines = []
     for kw, ln in stack:

scripts/plccheck/plcopen.py

@@ -242,15 +242,38 @@ def check_iec61131_10(root: Node) -> list[Finding]:
         return [error("PX101", f"root element is <{root.tag}>, expected <Project>", root.line)]
     if root.get("schemaVersion") is None:
         out.append(warning("PX102", "<Project> has no schemaVersion", root.line))
-    for bc in root.iter("BodyContent"):
+    for bc, outs, conns in _bodies_with_connections(root):
         kind = (bc.get("type") or "").rsplit(":", 1)[-1]
         st = bc.find("ST")
         if kind == "ST" and st is not None:
             f, _ = lint_st(st.text, "body")
             out.extend(with_context(f, "IEC 61131-10 ST body", (st.text_line or st.line) - 1))
-        outs = {n.get("connectionPointOutId") for n in bc.iter() if n.get("connectionPointOutId")}
-        for conn in bc.iter("Connection"):
+        for conn in conns:
             ref = conn.get("refConnectionPointOutId")
             if ref and ref not in outs:
                 out.append(warning("PX121", f"Connection refers to connectionPointOutId {ref}, not found in this body", conn.line))
     return out
+
+
+def _bodies_with_connections(root: Node) -> list[tuple[Node, set[str], list[Node]]]:
+    """Each BodyContent in document order, with its connectionPointOutId values and Connection elements.
+
+    One walk over the document: a node counts for its nearest enclosing BodyContent. Scanning the
+    subtree per BodyContent is quadratic when BodyContent elements nest, which a valid file never
+    does. For such a file the scope of PX121 is the nearest body, not every body around it.
+    """
+    bodies: list[tuple[Node, set[str], list[Node]]] = []
+    stack: list[tuple[Node, tuple[Node, set[str], list[Node]] | None]] = [(root, None)]
+    while stack:
+        node, current = stack.pop()
+        if node.tag == "BodyContent":
+            current = (node, set(), [])
+            bodies.append(current)
+        if current is not None:
+            out_id = node.get("connectionPointOutId")
+            if out_id:
+                current[1].add(out_id)
+            if node.tag == "Connection":
+                current[2].append(node)
+        stack.extend((child, current) for child in reversed(node.children))
+    return bodies

scripts/plccheck/schneider.py

@@ -10,6 +10,22 @@
 ROOTS = {"FEFExchangeFile", "VariablesExchangeFile"}
 
 
+def _own_st_sources(fb: Node):
+    """STSource elements of this DFB, in document order, not descending into a nested FBSource.
+
+    A nested FBSource is visited by the caller's own loop. Iterating the whole subtree per
+    FBSource would lint an STSource once for every FBSource around it.
+    """
+    stack = list(reversed(fb.children))
+    while stack:
+        node = stack.pop()
+        if node.tag == "FBSource":
+            continue
+        if node.tag == "STSource":
+            yield node
+        stack.extend(reversed(node.children))
+
+
 def check_control_expert(root: Node) -> list[Finding]:
     out: list[Finding] = []
     if root.tag not in ROOTS:
@@ -40,7 +56,7 @@ def check_control_expert(root: Node) -> list[Finding]:
                 if name.upper() in seen:
                     out.append(error("CX012", f"DFB parameter/variable '{name}' declared twice", var.line, ctx))
                 seen[name.upper()] = var.line
-        for st in fb.iter("STSource"):
+        for st in _own_st_sources(fb):
             f, _ = lint_st(st.text, "body")
             out.extend(with_context(f, f"{ctx} (ST)", (st.text_line or st.line) - 1))