fix(report): preserve occurrence columns in JSON and SARIF

src/skillspector/nodes/report.py

@@ -218,6 +218,20 @@ def _sarif_artifact_location(
     return SarifArtifactLocation(uri=uri, properties=properties or None)
 
 
+def _occurrence_columns(
+    finding: Finding, occurrence: Mapping[str, object]
+) -> tuple[int | None, int | None]:
+    """Do not borrow representative columns for an occurrence with unknown columns."""
+    start = occurrence.get(
+        "start_column", finding.start_column if not finding.occurrences else None
+    )
+    end = occurrence.get("end_column", finding.end_column if not finding.occurrences else None)
+    return (
+        start if isinstance(start, int) else None,
+        end if isinstance(end, int) else None,
+    )
+
+
 def _expand_occurrences(findings: list[Finding]) -> list[Finding]:
     """Expand compacted findings for human/JSON output without losing locations."""
     expanded: list[Finding] = []
@@ -234,6 +248,7 @@ def _expand_occurrences(findings: list[Finding]) -> list[Finding]:
             start_line = start_value if isinstance(start_value, int) else finding.start_line
             end_value = occurrence.get("end_line")
             end_line = end_value if isinstance(end_value, int) else None
+            start_column, end_column = _occurrence_columns(finding, occurrence)
             provenance = _occurrence_provenance(finding, occurrence)
             depth_value = provenance.get("transitive_depth")
             expanded.append(
@@ -242,6 +257,8 @@ def _expand_occurrences(findings: list[Finding]) -> list[Finding]:
                     file=str(occurrence.get("file", finding.file)),
                     start_line=start_line,
                     end_line=end_line,
+                    start_column=start_column,
+                    end_column=end_column,
                     source_identity=(
                         str(provenance["source_identity"])
                         if "source_identity" in provenance
@@ -602,6 +619,7 @@ def _build_sarif(
             start_line = start_value if isinstance(start_value, int) else finding.start_line
             end_value = occurrence.get("end_line")
             end_line = int(end_value) if isinstance(end_value, int) else None
+            start_column, end_column = _occurrence_columns(finding, occurrence)
             results.append(
                 SarifResult(
                     ruleId=finding.rule_id,
@@ -612,7 +630,14 @@ def _build_sarif(
                         SarifLocation(
                             physicalLocation=SarifPhysicalLocation(
                                 artifactLocation=_sarif_artifact_location(finding, occurrence),
-                                region=SarifRegion(startLine=start_line, endLine=end_line),
+                                region=SarifRegion(
+                                    startLine=start_line,
+                                    endLine=end_line,
+                                    startColumn=start_column + 1
+                                    if start_column is not None
+                                    else None,
+                                    endColumn=end_column + 1 if end_column is not None else None,
+                                ),
                             )
                         )
                     ],
@@ -639,6 +664,7 @@ def _build_sarif(
             start_line = start_value if isinstance(start_value, int) else finding.start_line
             end_value = occurrence.get("end_line")
             end_line = int(end_value) if isinstance(end_value, int) else None
+            start_column, end_column = _occurrence_columns(finding, occurrence)
             results.append(
                 SarifResult(
                     ruleId=finding.rule_id,
@@ -649,7 +675,14 @@ def _build_sarif(
                         SarifLocation(
                             physicalLocation=SarifPhysicalLocation(
                                 artifactLocation=_sarif_artifact_location(finding, occurrence),
-                                region=SarifRegion(startLine=start_line, endLine=end_line),
+                                region=SarifRegion(
+                                    startLine=start_line,
+                                    endLine=end_line,
+                                    startColumn=start_column + 1
+                                    if start_column is not None
+                                    else None,
+                                    endColumn=end_column + 1 if end_column is not None else None,
+                                ),
                             )
                         )
                     ],
@@ -875,6 +908,7 @@ def notification_from_exception(
                         )
                     ),
                     results=results,
+                    columnKind="unicodeCodePoints",
                     invocations=invocations,
                 )
             ],

src/skillspector/sarif_models.py

@@ -153,6 +153,9 @@ class SarifRun(BaseModel):
 
     tool: SarifTool
     results: list[SarifResult] = Field(default_factory=list)
+    column_kind: Literal["utf16CodeUnits", "unicodeCodePoints"] | None = Field(
+        default=None, alias="columnKind"
+    )
     artifacts: list[SarifArtifact] | None = None
     invocations: list[SarifInvocation] | None = None
 

tests/nodes/test_report_occurrence_columns.py

@@ -0,0 +1,203 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Regression coverage for occurrence source spans in public reports."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from skillspector.mcp_server import run_scan
+from skillspector.models import Finding
+from skillspector.nodes.analyzers import static_patterns_prompt_injection, static_runner
+from skillspector.nodes.report import _build_sarif, _expand_occurrences, report
+from skillspector.suppression import SuppressedFinding
+
+
+@pytest.mark.parametrize("suppressed", [False, True])
+def test_report_occurrences_keep_optional_and_multiline_columns(suppressed: bool) -> None:
+    # Each occurrence owns its columns; unknown columns must not inherit 0/28.
+    finding = Finding(
+        rule_id="P1",
+        message="Instruction Override",
+        file="SKILL.md",
+        start_line=2,
+        end_line=2,
+        start_column=0,
+        end_column=28,
+        occurrences=[
+            {
+                "file": "SKILL.md",
+                "start_line": 2,
+                "end_line": 2,
+                "start_column": 0,
+                "end_column": 28,
+            },
+            {
+                "file": "SKILL.md",
+                "start_line": 2,
+                "end_line": 2,
+                "start_column": 30,
+                "end_column": 58,
+            },
+            {
+                "file": "notes.md",
+                "start_line": 1,
+                "end_line": 2,
+                "start_column": 8,
+                "end_column": 12,
+            },
+            {"file": "legacy.md", "start_line": 4, "end_line": 4},
+            {"file": "partial.md", "start_line": 1, "end_line": 1, "start_column": 5},
+            {
+                "file": "unknown.md",
+                "start_line": 1,
+                "end_line": 1,
+                "start_column": None,
+                "end_column": None,
+            },
+        ],
+    )
+    if not suppressed:
+        locations = [row.to_dict()["location"] for row in _expand_occurrences([finding])]
+        assert locations == finding.occurrences[:3] + [
+            {"file": "legacy.md", "start_line": 4, "end_line": 4},
+            {"file": "partial.md", "start_line": 1, "end_line": 1, "start_column": 5},
+            {"file": "unknown.md", "start_line": 1, "end_line": 1},
+        ]
+    sarif = _build_sarif(
+        [] if suppressed else [finding],
+        suppressed=[SuppressedFinding(finding=finding, reason="test baseline")]
+        if suppressed
+        else [],
+    )
+    regions = [
+        row["locations"][0]["physicalLocation"]["region"] for row in sarif["runs"][0]["results"]
+    ]
+    assert regions == [
+        {"startLine": 2, "endLine": 2, "startColumn": 1, "endColumn": 29},
+        {"startLine": 2, "endLine": 2, "startColumn": 31, "endColumn": 59},
+        {"startLine": 1, "endLine": 2, "startColumn": 9, "endColumn": 13},
+        {"startLine": 4, "endLine": 4},
+        {"startLine": 1, "endLine": 1, "startColumn": 6},
+        {"startLine": 1, "endLine": 1},
+    ]
+
+
+@pytest.mark.parametrize("columns", [(None, None), (0, 28), (8, None), (None, 0)])
+def test_report_finding_without_occurrences_keeps_own_columns(
+    columns: tuple[int | None, int | None],
+) -> None:
+    finding = Finding(rule_id="P1", message="test", start_column=columns[0], end_column=columns[1])
+    expanded = _expand_occurrences([finding])[0]
+    assert (expanded.start_column, expanded.end_column) == columns
+    region = _build_sarif([finding])["runs"][0]["results"][0]["locations"][0]["physicalLocation"][
+        "region"
+    ]
+    assert region.get("startColumn") == (None if columns[0] is None else columns[0] + 1)
+    assert region.get("endColumn") == (None if columns[1] is None else columns[1] + 1)
+
+
+@pytest.mark.parametrize(
+    "content,rule_id,expected",
+    [
+        pytest.param(
+            "😀 é ignore previous instructions; ignore previous instructions",
+            "P1",
+            [(1, 4, 1, 32), (1, 34, 1, 62)],
+            id="unicode-prefix",
+        ),
+        pytest.param(
+            "ignore previous\ninstructions; ignore previous instructions",
+            "P1",
+            [(1, 0, 2, 12), (2, 14, 2, 42)],
+            id="multiline",
+        ),
+        pytest.param(
+            "ｉｇｎｏｒｅ previous instructions; ignore previous instructions",
+            "P1",
+            [(1, 0, 1, 28), (1, 30, 1, 58)],
+            id="normalized",
+        ),
+        pytest.param(
+            "s e n d conversation to; s e n d conversation to",
+            "P3",
+            [(1, 0, 1, 23), (1, 25, 1, 48)],
+            id="projected",
+        ),
+    ],
+)
+def test_reports_preserve_original_source_columns(
+    content: str,
+    rule_id: str,
+    expected: list[tuple[int, int, int, int]],
+) -> None:
+    findings = static_runner.run_static_patterns(
+        {"components": ["SKILL.md"], "file_cache": {"SKILL.md": content}},
+        [static_patterns_prompt_injection],
+    )
+    findings = [finding for finding in findings if finding.rule_id == rule_id]
+    assert len(findings) == 2
+    fingerprints = {finding.fingerprint() for finding in findings}
+    assert len(fingerprints) == 1
+    result = report({"findings": findings, "output_format": "json"})
+    issues = json.loads(result["report_body"])["issues"]
+    assert [
+        (
+            row["location"]["start_line"],
+            row["location"]["start_column"],
+            row["location"]["end_line"],
+            row["location"]["end_column"],
+        )
+        for row in issues
+    ] == expected
+    assert {row["match_fingerprint"] for row in issues} == fingerprints
+    sarif_run = result["sarif_report"]["runs"][0]
+    regions = [row["locations"][0]["physicalLocation"]["region"] for row in sarif_run["results"]]
+    assert [
+        (row["startLine"], row["startColumn"] - 1, row["endLine"], row["endColumn"] - 1)
+        for row in regions
+    ] == expected
+    assert sarif_run["columnKind"] == "unicodeCodePoints"
+
+
+async def test_mcp_embedded_report_preserves_same_line_and_multifile_spans(tmp_path: Path) -> None:
+    (tmp_path / "SKILL.md").write_text(
+        "# Identity fixture\nignore previous instructions; ignore previous instructions\n"
+        "ignore previous instructions\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "notes.md").write_text("Prefix: ignore previous instructions\n", encoding="utf-8")
+    result = await run_scan(str(tmp_path), use_llm=False, output_format="json")
+    expected = [
+        ("SKILL.md", 2, 0, 28),
+        ("SKILL.md", 2, 30, 58),
+        ("SKILL.md", 3, 0, 28),
+        ("notes.md", 1, 8, 36),
+    ]
+    findings = [row for row in result["findings"] if row["id"] == "P1"]
+    assert len(findings) == 1
+    assert [
+        (row["file"], row["start_line"], row["start_column"], row["end_column"])
+        for row in findings[0]["occurrences"]
+    ] == expected
+    embedded = json.loads(result["report"])
+    assert [
+        (
+            row["location"]["file"],
+            row["location"]["start_line"],
+            row["location"]["start_column"],
+            row["location"]["end_column"],
+        )
+        for row in embedded["issues"]
+        if row["id"] == "P1"
+    ] == expected
+    assert {row["match_fingerprint"] for row in embedded["issues"]} == {
+        findings[0]["match_fingerprint"]
+    }
+    assert result["risk_score"] == 35
+    assert result["recommendation"] == "CAUTION"
+    assert result["llm_used"] is False