9f0ffe031c
Immutable content-addressed artifact versions with If-Match, lineage that must resolve under the caller, unified diffs, promotion; sources/passages/citations with server-side hashing and dedupe, citation integrity checks and revisioned research notebooks. Cross-tenant access is 404 and audited. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
"""Version-to-version comparison for artifacts (HUX-04).
|
|
|
|
Text artifacts get a unified diff from ``difflib``; anything that is not
|
|
valid UTF-8, or whose artifact type is binary, is described by sizes and
|
|
hashes only so the UI never tries to render bytes as text.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import hashlib
|
|
from typing import Any
|
|
|
|
TEXT_TYPES = frozenset({"markdown", "code", "html", "svg", "json", "csv"})
|
|
|
|
|
|
def is_text_type(artifact_type: str) -> bool:
|
|
"""True for artifact types whose content is served as UTF-8 text."""
|
|
return artifact_type in TEXT_TYPES
|
|
|
|
|
|
def decode_text(data: bytes) -> str | None:
|
|
"""UTF-8 decode, or None when the bytes are not text."""
|
|
try:
|
|
return data.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
|
|
|
|
def unified(artifact_type: str, older: bytes, newer: bytes, from_n: int, to_n: int) -> dict[str, Any]:
|
|
"""Diff record ``{"from", "to", "unified"|"binary"}`` for two versions."""
|
|
record: dict[str, Any] = {"from": from_n, "to": to_n}
|
|
old_text = decode_text(older) if is_text_type(artifact_type) else None
|
|
new_text = decode_text(newer) if is_text_type(artifact_type) else None
|
|
if old_text is None or new_text is None:
|
|
record["binary"] = {
|
|
"from_bytes": len(older),
|
|
"to_bytes": len(newer),
|
|
"from_hash": "sha256:" + hashlib.sha256(older).hexdigest(),
|
|
"to_hash": "sha256:" + hashlib.sha256(newer).hexdigest(),
|
|
}
|
|
return record
|
|
lines = difflib.unified_diff(
|
|
old_text.splitlines(keepends=True),
|
|
new_text.splitlines(keepends=True),
|
|
fromfile=f"v{from_n}",
|
|
tofile=f"v{to_n}",
|
|
)
|
|
record["unified"] = "".join(lines)
|
|
return record
|