snapshots.py

"""Self-contained, atomically published web-policy source snapshots.

Legacy archives under ``corpus/versions/<corpus_version>/`` retain processed
chunks only.  Schema-2 archives in ``corpus/snapshots/<snapshot_version>/``
retain the ordered chunks, normalized manifest, exact raw source bytes, and
exact fetch receipts needed to recompute the full content and snapshot
identities.

Publication is fail-closed: build in a hidden same-filesystem directory, fsync,
reload and validate every artifact, then rename into place.  An existing archive
is never overwritten; retries and concurrent writers must validate the winner.
"""

from __future__ import annotations

import dataclasses
import fcntl
import hashlib
import json
import os
import re
import shutil
import tempfile
import threading
from collections.abc import Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path

import yaml

from assistant import config
from assistant.identity import (
    CONTENT_SCHEMA,
    SNAPSHOT_SCHEMA,
    DocumentObservation,
    IdentityValidationError,
    SnapshotIdentity,
    build_content_identity,
    build_snapshot_identity,
)
from assistant.ingest import Chunk, load_chunks

ARCHIVE_SCHEMA = 2
ARCHIVE_SCOPE = "web_policy"
EVIDENCE_SCHEMA = "fare-assistant.source-evidence.v1"

_DOCUMENT_ID = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_SUPPORTED_FORMATS = frozenset({"html", "pdf"})
_MANIFEST_FIELDS = ("id", "agency", "agency_full", "title", "url", "format", "language")
_PROCESS_ARCHIVE_LOCK = threading.Lock()


class SnapshotArchiveError(ValueError):
    """A source set or on-disk snapshot is incomplete, corrupt, or conflicting."""


@dataclass(frozen=True)
class SourceMaterial:
    """The exact bytes and normalized metadata archived for one document."""

    manifest_document: dict[str, str]
    observation: DocumentObservation
    raw_name: str
    raw_bytes: bytes
    metadata_name: str
    metadata_bytes: bytes


@dataclass(frozen=True)
class SnapshotMaterial:
    """Validated ordered chunks and their complete source evidence."""

    chunks: tuple[Chunk, ...]
    sources: tuple[SourceMaterial, ...]
    identity: SnapshotIdentity


def _required_string(mapping: Mapping[str, object], key: str, context: str) -> str:
    value = mapping.get(key)
    if not isinstance(value, str) or not value.strip():
        raise SnapshotArchiveError(f"{context}.{key} must be a non-empty string")
    return value


def _document_id(value: object, context: str) -> str:
    if not isinstance(value, str) or not _DOCUMENT_ID.fullmatch(value):
        raise SnapshotArchiveError(
            f"{context} must contain lowercase letters, digits, and single hyphens"
        )
    return value


def _manifest_documents(manifest: Mapping[str, object]) -> dict[str, dict[str, str]]:
    raw_documents = manifest.get("documents")
    if not isinstance(raw_documents, list) or not raw_documents:
        raise SnapshotArchiveError("manifest.documents must be a non-empty list")

    documents: dict[str, dict[str, str]] = {}
    for index, item in enumerate(raw_documents):
        context = f"manifest.documents[{index}]"
        if not isinstance(item, Mapping):
            raise SnapshotArchiveError(f"{context} must be a mapping")
        doc_id = _document_id(item.get("id"), f"{context}.id")
        if doc_id in documents:
            raise SnapshotArchiveError(f"manifest contains duplicate document id: {doc_id}")
        language = item.get("language", "en")
        if not isinstance(language, str) or not language.strip():
            raise SnapshotArchiveError(f"{context}.language must be a non-empty string")
        normalized: dict[str, str] = {
            "id": doc_id,
            "agency": _required_string(item, "agency", context),
            "agency_full": _required_string(item, "agency_full", context),
            "title": _required_string(item, "title", context),
            "url": _required_string(item, "url", context),
            "language": language,
        }
        explicit_format = item.get("format")
        if explicit_format is not None:
            if not isinstance(explicit_format, str) or explicit_format not in _SUPPORTED_FORMATS:
                raise SnapshotArchiveError(
                    f"{context}.format must be one of {sorted(_SUPPORTED_FORMATS)}"
                )
            normalized["format"] = explicit_format
        documents[doc_id] = normalized
    return documents


def _chunk_documents(chunks: tuple[Chunk, ...]) -> dict[str, list[Chunk]]:
    by_document: dict[str, list[Chunk]] = {}
    for chunk in chunks:
        by_document.setdefault(chunk.doc_id, []).append(chunk)
    return by_document


def _assert_manifest_matches_chunks(
    document: Mapping[str, str],
    document_chunks: list[Chunk],
) -> None:
    expected = {
        "doc_id": document["id"],
        "agency": document["agency"],
        "agency_full": document["agency_full"],
        "doc_title": document["title"],
        "url": document["url"],
        "language": document["language"],
    }
    for field, value in expected.items():
        if any(getattr(chunk, field) != value for chunk in document_chunks):
            raise SnapshotArchiveError(
                f"manifest {document['id']!r} does not match chunk field {field}"
            )


def _read_regular_file(path: Path, context: str) -> bytes:
    if path.is_symlink() or not path.is_file():
        raise SnapshotArchiveError(f"{context} is missing or is not a regular file: {path}")
    return path.read_bytes()


def _load_receipt(metadata_bytes: bytes, doc_id: str) -> Mapping[str, object]:
    try:
        metadata = yaml.safe_load(metadata_bytes)
    except yaml.YAMLError as exc:
        raise SnapshotArchiveError(f"metadata for {doc_id} is invalid YAML") from exc
    if not isinstance(metadata, Mapping):
        raise SnapshotArchiveError(f"metadata for {doc_id} must be a mapping")
    return metadata


def _source_from_captured_bytes(
    document: Mapping[str, str],
    *,
    metadata_bytes: bytes,
    raw_bytes: bytes,
) -> SourceMaterial:
    """Validate and normalize one already-captured receipt/raw byte pair."""
    doc_id = document["id"]
    metadata = _load_receipt(metadata_bytes, doc_id)
    effective_format = metadata.get("format")
    if effective_format not in _SUPPORTED_FORMATS:
        raise SnapshotArchiveError(f"metadata for {doc_id} has unsupported or missing format")
    assert isinstance(effective_format, str)
    explicit_format = document.get("format")
    if explicit_format is not None and explicit_format != effective_format:
        raise SnapshotArchiveError(f"manifest and metadata formats disagree for {doc_id}")

    try:
        observation = DocumentObservation.from_metadata(
            metadata,
            raw=raw_bytes,
            effective_format=effective_format,
        )
    except IdentityValidationError as exc:
        raise SnapshotArchiveError(f"{doc_id}: {exc}") from exc
    if observation.doc_id != doc_id:
        raise SnapshotArchiveError(f"metadata doc_id does not match manifest for {doc_id}")
    if observation.requested_url != document["url"]:
        raise SnapshotArchiveError(f"metadata requested URL does not match manifest for {doc_id}")

    normalized_document = dict(document)
    normalized_document["format"] = effective_format
    return SourceMaterial(
        manifest_document={key: normalized_document[key] for key in _MANIFEST_FIELDS},
        observation=observation,
        raw_name=f"{doc_id}.{effective_format}",
        raw_bytes=raw_bytes,
        metadata_name=f"{doc_id}.meta.yaml",
        metadata_bytes=metadata_bytes,
    )


def capture_source_material(
    manifest: Mapping[str, object],
    raw_dir: Path,
) -> tuple[SourceMaterial, ...]:
    """Capture and validate every manifest source exactly once.

    Callers that derive chunks should parse these retained ``raw_bytes`` and
    pass the returned tuple back to :func:`archive_snapshot`. That binds the
    archived evidence to the exact bytes used for processing even if a
    concurrent fetch replaces files in ``raw_dir`` later.
    """
    documents = _manifest_documents(manifest)
    sources: list[SourceMaterial] = []
    for doc_id in sorted(documents):
        document = documents[doc_id]
        metadata_bytes = _read_regular_file(
            raw_dir / f"{doc_id}.meta.yaml",
            f"metadata for {doc_id}",
        )
        metadata = _load_receipt(metadata_bytes, doc_id)
        effective_format = metadata.get("format")
        if effective_format not in _SUPPORTED_FORMATS:
            raise SnapshotArchiveError(f"metadata for {doc_id} has unsupported or missing format")
        assert isinstance(effective_format, str)
        raw_name = f"{doc_id}.{effective_format}"
        raw_bytes = _read_regular_file(
            raw_dir / raw_name,
            f"raw source for {doc_id}",
        )
        ambiguous = [
            raw_dir / f"{doc_id}.{candidate}"
            for candidate in _SUPPORTED_FORMATS - {effective_format}
            if (raw_dir / f"{doc_id}.{candidate}").exists()
        ]
        if ambiguous:
            raise SnapshotArchiveError(
                f"raw source format is ambiguous for {doc_id}: "
                + ", ".join(path.name for path in ambiguous)
            )
        sources.append(
            _source_from_captured_bytes(
                document,
                metadata_bytes=metadata_bytes,
                raw_bytes=raw_bytes,
            )
        )
    return tuple(sources)


def _validate_captured_sources(
    documents: Mapping[str, Mapping[str, str]],
    sources: tuple[SourceMaterial, ...],
) -> tuple[SourceMaterial, ...]:
    if not isinstance(sources, tuple):
        raise SnapshotArchiveError("captured sources must be an immutable tuple")
    if not sources:
        raise SnapshotArchiveError("captured sources must not be empty")
    by_document: dict[str, SourceMaterial] = {}
    for index, source in enumerate(sources):
        if not isinstance(source, SourceMaterial):
            raise SnapshotArchiveError(f"captured sources[{index}] must be SourceMaterial")
        doc_id = source.observation.doc_id
        if doc_id in by_document:
            raise SnapshotArchiveError(f"captured sources contain duplicate document id: {doc_id}")
        by_document[doc_id] = source
    if set(by_document) != set(documents):
        missing = sorted(set(documents) - set(by_document))
        unexpected = sorted(set(by_document) - set(documents))
        details: list[str] = []
        if missing:
            details.append("missing captured sources: " + ", ".join(missing))
        if unexpected:
            details.append("unexpected captured sources: " + ", ".join(unexpected))
        raise SnapshotArchiveError("; ".join(details))

    validated: list[SourceMaterial] = []
    for doc_id in sorted(documents):
        supplied = by_document[doc_id]
        rebuilt = _source_from_captured_bytes(
            documents[doc_id],
            metadata_bytes=supplied.metadata_bytes,
            raw_bytes=supplied.raw_bytes,
        )
        if supplied != rebuilt:
            raise SnapshotArchiveError(
                f"captured source metadata does not match retained bytes for {doc_id}"
            )
        validated.append(rebuilt)
    return tuple(validated)


def collect_snapshot_material(
    chunks: list[Chunk] | tuple[Chunk, ...],
    manifest: Mapping[str, object],
    raw_dir: Path,
    *,
    sources: tuple[SourceMaterial, ...] | None = None,
) -> SnapshotMaterial:
    """Validate one complete manifest/chunk/raw/receipt set."""
    ordered_chunks = tuple(chunks)
    # The identity builder provides strict Chunk validation before paths are used.
    try:
        build_content_identity(ordered_chunks)
    except IdentityValidationError as exc:
        raise SnapshotArchiveError(str(exc)) from exc

    documents = _manifest_documents(manifest)
    chunks_by_document = _chunk_documents(ordered_chunks)
    manifest_ids = set(documents)
    chunk_ids = set(chunks_by_document)
    if manifest_ids != chunk_ids:
        missing = sorted(manifest_ids - chunk_ids)
        unexpected = sorted(chunk_ids - manifest_ids)
        details: list[str] = []
        if missing:
            details.append("manifest documents with zero chunks: " + ", ".join(missing))
        if unexpected:
            details.append("chunks absent from manifest: " + ", ".join(unexpected))
        raise SnapshotArchiveError("; ".join(details))

    captured_sources = (
        capture_source_material(manifest, raw_dir)
        if sources is None
        else _validate_captured_sources(documents, sources)
    )
    for doc_id, document in documents.items():
        _assert_manifest_matches_chunks(document, chunks_by_document[doc_id])

    try:
        identity = build_snapshot_identity(
            ordered_chunks,
            [source.observation for source in captured_sources],
        )
    except IdentityValidationError as exc:
        raise SnapshotArchiveError(str(exc)) from exc
    return SnapshotMaterial(ordered_chunks, captured_sources, identity)


def _json_bytes(payload: object) -> bytes:
    return (json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode(
        "utf-8"
    )


def _chunks_bytes(chunks: tuple[Chunk, ...]) -> bytes:
    lines = [
        json.dumps(
            dataclasses.asdict(chunk),
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        )
        for chunk in chunks
    ]
    return ("\n".join(lines) + "\n").encode("utf-8")


def _write_file(path: Path, content: bytes) -> None:
    """Create one staged artifact durably; exposed for fault-injection tests."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("xb") as handle:
        handle.write(content)
        handle.flush()
        os.fsync(handle.fileno())


def _fsync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


@contextmanager
def _archive_root_lock(root: Path):
    """Serialize cooperative writers across both threads and processes."""
    lock_path = root / ".archive.lock"
    with _PROCESS_ARCHIVE_LOCK:
        descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
        try:
            fcntl.flock(descriptor, fcntl.LOCK_EX)
            yield
        finally:
            fcntl.flock(descriptor, fcntl.LOCK_UN)
            os.close(descriptor)


def _artifact_manifest(root: Path) -> dict[str, dict[str, object]]:
    artifacts: dict[str, dict[str, object]] = {}
    for path in sorted(root.rglob("*")):
        relative = path.relative_to(root).as_posix()
        if path.is_symlink():
            raise SnapshotArchiveError(f"snapshot artifact {relative} is a symbolic link")
        if path == root / "version.json" or path.is_dir():
            continue
        content = _read_regular_file(path, f"snapshot artifact {relative}")
        artifacts[relative] = {
            "sha256": hashlib.sha256(content).hexdigest(),
            "bytes": len(content),
        }
    return artifacts


def _archive_metadata(
    material: SnapshotMaterial,
    artifacts: dict[str, dict[str, object]],
    archived_at: str,
) -> dict[str, object]:
    from assistant.corpus import corpus_version

    chunks = material.chunks
    identity = material.identity
    return {
        "archive_schema": ARCHIVE_SCHEMA,
        "scope": ARCHIVE_SCOPE,
        "identity_schemas": {
            "content": CONTENT_SCHEMA,
            "snapshot": SNAPSHOT_SCHEMA,
        },
        "corpus_version": corpus_version(list(chunks)),
        "content_version": identity.content_version,
        "content_version_short": identity.content_version[:12],
        "snapshot_version": identity.snapshot_version,
        "snapshot_version_short": identity.snapshot_version[:12],
        "as_of": max(chunk.fetch_date for chunk in chunks),
        "agencies": sorted({chunk.agency for chunk in chunks}),
        "documents": len({chunk.doc_id for chunk in chunks}),
        "chunks": len(chunks),
        "evidence": "raw_and_receipts_verified",
        "artifacts": artifacts,
        "archived_at": archived_at,
    }


def _write_staged_archive(
    stage: Path,
    material: SnapshotMaterial,
    *,
    archived_at: str,
) -> None:
    _write_file(stage / "chunks.jsonl", _chunks_bytes(material.chunks))
    manifest_snapshot = {
        "archive_schema": ARCHIVE_SCHEMA,
        "scope": ARCHIVE_SCOPE,
        "documents": [source.manifest_document for source in material.sources],
    }
    _write_file(
        stage / "manifest.snapshot.yaml",
        yaml.safe_dump(manifest_snapshot, sort_keys=False).encode("utf-8"),
    )
    evidence = {
        "schema": EVIDENCE_SCHEMA,
        "observations": [source.observation.canonical_dict() for source in material.sources],
    }
    _write_file(stage / "source-evidence.json", _json_bytes(evidence))
    for source in material.sources:
        _write_file(stage / "raw" / source.raw_name, source.raw_bytes)
        _write_file(stage / "raw" / source.metadata_name, source.metadata_bytes)
    _fsync_directory(stage / "raw")
    artifacts = _artifact_manifest(stage)
    _write_file(
        stage / "version.json",
        _json_bytes(_archive_metadata(material, artifacts, archived_at)),
    )
    _fsync_directory(stage)


def _load_json_mapping(path: Path, context: str) -> Mapping[str, object]:
    try:
        payload = json.loads(_read_regular_file(path, context))
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise SnapshotArchiveError(f"{context} is invalid JSON") from exc
    if not isinstance(payload, Mapping):
        raise SnapshotArchiveError(f"{context} must be a JSON object")
    return payload


def _load_archived_observations(path: Path) -> tuple[dict[str, object], ...]:
    payload = _load_json_mapping(path, "source evidence")
    if payload.get("schema") != EVIDENCE_SCHEMA:
        raise SnapshotArchiveError("source evidence has an unsupported schema")
    rows = payload.get("observations")
    if not isinstance(rows, list) or not rows:
        raise SnapshotArchiveError("source evidence observations must be a non-empty list")
    expected_fields = {field.name for field in dataclasses.fields(DocumentObservation)}
    observations: list[dict[str, object]] = []
    for index, row in enumerate(rows):
        if not isinstance(row, Mapping):
            raise SnapshotArchiveError(f"source evidence observation {index} is not an object")
        if set(row) != expected_fields:
            raise SnapshotArchiveError(
                f"source evidence observation {index} has an invalid field set"
            )
        observations.append(dict(row))
    return tuple(observations)


def validate_snapshot_archive(
    path: Path,
    *,
    require_identity_directory_name: bool = True,
) -> SnapshotIdentity:
    """Reload and cryptographically verify every schema-2 archive artifact."""
    if path.is_symlink() or not path.is_dir():
        raise SnapshotArchiveError(f"snapshot archive is not a regular directory: {path}")
    version = _load_json_mapping(path / "version.json", "snapshot version metadata")
    if version.get("archive_schema") != ARCHIVE_SCHEMA:
        raise SnapshotArchiveError("snapshot archive has an unsupported schema")
    if version.get("scope") != ARCHIVE_SCOPE:
        raise SnapshotArchiveError("snapshot archive has an unsupported scope")

    manifest_bytes = _read_regular_file(
        path / "manifest.snapshot.yaml",
        "snapshot manifest",
    )
    try:
        manifest = yaml.safe_load(manifest_bytes)
    except yaml.YAMLError as exc:
        raise SnapshotArchiveError("snapshot manifest is invalid YAML") from exc
    if not isinstance(manifest, Mapping):
        raise SnapshotArchiveError("snapshot manifest must be a mapping")

    try:
        chunks = tuple(load_chunks(path / "chunks.jsonl"))
    except (OSError, UnicodeError, json.JSONDecodeError, TypeError, KeyError) as exc:
        raise SnapshotArchiveError("snapshot chunks are malformed or unreadable") from exc
    material = collect_snapshot_material(list(chunks), manifest, path / "raw")
    archived_observations = _load_archived_observations(path / "source-evidence.json")
    if archived_observations != tuple(
        source.observation.canonical_dict() for source in material.sources
    ):
        raise SnapshotArchiveError("source evidence does not match archived raw receipts")

    artifacts = _artifact_manifest(path)
    if version.get("artifacts") != artifacts:
        raise SnapshotArchiveError("snapshot artifact digest manifest does not match")

    from assistant.corpus import corpus_version

    identity = material.identity
    expected_scalars = {
        "corpus_version": corpus_version(list(chunks)),
        "content_version": identity.content_version,
        "content_version_short": identity.content_version[:12],
        "snapshot_version": identity.snapshot_version,
        "snapshot_version_short": identity.snapshot_version[:12],
        "as_of": max(chunk.fetch_date for chunk in chunks),
        "agencies": sorted({chunk.agency for chunk in chunks}),
        "documents": len({chunk.doc_id for chunk in chunks}),
        "chunks": len(chunks),
        "evidence": "raw_and_receipts_verified",
        "identity_schemas": {
            "content": CONTENT_SCHEMA,
            "snapshot": SNAPSHOT_SCHEMA,
        },
    }
    for key, expected in expected_scalars.items():
        if version.get(key) != expected:
            raise SnapshotArchiveError(f"snapshot version metadata field {key} does not match")

    archived_at = version.get("archived_at")
    if not isinstance(archived_at, str):
        raise SnapshotArchiveError("snapshot archived_at must be an ISO timestamp")
    try:
        datetime.fromisoformat(archived_at)
    except ValueError as exc:
        raise SnapshotArchiveError("snapshot archived_at must be an ISO timestamp") from exc

    if require_identity_directory_name and path.name != identity.snapshot_version:
        raise SnapshotArchiveError("snapshot directory name does not match snapshot_version")
    return identity


def archive_snapshot(
    chunks: list[Chunk],
    manifest: Mapping[str, object],
    *,
    raw_dir: Path | None = None,
    snapshots_dir: Path | None = None,
    archived_at: str | None = None,
    sources: tuple[SourceMaterial, ...] | None = None,
) -> SnapshotIdentity:
    """Stage, verify, and atomically retain a complete schema-2 snapshot."""
    source_dir = raw_dir if raw_dir is not None else config.RAW_DIR
    destination_root = snapshots_dir if snapshots_dir is not None else config.SNAPSHOTS_DIR
    material = collect_snapshot_material(
        chunks,
        manifest,
        source_dir,
        sources=sources,
    )
    destination_root.mkdir(parents=True, exist_ok=True)
    final = destination_root / material.identity.snapshot_version
    with _archive_root_lock(destination_root):
        if final.exists() or final.is_symlink():
            winner = validate_snapshot_archive(final)
            if winner != material.identity:
                raise SnapshotArchiveError("existing snapshot identity conflicts with input")
            return winner

    timestamp = archived_at or datetime.now(UTC).isoformat(timespec="seconds")
    stage = Path(
        tempfile.mkdtemp(
            prefix=f".{material.identity.snapshot_version[:12]}.",
            dir=destination_root,
        )
    )
    try:
        _write_staged_archive(stage, material, archived_at=timestamp)
        staged_identity = validate_snapshot_archive(
            stage,
            require_identity_directory_name=False,
        )
        if staged_identity != material.identity:
            raise SnapshotArchiveError("staged snapshot identity changed during write")
        with _archive_root_lock(destination_root):
            if final.exists() or final.is_symlink():
                winner = validate_snapshot_archive(final)
                if winner != material.identity:
                    raise SnapshotArchiveError("concurrent snapshot winner conflicts with input")
                return winner
            os.rename(stage, final)
            _fsync_directory(destination_root)
        return validate_snapshot_archive(final)
    finally:
        if stage.exists():
            shutil.rmtree(stage)


def list_snapshots(snapshots_dir: Path | None = None) -> list[str]:
    """Return verified schema-2 snapshot IDs oldest first; ignore hidden stages."""
    root = snapshots_dir if snapshots_dir is not None else config.SNAPSHOTS_DIR
    if not root.exists():
        return []
    entries: list[tuple[str, str]] = []
    for path in root.iterdir():
        if path.name.startswith(".") or not path.is_dir() or path.is_symlink():
            continue
        if not _SHA256.fullmatch(path.name):
            continue
        validate_snapshot_archive(path)
        version = _load_json_mapping(path / "version.json", "snapshot version metadata")
        entries.append((str(version["archived_at"]), path.name))
    return [snapshot for _, snapshot in sorted(entries)]


def load_snapshot_chunks(
    snapshot: str,
    snapshots_dir: Path | None = None,
) -> list[Chunk]:
    """Load chunks only after validating the complete referenced snapshot."""
    if not _SHA256.fullmatch(snapshot):
        raise FileNotFoundError(f"invalid snapshot version: {snapshot!r}")
    root = snapshots_dir if snapshots_dir is not None else config.SNAPSHOTS_DIR
    path = root / snapshot
    if not path.exists():
        known = ", ".join(list_snapshots(root)) or "(none)"
        raise FileNotFoundError(
            f"snapshot version {snapshot!r} is not archived under {root} (known snapshots: {known})"
        )
    validate_snapshot_archive(path)
    return load_chunks(path / "chunks.jsonl")