fix(llm): analyze oversized source lines without losing coverage

src/skillspector/nodes/meta_analyzer.py

@@ -48,6 +48,8 @@
     BatchFailure,
     LLMAnalyzerBase,
     LLMRuntimeLimitError,
+    _consolidate_fragment_ledger_events,
+    _project_line_fragments,
     append_output_language_instruction,
     estimate_tokens,
 )
@@ -452,13 +454,21 @@ def get_batches(
         aligned with the original file.
         """
         sc10_files = {finding.file for finding in findings or [] if finding.rule_id == "SC10"}
+        # Redacted columns can change length and must not claim original-source
+        # provenance. Keep these physical lines whole for redaction and the guard.
+        self._redacted_source_files = sc10_files
         if sc10_files:
             file_cache = {
                 path: redact_text(text) if path in sc10_files else text
                 for path, text in file_cache.items()
             }
         return super().get_batches(file_paths, file_cache, findings, **prompt_kwargs)
 
+    def _fragment_source_line(self, batch: Batch, **prompt_kwargs: object) -> list[Batch]:
+        if batch.file_path in getattr(self, "_redacted_source_files", set()):
+            return [batch]
+        return super()._fragment_source_line(batch, **prompt_kwargs)
+
     def parse_response(  # type: ignore[override]  # Base class permits custom parsed values.
         self,
         response: MetaAnalyzerResult,
@@ -619,12 +629,62 @@ def apply_filter(
 
 def _meta_batch_work_id(batch: Batch) -> str:
     """Return the ledger identity for one submitted meta-analysis batch."""
-    return inspection_work_id(
+    identity = inspection_work_id(
         "meta_analyzer",
         batch.file_path,
         batch.start_line if batch.end_line is not None else None,
         batch.end_line,
     )
+    if batch.source_start_column is not None:
+        # Internal return reconciliation needs a distinct key for each span.
+        # Public ledger identity remains the original source line range.
+        identity += (
+            f":columns:{batch.source_start_column}:{batch.source_end_column}"
+            f":length:{batch.source_line_length}"
+        )
+    return identity
+
+
+def _account_for_missing_meta_batches(
+    batches: list[Batch], outcome: BatchExecutionResult
+) -> BatchExecutionResult:
+    submitted = {_meta_batch_work_id(batch): batch for batch in batches}
+    successful: dict[str, tuple[Batch, list]] = {}
+    for returned, items in outcome.successful:
+        identity = _meta_batch_work_id(returned)
+        if identity in submitted and identity not in successful:
+            successful[identity] = (submitted[identity], items)
+    failures = [
+        BatchFailure(submitted[identity], failure.error_class, failure.reason)
+        for failure in outcome.failures
+        if (identity := _meta_batch_work_id(failure.batch)) in submitted
+    ]
+    warnings = [
+        BatchFailure(submitted[identity], warning.error_class, warning.reason)
+        for warning in outcome.warnings
+        if (identity := _meta_batch_work_id(warning.batch)) in submitted
+    ]
+    recovered = [
+        (submitted[identity], values)
+        for returned, values in outcome.recovered
+        if (identity := _meta_batch_work_id(returned)) in submitted
+    ]
+    observed = set(successful) | {
+        _meta_batch_work_id(failure.batch) for failure in [*failures, *warnings]
+    }
+    return BatchExecutionResult(
+        successful=list(successful.values()),
+        failures=[
+            *failures,
+            *(
+                BatchFailure(batch, "MissingBatchResult")
+                for identity, batch in submitted.items()
+                if identity not in observed
+            ),
+        ],
+        recovered=recovered,
+        warnings=warnings,
+    )
 
 
 def _meta_ledger_response(
@@ -633,6 +693,27 @@ def _meta_ledger_response(
     filtered: list[Finding],
 ) -> tuple[list[InspectionLedgerEvent], AnalyzerStatusEvent]:
     """Account for each meta batch while preserving fail-closed finding identity."""
+    outcome = _account_for_missing_meta_batches(batches, outcome)
+    projected_successes, projected_failures, _ = _project_line_fragments(
+        [(batch, batch.findings) for batch, _ in outcome.successful],
+        [*outcome.failures, *outcome.warnings],
+    )
+    outcome = BatchExecutionResult(
+        successful=[
+            (replace(batch, findings=findings), []) for batch, findings in projected_successes
+        ],
+        failures=[
+            failure
+            for failure in projected_failures
+            if failure.reason is not LedgerReason.LLM_FORMAT_RECOVERY_EXHAUSTED
+        ],
+        recovered=outcome.recovered,
+        warnings=[
+            failure
+            for failure in projected_failures
+            if failure.reason is LedgerReason.LLM_FORMAT_RECOVERY_EXHAUSTED
+        ],
+    )
     retained_ids = {finding.finding_id for finding in filtered}
     completed_ids = {
         finding.finding_id for batch, _ in outcome.successful for finding in batch.findings
@@ -687,7 +768,7 @@ def _meta_ledger_response(
         )
     if not events:
         return events, analyzer_status_event(analyzer_id="meta_analyzer", status="completed")
-    return events, analyzer_status_for_events("meta_analyzer", events)
+    return _consolidate_fragment_ledger_events("meta_analyzer", events, batches)
 
 
 def _effective_finding_ids(findings: list[Finding]) -> list[str]:
@@ -919,14 +1000,22 @@ def meta_analyzer(state: SkillspectorState) -> MetaAnalyzerResponse:
                 ],
             )
 
+        detailed = _account_for_missing_meta_batches(batches, detailed)
+        completed_sources, _, _ = _project_line_fragments(
+            [(batch, batch.findings) for batch, _ in detailed.successful],
+            [*detailed.failures, *detailed.warnings],
+        )
+        analysed_ids = {finding.finding_id for _, values in completed_sources for finding in values}
         completed_work_ids = {_meta_batch_work_id(batch) for batch, _ in detailed.successful}
-        analysed_ids: set[str] = set()
         for batch, assessments in batch_results:
-            if _meta_batch_work_id(batch) in completed_work_ids:
-                analysed_ids.update(finding.finding_id for finding in batch.findings)
+            if (
+                batch.source_start_column is not None
+                or _meta_batch_work_id(batch) in completed_work_ids
+            ):
+                # A fragment cannot establish its original line's full context.
+                # Only the complete successful coverage above permits filtering.
                 continue
-            # Partial recovery proves only the returned item assessments, even
-            # when all submitted batches have a partial result.
+            # Preserve main's per-item partial recovery for ordinary batches.
             originals = {finding.finding_id: finding for finding in batch.findings}
             for assessment in assessments:
                 finding_id = assessment.get("finding_id")

tests/nodes/analyzers/test_dependency_sources.py

@@ -14,7 +14,6 @@
 from skillspector.llm_analyzer_base import Batch
 from skillspector.models import Finding
 from skillspector.nodes.meta_analyzer import (
-    PER_FILE_ANALYSIS_PROMPT,
     LLMMetaAnalyzer,
     _fallback_filtered,
     _passthrough_with_defaults,
@@ -1234,14 +1233,13 @@ def test_cargo_url_credentials_are_redacted_for_supported_schemes(destination: s
     assert "packages.example.invalid" in serialized
 
 
-def test_sc10_credentials_are_redacted_before_provider_prompt_construction() -> None:
+def test_sc10_credentials_are_redacted_before_provider_prompt_construction(mock_llm) -> None:
     username = "provider-user-sentinel"
     password = "provider-password-sentinel"
     token = "provider-token-sentinel"
     content = f"registry=ssh://{username}:{password}@packages.example.invalid/index?token={token}\n"
     finding = _analyze({".npmrc": content})[0]
-    analyzer = LLMMetaAnalyzer.__new__(LLMMetaAnalyzer)
-    analyzer.base_prompt = PER_FILE_ANALYSIS_PROMPT
+    analyzer = LLMMetaAnalyzer(model="fixture")
     analyzer._input_budget = 100_000
 
     batch = analyzer.get_batches([".npmrc"], {".npmrc": content}, [finding])[0]
@@ -1251,6 +1249,8 @@ def test_sc10_credentials_are_redacted_before_provider_prompt_construction() ->
         assert secret not in batch.content
         assert secret not in prompt
     assert "packages.example.invalid" in prompt
+    mock_llm.invoke.assert_not_called()
+    mock_llm.ainvoke.assert_not_called()
 
 
 def test_hidden_source_finding_is_marked_local_only() -> None:

tests/nodes/analyzers/test_semantic_security_discovery.py

@@ -314,7 +314,8 @@ def test_oversized_file_multiple_batches(self, base_state) -> None:
 
         with patch(
             "skillspector.llm_analyzer_base.get_max_input_tokens",
-            return_value=50,
+            # Leave room for the fixed prompt/schema; only source needs splitting.
+            return_value=8192,
         ):
             with patch.object(
                 LLMAnalyzerBase, "arun_batches", new_callable=AsyncMock, return_value=[]