fix(analyzer): index variable shell-flag scopes once per window

src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py

@@ -44,15 +44,16 @@
     LINE_BREAK_CHARS,
     MARKDOWN_FENCE_CLOSE,
     MARKDOWN_FENCE_OPEN,
-    get_context,
-    get_line_number,
+    SourceLocationIndex,
+    get_context_from_lines,
     is_reference_material,
 )
 from .pattern_defaults import PatternCategory
 
 logger = get_logger(__name__)
 
 ANALYZER_ID = "static_patterns_tool_misuse"
+USES_RUNTIME_CHECK = True
 ANALYZE_USES_POSTPROCESS = True
 POSTPROCESS_USES_PYTHON_AST = True
 _VARIABLE_SHELL_FLAG_EVIDENCE = "_tm1_variable_shell_flag"
@@ -4715,16 +4716,22 @@ def analyze(
     file_path: str,
     file_type: str,
     *,
+    check_runtime: Callable[[], None] | None = None,
     defer_variable_reconciliation: bool = False,
 ) -> list[AnalyzerFinding]:
     """Analyze content for tool misuse patterns (TM1–TM3)."""
+    runtime_check = check_runtime or (lambda: None)
+    runtime_check()
     findings: list[AnalyzerFinding] = []
+    locations = SourceLocationIndex(content, file_path)
+    lines = content.splitlines()
 
     def loc(ln: int) -> Location:
         return Location(file=file_path, start_line=ln)
 
     def ctx(start: int) -> str:
-        return get_context(content, start)
+        line, column = locations.line_and_column(start)
+        return get_context_from_lines(lines, line, column=column)
 
     tag = [PatternCategory.TOOL_MISUSE.value]
     tm1_findings_by_key: dict[tuple[int, str, int], AnalyzerFinding] = {}
@@ -4735,10 +4742,14 @@ def ctx(start: int) -> str:
     }
     invisible_variable_matches: set[tuple[int, int]] = set()
     if file_type == "python" and variable_matches and not defer_variable_reconciliation:
+        runtime_check()
         parsed = parse_python_source(content, file_path)
+        runtime_check()
         ast_index = _build_variable_shell_ast_index(parsed)
+        runtime_check()
         if ast_index is not None:
             for span, (_, match) in variable_matches.items():
+                runtime_check()
                 assignment_line = bisect_right(ast_index.line_character_starts, match.start(1))
                 candidate = _resolve_variable_shell_candidate(ast_index, match, assignment_line)
                 if candidate is not None and (
@@ -4750,19 +4761,20 @@ def ctx(start: int) -> str:
                     invisible_variable_matches.add(span)
 
     shell_content = (
-        _perl_literal_print_shell_text(content, lambda: None) if file_type == "perl" else None
+        _perl_literal_print_shell_text(content, runtime_check) if file_type == "perl" else None
     )
     for match_start, match_end, matched_text, confidence in _tm1_candidates(
         content, shell_content=shell_content
     ):
+        runtime_check()
         variable_match = variable_matches.get((match_start, match_end))
         if variable_match is not None and variable_match[0].casefold().startswith("true"):
             # The case-insensitive direct ``shell=True`` pattern already owns
             # true-prefixed names at the exact call location.
             continue
         if variable_match is not None and (match_start, match_end) in invisible_variable_matches:
             continue
-        line_num = get_line_number(content, match_start)
+        line_num = locations.line_and_column(match_start)[0]
         context_text = ctx(match_start)
         matched = matched_text[:200]
         matched_line = _line_containing(content, match_start, match_end)
@@ -4807,7 +4819,8 @@ def ctx(start: int) -> str:
             else re.finditer
         )
         for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE):
-            line_num = get_line_number(content, match.start())
+            runtime_check()
+            line_num = locations.line_and_column(match.start())[0]
             context_text = ctx(match.start())
             matched = match.group(0)[:200]
 
@@ -4837,7 +4850,8 @@ def ctx(start: int) -> str:
             else re.finditer
         )
         for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE):
-            line_num = get_line_number(content, match.start())
+            runtime_check()
+            line_num = locations.line_and_column(match.start())[0]
             findings.append(
                 AnalyzerFinding(
                     rule_id="TM3",
@@ -4859,7 +4873,8 @@ def ctx(start: int) -> str:
     tm4_tags = [*tag, "contextual-triage", "likely-benign-context"] if reference_material else tag
     for pattern, confidence in TM4_PATTERNS:
         for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE):
-            line_num = get_line_number(content, match.start())
+            runtime_check()
+            line_num = locations.line_and_column(match.start())[0]
             findings.append(
                 AnalyzerFinding(
                     rule_id="TM4",

tests/unit/test_patterns_new.py

@@ -1527,6 +1527,67 @@ def test_tm1_keeps_global_shell_variable(self) -> None:
         findings = tm_mod.analyze(content, "runner.py", "python")
         assert any(finding.rule_id == "TM1" for finding in findings)
 
+    def test_tm1_parses_dense_shell_flags_once(self) -> None:
+        content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 4000
+        content += "#" + " ordinary" * ((256_000 - len(content)) // 9)
+        content = content.ljust(256_000)
+        assert len(content) == 256_000
+        with patch.object(tm_mod, "parse_python_source", wraps=tm_mod.parse_python_source) as parse:
+            findings = tm_mod.analyze(content, "runner.py", "python")
+        assert parse.call_count == 1
+        assert len([finding for finding in findings if finding.rule_id == "TM1"]) == 4000
+
+    def test_tm1_caches_unparseable_window_conservatively(self) -> None:
+        content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 20 + "\nif :"
+        with patch.object(tm_mod, "parse_python_source", wraps=tm_mod.parse_python_source) as parse:
+            findings = tm_mod.analyze(content, "runner.py", "python")
+        assert parse.call_count == 1
+        assert len([finding for finding in findings if finding.rule_id == "TM1"]) == 20
+
+    def test_tm1_remains_lexical_for_large_and_normalized_views(self) -> None:
+        from skillspector.nodes.analyzers import static_runner
+
+        for content in (
+            "# ordinary source\n" * 16_000 + "subprocess.run(cmd, shell=True)",
+            "subprocess.run(cmd, shell=Tru\u200be)",
+        ):
+            result = static_runner.run_static_patterns_with_ledger(
+                {"components": ["runner.py"], "file_cache": {"runner.py": content}}, [tm_mod]
+            )
+            assert any(finding.rule_id == "TM1" for finding in result["findings"])
+
+    def test_tm1_scope_index_honors_runtime_check(self) -> None:
+        content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 300
+        calls = 0
+
+        def check_runtime() -> None:
+            nonlocal calls
+            calls += 1
+            if calls == 20:
+                raise RuntimeError("deadline")
+
+        with pytest.raises(RuntimeError, match="deadline"):
+            tm_mod.analyze(content, "runner.py", "python", check_runtime=check_runtime)
+        assert calls == 20
+
+    def test_tm1_skips_parse_without_variable_shell_flags(self) -> None:
+        with patch.object(tm_mod, "parse_python_source", side_effect=AssertionError("parsed")):
+            findings = tm_mod.analyze("subprocess.run(cmd, shell=True)", "runner.py", "python")
+        assert any(finding.rule_id == "TM1" for finding in findings)
+
+    def test_tm1_preserves_nonlocal_binding_scope(self) -> None:
+        content = (
+            "def outer():\n"
+            "    use_shell = False\n"
+            "    def assign():\n"
+            "        use_shell = True\n"
+            "    def run():\n"
+            "        nonlocal use_shell\n"
+            "        subprocess.run(cmd, shell=use_shell)\n"
+        )
+        findings = tm_mod.analyze(content, "runner.py", "python")
+        assert not any(finding.rule_id == "TM1" for finding in findings)
+
     def test_application_specific_no_verify_flag_is_not_tool_misuse(self) -> None:
         content = """\
 print("verification: skipped (--no-verify)")