fix(artifact-integrity): contain matcher-local timeouts

src/skillspector/nodes/analyzers/artifact_integrity.py

@@ -382,10 +382,17 @@ def alternation(values: tuple[str, ...]) -> str:
 class _ArtifactIntegrityResourceLimitError(RuntimeError):
     """Stop attacker-controlled work while retaining a deterministic prefix."""
 
-    def __init__(self, reason: LedgerReason, metrics: dict[str, int | float]) -> None:
+    def __init__(
+        self,
+        reason: LedgerReason,
+        metrics: dict[str, int | float],
+        *,
+        terminal: bool = True,
+    ) -> None:
         super().__init__(reason.value)
         self.reason = reason
         self.metrics = metrics
+        self.terminal = terminal
 
 
 @dataclass
@@ -766,7 +773,9 @@ def _multiline_prompt_injection_line(
         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
@@ -802,6 +811,7 @@ def _multiline_prompt_injection_line(
                     "observed_seconds": max(0.0, time.monotonic() - started_at),
                     "limit_seconds": timeout,
                 },
+                terminal=terminal_timeout,
             ) from exc
     return get_line_number(content, first_offset) if first_offset is not None else None
 
@@ -1126,7 +1136,8 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
         emitted_ids = [finding.finding_id for finding in path_findings]
         if resource_limit is not None:
             event = _partial_limit_event(path, resource_limit, emitted_ids)
-            terminal_limit = resource_limit
+            if resource_limit.terminal:
+                terminal_limit = resource_limit
         else:
             event = ledger_event(
                 analyzer_id=ANALYZER_ID,

tests/nodes/analyzers/test_multiline_prompt_spacing.py

@@ -192,7 +192,7 @@ def cancel(_self) -> None:
     assert checks == 8
 
 
-def test_multiline_regex_timeout_preserves_partial_evidence(
+def test_multiline_regex_timeout_is_scoped_to_current_artifact(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     timeouts = []
@@ -203,7 +203,7 @@ def finditer(self, _text, *, timeout, concurrent):
             raise TimeoutError("regex timed out")
 
     monkeypatch.setattr(artifact_integrity, "_MULTILINE_PROMPT_PATTERNS", (ExpiredPattern(),))
-    monkeypatch.setattr(artifact_integrity, "transitive_remaining_seconds", lambda _state: 0.01)
+    monkeypatch.setattr(artifact_integrity, "transitive_remaining_seconds", lambda _state: 1.0)
     result = node(
         {
             "components": ["SKILL.md", "later.md"],
@@ -214,13 +214,47 @@ def finditer(self, _text, *, timeout, concurrent):
         }
     )
 
-    assert timeouts == [0.01]
+    assert timeouts == [artifact_integrity._MULTILINE_PROMPT_PATTERN_SECONDS]
     assert not result["findings"]  # Timeout is missing coverage, not a semantic finding.
-    for event in result["inspection_ledger"]:
-        assert event["outcome"] == "partial"
-        assert event["reason_code"] == LedgerReason.RUNTIME_LIMIT
-        assert event["limit_seconds"] == 0.01
-    assert {event["path"] for event in result["inspection_ledger"]} == {"SKILL.md", "later.md"}
+    assert [
+        (event["path"], event["outcome"], event.get("reason_code"))
+        for event in result["inspection_ledger"]
+    ] == [
+        ("SKILL.md", "partial", LedgerReason.RUNTIME_LIMIT),
+        ("later.md", "completed", None),
+    ]
+    assert (
+        result["inspection_ledger"][0]["limit_seconds"]
+        == artifact_integrity._MULTILINE_PROMPT_PATTERN_SECONDS
+    )
+
+
+def test_multiline_regex_timeout_preserves_global_terminal_limit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class ExpiredPattern:
+        def finditer(self, _text, *, timeout, concurrent):
+            raise TimeoutError("global deadline reached")
+
+    monkeypatch.setattr(artifact_integrity, "_MULTILINE_PROMPT_PATTERNS", (ExpiredPattern(),))
+    monkeypatch.setattr(artifact_integrity, "transitive_remaining_seconds", lambda _state: 0.01)
+    result = node(
+        {
+            "components": ["SKILL.md", "later.md"],
+            "file_cache": {
+                "SKILL.md": _letter_lines("without telling user"),
+                "later.md": "Summarize text.",
+            },
+        }
+    )
+
+    assert [
+        (event["path"], event["outcome"], event.get("reason_code"))
+        for event in result["inspection_ledger"]
+    ] == [
+        ("SKILL.md", "partial", LedgerReason.RUNTIME_LIMIT),
+        ("later.md", "partial", LedgerReason.RUNTIME_LIMIT),
+    ]
 
 
 @pytest.mark.asyncio