feat: classify structured artifact changes

docs/reviewer-artifact-diff.md

@@ -27,16 +27,26 @@ lone CR to LF before comparison. A binary path that exists in both trees is
 checked for presence only; renderer-dependent bytes are not treated as a
 reproducibility contract.
 
+For changed, missing, or extra JSON and JSONL, the comparator builds a bounded
+summary of the container, record count, top-level keys, safe schema/version
+markers, and validated run-manifest digest fields. Those summaries distinguish
+plain content changes from structure, schema-version, and provenance-digest
+changes. Identical JSON and JSONL are not parsed after their normalized bytes
+match. Artifact bodies are never printed.
+
 Exit status is `0` when comparable artifacts are unchanged, `1` when the tool
 finds contract differences, and `2` when an input cannot be compared safely.
 The output contains no artifact bodies, timestamps, or absolute checkout paths.
 This tool explains a mismatch; it does not replace
 `python scripts/regenerate_artifacts.py --check`, accept regenerated output, or
 assign a release compatibility label.
 
-Each root is limited to 10,000 files. Symlink or reparse-point roots, linked
-entries, special files, unsafe relative paths, unreadable files, and invalid
-UTF-8 text fail closed with exit `2`.
+Each root is limited to 10,000 files. A structured summary accepts at most 64
+MiB of normalized content, 4,096 top-level keys, 4,096 schema markers, and
+10,000 entries per run-manifest digest map. Symlink or reparse-point roots,
+linked entries, special files, unsafe relative paths or metadata, malformed
+digest fields, unreadable files, invalid JSON/JSONL, exceeded limits, and
+invalid UTF-8 text fail closed with exit `2`.
 
 ## Required Release Diff Sections
 

src/telemetry_lab/artifact_contract_diff.py

@@ -1,36 +1,64 @@
 from __future__ import annotations
 
 import codecs
+import io
+import json
 import os
 import re
 import stat
 from collections import Counter
-from collections.abc import Iterable, Iterator
+from collections.abc import Iterable, Iterator, Mapping
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from hashlib import sha256
 from pathlib import Path
-from typing import BinaryIO, Final, Literal, TypeAlias
+from typing import Any, BinaryIO, Final, Literal, TypeAlias
 
 
 TEXT_ARTIFACT_SUFFIXES: Final = frozenset(
     {".csv", ".json", ".jsonl", ".md", ".txt"}
 )
 MAX_FILES: Final = 10_000
+MAX_STRUCTURED_BYTES: Final = 64 * 1024 * 1024
+MAX_STRUCTURE_ITEMS: Final = 4_096
+MAX_DIGEST_ENTRIES: Final = 10_000
 CHUNK_SIZE: Final = 64 * 1024
+RUN_MANIFEST_DIGEST_FIELDS: Final = (
+    "input_digest",
+    "config_digest",
+    "input_file_digests",
+    "config_file_digests",
+)
 
 ArtifactKind: TypeAlias = Literal["json", "jsonl", "text", "binary"]
 DifferenceStatus: TypeAlias = Literal["missing", "extra", "changed"]
 ChangeReason: TypeAlias = Literal[
     "missing-from-actual",
     "extra-in-actual",
     "content-changed",
+    "structure-changed",
+    "schema-version-changed",
+    "run-manifest-digest-changed",
 ]
 
 _DIGEST = re.compile(r"sha256:[0-9a-f]{64}")
 _RELATIVE_PATH = re.compile(
     r"^(?![A-Za-z]:)(?!.*\\)(?!.*(?:^|/)\.\.?(?:/|$))[^/]+(?:/[^/]+)*$"
 )
+_LOCAL_PATH = re.compile(r"^(?:[A-Za-z]:[\\/]|\\\\|/)")
+
+_ALLOWED_CHANGE_REASONS = frozenset(
+    {
+        "missing-from-actual",
+        "extra-in-actual",
+        "content-changed",
+        "structure-changed",
+        "schema-version-changed",
+        "run-manifest-digest-changed",
+    }
+)
+
+
 class ArtifactContractDiffError(ValueError):
     """Raised when an artifact comparison cannot produce a safe result."""
 
@@ -39,6 +67,7 @@ class ArtifactContractDiffError(ValueError):
 class ArtifactSnapshot:
     comparison_digest: str
     comparison_size_bytes: int
+    structure: Mapping[str, Any] | None = None
 
     def __post_init__(self) -> None:
         if _DIGEST.fullmatch(self.comparison_digest) is None:
@@ -63,6 +92,8 @@ def __post_init__(self) -> None:
             raise ArtifactContractDiffError("difference artifact kind is invalid")
         if len(set(self.change_reasons)) != len(self.change_reasons):
             raise ArtifactContractDiffError("difference reasons must be unique")
+        if not set(self.change_reasons) <= _ALLOWED_CHANGE_REASONS:
+            raise ArtifactContractDiffError("difference reason is invalid")
         if self.status == "missing":
             valid = (
                 self.change_reasons == ("missing-from-actual",)
@@ -77,7 +108,10 @@ def __post_init__(self) -> None:
             )
         elif self.status == "changed":
             valid = (
-                self.change_reasons == ("content-changed",)
+                bool(self.change_reasons)
+                and self.change_reasons[0] == "content-changed"
+                and "missing-from-actual" not in self.change_reasons
+                and "extra-in-actual" not in self.change_reasons
                 and self.expected is not None
                 and self.actual is not None
                 and self.artifact_kind != "binary"
@@ -198,13 +232,15 @@ def compare_artifact_trees(
             ):
                 unchanged += 1
                 continue
+            before = _with_structure(before, expected_path, relative_path, kind)
+            after = _with_structure(after, actual_path, relative_path, kind)
             changed += 1
             differences.append(
                 ArtifactDifference(
                     relative_path,
                     "changed",
                     kind,
-                    ("content-changed",),
+                    _change_reasons(before, after),
                     expected=before,
                     actual=after,
                 )
@@ -300,7 +336,7 @@ def _artifact_kind(relative_path: str) -> ArtifactKind:
 def _snapshot(path: Path | None, relative_path: str, kind: ArtifactKind) -> ArtifactSnapshot:
     if path is None:
         raise ArtifactContractDiffError(f"artifact is unavailable: {relative_path}")
-    return _identity(path, relative_path, kind)
+    return _with_structure(_identity(path, relative_path, kind), path, relative_path, kind)
 
 
 def _identity(path: Path, relative_path: str, kind: ArtifactKind) -> ArtifactSnapshot:
@@ -359,3 +395,195 @@ def _normalized_chunks(chunks: Iterable[bytes]) -> Iterator[bytes]:
     decoder.decode(b"", final=True)
     if pending_cr:
         yield b"\n"
+
+
+def _with_structure(
+    snapshot: ArtifactSnapshot,
+    path: Path,
+    relative_path: str,
+    kind: ArtifactKind,
+) -> ArtifactSnapshot:
+    if kind not in {"json", "jsonl"}:
+        return snapshot
+    if snapshot.comparison_size_bytes > MAX_STRUCTURED_BYTES:
+        raise ArtifactContractDiffError(
+            f"structured artifact exceeds summary limit: {relative_path}"
+        )
+    return replace(snapshot, structure=_structure(path, relative_path, kind))
+
+
+def _structure(path: Path, relative_path: str, kind: ArtifactKind) -> dict[str, Any]:
+    text = _structured_text(path, relative_path)
+    if kind == "jsonl":
+        return _summarize(_jsonl_records(text, relative_path), "jsonl", relative_path)
+    try:
+        value = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ArtifactContractDiffError(f"invalid JSON artifact: {relative_path}") from exc
+    records = value if isinstance(value, list) else [value]
+    manifest = value if isinstance(value, Mapping) else None
+    return _summarize(records, _container(value), relative_path, manifest)
+
+
+def _structured_text(path: Path, relative_path: str) -> str:
+    try:
+        with _regular_file(path, relative_path) as handle:
+            raw = handle.read(2 * MAX_STRUCTURED_BYTES + 1)
+        normalized = normalize_artifact_text_bytes(raw)
+    except ArtifactContractDiffError:
+        raise
+    except UnicodeDecodeError as exc:
+        raise ArtifactContractDiffError(
+            f"text artifact is not valid UTF-8: {relative_path}"
+        ) from exc
+    except OSError as exc:
+        raise ArtifactContractDiffError(f"cannot read artifact: {relative_path}") from exc
+    if len(normalized) > MAX_STRUCTURED_BYTES:
+        raise ArtifactContractDiffError(
+            f"structured artifact exceeds summary limit: {relative_path}"
+        )
+    return normalized.decode("utf-8")
+
+
+def _jsonl_records(text: str, relative_path: str) -> Iterator[object]:
+    for line_number, line in enumerate(io.StringIO(text), start=1):
+        if not line.strip():
+            continue
+        try:
+            yield json.loads(line)
+        except json.JSONDecodeError as exc:
+            raise ArtifactContractDiffError(
+                f"invalid JSONL artifact: {relative_path}:{line_number}"
+            ) from exc
+
+
+def _summarize(
+    records: Iterable[object],
+    container: str,
+    relative_path: str,
+    manifest: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    count = 0
+    keys: set[str] = set()
+    markers: dict[str, set[str]] = {}
+    for record in records:
+        count += 1
+        if not isinstance(record, Mapping):
+            continue
+        for key in record:
+            if not _safe_metadata(key):
+                raise ArtifactContractDiffError(f"unsafe JSON key: {relative_path}")
+            keys.add(key)
+        for field in ("$id", "$schema", "schema_id", "schema_version"):
+            candidate = record.get(field)
+            if isinstance(candidate, str):
+                _marker(markers, field, candidate, relative_path)
+        versions = record.get("artifact_schema_versions")
+        if isinstance(versions, Mapping):
+            for field, value in versions.items():
+                if isinstance(field, str) and isinstance(value, str):
+                    _marker(
+                        markers,
+                        f"artifact_schema_versions.{field}",
+                        value,
+                        relative_path,
+                    )
+        if len(keys) > MAX_STRUCTURE_ITEMS:
+            raise ArtifactContractDiffError(f"too many JSON keys: {relative_path}")
+
+    summary: dict[str, Any] = {
+        "container": container,
+        "record_count": count,
+        "top_level_keys": sorted(keys),
+    }
+    if markers:
+        summary["schema_versions"] = {
+            field: sorted(values) for field, values in sorted(markers.items())
+        }
+    if manifest is not None:
+        digests = _manifest_digests(manifest, relative_path)
+        if digests:
+            summary["run_manifest_digests"] = digests
+    return summary
+
+
+def _safe_metadata(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) <= 1_024
+        and not any(ord(character) < 32 for character in value)
+        and _LOCAL_PATH.match(value) is None
+        and not value.lower().startswith("file:")
+    )
+
+
+def _marker(
+    markers: dict[str, set[str]], field: str, value: str, relative_path: str
+) -> None:
+    if not _safe_metadata(field) or not _safe_metadata(value):
+        raise ArtifactContractDiffError(f"unsafe schema marker: {relative_path}")
+    markers.setdefault(field, set()).add(value)
+    if sum(len(values) for values in markers.values()) > MAX_STRUCTURE_ITEMS:
+        raise ArtifactContractDiffError(f"too many schema markers: {relative_path}")
+
+
+def _manifest_digests(value: Mapping[str, Any], relative_path: str) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for field in RUN_MANIFEST_DIGEST_FIELDS:
+        if field not in value:
+            continue
+        candidate = value[field]
+        if field in {"input_digest", "config_digest"}:
+            if not isinstance(candidate, str) or _DIGEST.fullmatch(candidate) is None:
+                raise ArtifactContractDiffError(
+                    f"invalid run-manifest digest: {relative_path}"
+                )
+            result[field] = candidate
+            continue
+        if not isinstance(candidate, Mapping) or len(candidate) > MAX_DIGEST_ENTRIES:
+            raise ArtifactContractDiffError(
+                f"invalid run-manifest digest map: {relative_path}"
+            )
+        checked: dict[str, str] = {}
+        for item_path, digest in candidate.items():
+            if (
+                not _safe_relative_path(item_path)
+                or not isinstance(digest, str)
+                or _DIGEST.fullmatch(digest) is None
+            ):
+                raise ArtifactContractDiffError(
+                    f"invalid run-manifest digest map: {relative_path}"
+                )
+            checked[item_path] = digest
+        result[field] = dict(sorted(checked.items()))
+    return result
+
+
+def _container(value: object) -> str:
+    if isinstance(value, Mapping):
+        return "object"
+    if isinstance(value, list):
+        return "array"
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "boolean"
+    if isinstance(value, (int, float)):
+        return "number"
+    return "string"
+
+
+def _change_reasons(
+    expected: ArtifactSnapshot, actual: ArtifactSnapshot
+) -> tuple[ChangeReason, ...]:
+    reasons: list[ChangeReason] = ["content-changed"]
+    before = expected.structure or {}
+    after = actual.structure or {}
+    structural_fields = ("container", "record_count", "top_level_keys")
+    if any(before.get(field) != after.get(field) for field in structural_fields):
+        reasons.append("structure-changed")
+    if before.get("schema_versions") != after.get("schema_versions"):
+        reasons.append("schema-version-changed")
+    if before.get("run_manifest_digests") != after.get("run_manifest_digests"):
+        reasons.append("run-manifest-digest-changed")
+    return tuple(reasons)