fix(analyzer): enforce deadlines for projected prompt matching

src/skillspector/nodes/analyzers/artifact_integrity.py

@@ -98,8 +98,8 @@
 _PROJECTED_PROMPT_PATTERNS = tuple(
     pattern for pattern, _confidence in (*COMPILED_P3_PATTERNS, *COMPILED_P4_PATTERNS)
 )
-# Removing line breaks can give the existing wildcard patterns a much longer
-# search space. Interrupt the regex itself, not just work between matches.
+# Reconstructed projections can give wildcard patterns a much longer search
+# space. Interrupt the regex itself, not just work between matches.
 _MULTILINE_PROMPT_PATTERN_SECONDS = 0.25
 _MULTILINE_PROMPT_PATTERNS = tuple(
     regex.compile(pattern.pattern, regex.ASCII | regex.IGNORECASE | regex.MULTILINE)
@@ -682,9 +682,9 @@ def _projected_prompt_injection_line(
         else (view.text,)
     )
     for projected_text in projected_texts:
-        for pattern in _PROJECTED_PROMPT_PATTERNS:
-            budget.check_runtime()
-            match = pattern.search(projected_text)
+        matching_text = _multiline_prompt_matching_text(projected_text, budget)
+        for pattern in _MULTILINE_PROMPT_PATTERNS:
+            match = next(_timed_prompt_matches(pattern, matching_text, budget), None)
             if match is None:
                 continue
             reconstructed_gaps = view.reconstructed_source_spans(match.start(), match.end())
@@ -705,9 +705,9 @@ def _projected_prompt_injection_line(
         )
         candidate_offsets = tuple(point[0] for point in join_points)
         for projected_text in irregular_texts:
-            for pattern in _PROJECTED_PROMPT_PATTERNS:
-                budget.check_runtime()
-                for match in pattern.finditer(projected_text):
+            matching_text = _multiline_prompt_matching_text(projected_text, budget)
+            for pattern in _MULTILINE_PROMPT_PATTERNS:
+                for match in _timed_prompt_matches(pattern, matching_text, budget):
                     budget.check_runtime()
                     point_index = bisect_right(candidate_offsets, match.start())
                     if (
@@ -759,6 +759,57 @@ def _multiline_prompt_matching_text(text: str, budget: _ArtifactIntegrityBudget)
     return "".join(parts)
 
 
+def _timed_prompt_matches(
+    pattern: regex.Pattern[str],
+    text: str,
+    budget: _ArtifactIntegrityBudget,
+) -> Iterator[regex.Match[str]]:
+    """Apply the same interruptible matching budget to every prompt projection."""
+    matching_seconds = 0.0
+    matching_limit = _MULTILINE_PROMPT_PATTERN_SECONDS
+    terminal_timeout = False
+    start = 0
+    skip_empty = False
+    while True:
+        budget.check_runtime()
+        remaining = transitive_remaining_seconds(budget.state)
+        if remaining is not None:
+            workflow_limit = matching_seconds + max(0.0, remaining)
+            # A per-pattern cap ends this artifact; workflow exhaustion remains terminal.
+            if workflow_limit <= matching_limit:
+                matching_limit = workflow_limit
+                terminal_timeout = True
+        timeout = matching_limit - matching_seconds
+        if timeout <= 0:
+            raise _ArtifactIntegrityResourceLimitError(
+                LedgerReason.RUNTIME_LIMIT,
+                {"observed_seconds": matching_seconds, "limit_seconds": matching_limit},
+                terminal=terminal_timeout,
+            )
+        started_at = time.thread_time()
+        expired = False
+        try:
+            # Restart across yields so consumer work never uses the match budget.
+            matches = pattern.finditer(text, pos=start, timeout=timeout, concurrent=False)
+            if skip_empty:
+                next(matches, None)
+            match = next(matches, None)
+        except TimeoutError:
+            expired = True
+        finally:
+            matching_seconds += max(0.0, time.thread_time() - started_at)
+        budget.check_runtime()
+        if expired:
+            # regex counts process CPU, including native work in other threads.
+            # Retry only the same search; the thread and workflow budgets still bound it.
+            continue
+        if match is None:
+            return
+        start = match.end()
+        skip_empty = match.start() == start
+        yield match
+
+
 def _multiline_prompt_injection_line(
     content: str,
     budget: _ArtifactIntegrityBudget,
@@ -770,49 +821,27 @@ def _multiline_prompt_injection_line(
     matching_text = _multiline_prompt_matching_text(view.text, budget)
     first_offset: int | None = None
     for pattern in _MULTILINE_PROMPT_PATTERNS:
-        budget.check_runtime()
-        remaining = transitive_remaining_seconds(budget.state)
-        timeout = _MULTILINE_PROMPT_PATTERN_SECONDS
-        terminal_timeout = False
-        if remaining is not None:
-            terminal_timeout = remaining <= timeout
-            timeout = min(timeout, max(0.0, remaining))
-        started_at = time.monotonic()
         reconstruction_index = 0
-        try:
-            # Keep this short, interruptible search on the current thread.
-            # Releasing the GIL lets another analyzer consume its wall-clock
-            # allowance and turn ordinary prose into a false timeout.
-            for match in pattern.finditer(matching_text, timeout=timeout, concurrent=False):
+        for match in _timed_prompt_matches(pattern, matching_text, budget):
+            # Matches and reconstruction spans are both ordered. Advance
+            # once per span, including ordinary matches before a spaced
+            # instruction, instead of rescanning all provenance per match.
+            while (
+                reconstruction_index < len(view.reconstructions)
+                and view.reconstructions[reconstruction_index].derived_end <= match.start() + 1
+            ):
                 budget.check_runtime()
-                # Matches and reconstruction spans are both ordered. Advance
-                # once per span, including ordinary matches before a spaced
-                # instruction, instead of rescanning all provenance per match.
-                while (
-                    reconstruction_index < len(view.reconstructions)
-                    and view.reconstructions[reconstruction_index].derived_end <= match.start() + 1
-                ):
-                    budget.check_runtime()
-                    reconstruction_index += 1
-                if reconstruction_index == len(view.reconstructions):
-                    break
-                reconstruction = view.reconstructions[reconstruction_index]
-                right = max(match.start() + 1, reconstruction.derived_start + 1)
-                if right < min(match.end(), reconstruction.derived_end):
-                    source_offset = view.source_offset(right - 1) + 1
-                    if first_offset is None or source_offset < first_offset:
-                        first_offset = source_offset
-                    # Later matches cannot precede this pattern's first gap.
-                    break
-        except TimeoutError as exc:
-            raise _ArtifactIntegrityResourceLimitError(
-                LedgerReason.RUNTIME_LIMIT,
-                {
-                    "observed_seconds": max(0.0, time.monotonic() - started_at),
-                    "limit_seconds": timeout,
-                },
-                terminal=terminal_timeout,
-            ) from exc
+                reconstruction_index += 1
+            if reconstruction_index == len(view.reconstructions):
+                break
+            reconstruction = view.reconstructions[reconstruction_index]
+            right = max(match.start() + 1, reconstruction.derived_start + 1)
+            if right < min(match.end(), reconstruction.derived_end):
+                source_offset = view.source_offset(right - 1) + 1
+                if first_offset is None or source_offset < first_offset:
+                    first_offset = source_offset
+                # Later matches cannot precede this pattern's first gap.
+                break
     return get_line_number(content, first_offset) if first_offset is not None else None
 
 

tests/nodes/analyzers/test_multiline_prompt_spacing.py

@@ -196,9 +196,11 @@ def test_multiline_regex_timeout_is_scoped_to_current_artifact(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     timeouts = []
+    clock = iter((0.0, artifact_integrity._MULTILINE_PROMPT_PATTERN_SECONDS + 0.02))
+    monkeypatch.setattr(artifact_integrity.time, "thread_time", lambda: next(clock))
 
     class ExpiredPattern:
-        def finditer(self, _text, *, timeout, concurrent):
+        def finditer(self, _text, *, pos, timeout, concurrent):
             timeouts.append(timeout)
             raise TimeoutError("regex timed out")
 
@@ -232,8 +234,11 @@ def finditer(self, _text, *, timeout, concurrent):
 def test_multiline_regex_timeout_preserves_global_terminal_limit(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    clock = iter((0.0, 0.02))
+    monkeypatch.setattr(artifact_integrity.time, "thread_time", lambda: next(clock))
+
     class ExpiredPattern:
-        def finditer(self, _text, *, timeout, concurrent):
+        def finditer(self, _text, *, pos, timeout, concurrent):
             raise TimeoutError("global deadline reached")
 
     monkeypatch.setattr(artifact_integrity, "_MULTILINE_PROMPT_PATTERNS", (ExpiredPattern(),))
@@ -413,7 +418,9 @@ def compete_for_gil() -> None:
     class ContendedPattern:
         def finditer(self, text, **kwargs):
             matches = original.finditer(text, **kwargs)
-            first = next(matches)
+            first = next(matches, None)
+            if first is None:
+                return
             start_work.set()
             yield first
             yield from matches
@@ -460,3 +467,102 @@ async def test_parallel_scan_of_contractions_remains_complete(tmp_path: Path) ->
     assert result["analysis_completeness"]["is_complete"] is True
     mcp = await run_scan(str(tmp_path), use_llm=False)
     assert mcp["safe_to_install"] is True
+
+
+@pytest.mark.parametrize("irregular", [False, True])
+def test_projected_prompt_regex_timeout_records_incomplete_coverage(
+    monkeypatch: pytest.MonkeyPatch, irregular: bool
+) -> None:
+    calls = []
+    clock = iter((0.0, 0.02, 0.02, 0.04))
+    monkeypatch.setattr(artifact_integrity.time, "thread_time", lambda: next(clock))
+    monkeypatch.setattr(
+        artifact_integrity,
+        "_multiline_prompt_injection_line",
+        lambda *_args: pytest.fail("ordinary projection reached the multiline fallback"),
+    )
+
+    class ExpiredPattern:
+        def finditer(self, text, *, pos, timeout, concurrent):
+            calls.append((timeout, concurrent))
+            # Exercise the irregular projection after the normal search.
+            if irregular and len(calls) == 1:
+                return iter(())
+            raise TimeoutError("regex timed out")
+
+    monkeypatch.setattr(artifact_integrity, "_MULTILINE_PROMPT_PATTERNS", (ExpiredPattern(),))
+    monkeypatch.setattr(artifact_integrity, "transitive_remaining_seconds", lambda _state: 0.01)
+    if irregular:
+        monkeypatch.setattr(
+            artifact_integrity,
+            "_irregular_spacing_prompt_projection",
+            lambda _view: ("without telling the user", ((1, 0),)),
+        )
+    content = "w i t h o u t   t e l l i n g   u s e r"
+    result = node({"components": ["SKILL.md"], "file_cache": {"SKILL.md": content}})
+    assert len(calls) == (2 if irregular else 1)
+    assert all(call == (0.01, False) for call in calls)
+    assert all(event["outcome"] == "partial" for event in result["inspection_ledger"])
+    assert all(
+        event["reason_code"] == LedgerReason.RUNTIME_LIMIT for event in result["inspection_ledger"]
+    )
+
+
+def test_repeated_projected_prompt_prefix_has_engine_deadline(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(artifact_integrity, "_MULTILINE_PROMPT_PATTERN_SECONDS", 0.000001)
+    monkeypatch.setattr(
+        artifact_integrity,
+        "_multiline_prompt_injection_line",
+        lambda *_args: pytest.fail("ordinary projection reached the multiline fallback"),
+    )
+    content = "w i t h o u t   t e l l i n g   u s e r " * 1000
+    with pytest.raises(artifact_integrity._ArtifactIntegrityResourceLimitError) as caught:
+        artifact_integrity._projected_prompt_injection_line(
+            content, artifact_integrity._ArtifactIntegrityBudget({})
+        )
+    assert caught.value.reason == LedgerReason.RUNTIME_LIMIT
+
+
+def test_timed_prompt_matches_excludes_consumer_cpu(monkeypatch: pytest.MonkeyPatch) -> None:
+    import time
+
+    import regex
+
+    monkeypatch.setattr(artifact_integrity, "_MULTILINE_PROMPT_PATTERN_SECONDS", 0.01)
+    matches = artifact_integrity._timed_prompt_matches(
+        regex.compile(r".*?"), "ab", artifact_integrity._ArtifactIntegrityBudget({})
+    )
+    first = next(matches)
+    until = time.thread_time() + 0.03
+    while time.thread_time() < until:
+        pass
+    assert [first.span(), *(match.span() for match in matches)] == [
+        (0, 0),
+        (0, 1),
+        (1, 1),
+        (1, 2),
+        (2, 2),
+    ]
+
+
+def test_timed_prompt_matches_retries_other_thread_cpu(monkeypatch: pytest.MonkeyPatch) -> None:
+    import regex
+
+    calls = 0
+    original = regex.compile("x")
+
+    class ContendedPattern:
+        def finditer(self, text, **kwargs):
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                raise TimeoutError("another thread used process CPU")
+            return original.finditer(text, **kwargs)
+
+    matches = artifact_integrity._timed_prompt_matches(
+        ContendedPattern(), "x", artifact_integrity._ArtifactIntegrityBudget({})
+    )
+    assert [match.span() for match in matches] == [(0, 1)]
+    assert calls == 3