orchestrata_ultimate.semantic_graph._freshness

Reference › orchestrata_ultimate › semantic_graph › _freshness

Content-hash freshness tracking for semantic graph re-ingestion (audit item 7). The enrichment pass already writes with MERGE and is cheap to repeat, but the ingestion script clears and re-embeds the whole source tree on each run; this module supplies the primitive that makes that loop incremental without touching the script. file_digest returns the SHA-256 hex digest of a file, load_manifest and save_manifest read and write a path-to-digest manifest on disk (creating parent directories, with a default location), and changed_files returns the subset of paths whose digest differs from the manifest. The intended use in the docstring is load the manifest, compute the dirty files, re-ingest only those, then save. No module imports it yet and it touches no graph.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/semantic_graph/_freshness.py

"""Content-hash based freshness tracking for semantic-graph re-ingestion.

Audit item 7: re-running the enrichment/ingestion pipeline should be cheap.
``spec_trace_enricher``/``scripts/enrich_semantic_graph_specs.py`` already
write via ``MERGE`` (see the Memgraph adapter), so re-running *that* pass is
idempotent and inexpensive. The costly piece is
``scripts/ingest_orchestrata_ultimate.py`` (owned by a different workstream),
which unconditionally clears and fully re-embeds the source tree on every
run.

This module provides the reusable primitive for making that loop
incremental, without reaching into that script's ownership boundary:

    manifest = load_manifest()
    dirty = changed_files(all_source_files, manifest)
    # ... re-embed / re-ingest only `dirty` ...
    for path in all_source_files:
        manifest[str(path)] = file_digest(path)
    save_manifest(manifest)

Intended integration for ``ingest_orchestrata_ultimate.py``: before the
clear-and-rebuild pass, compute ``changed_files()`` against the persisted
manifest and skip (or only re-embed) files whose digest is unchanged; update
and persist the manifest after a successful run. This module is pure
filesystem/hash bookkeeping -- it never talks to the graph backend -- so it
is unit-testable without any live infra and safe to adopt incrementally.
"""

from __future__ import annotations

import hashlib
import json
import logging
from pathlib import Path

logger = logging.getLogger(__name__)

#: Default location of the persisted digest manifest.
DEFAULT_MANIFEST_PATH = Path("plans/.semantic_graph_digests.json")

#: Chunk size used when streaming a file through the hash function.
_CHUNK_SIZE = 65536


def file_digest(path: Path) -> str:
    """Return the sha256 hex digest of a file's contents.

    Args:
        path: File to hash.

    Returns:
        Hex-encoded sha256 digest of the file's bytes.

    Raises:
        OSError: If the file cannot be opened/read.
    """
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        while chunk := handle.read(_CHUNK_SIZE):
            digest.update(chunk)
    return digest.hexdigest()


def load_manifest(manifest_path: Path = DEFAULT_MANIFEST_PATH) -> dict[str, str]:
    """Load a digest manifest from disk.

    Args:
        manifest_path: Location of the manifest JSON file.

    Returns:
        Mapping from (string) file path to its last-known sha256 digest.
        Returns ``{}`` if the file is missing, unreadable, or not a JSON
        object -- freshness tracking degrades gracefully to "everything is
        changed" rather than raising.
    """
    if not manifest_path.is_file():
        return {}
    try:
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        logger.warning("Cannot read digest manifest %s: %s", manifest_path, exc)
        return {}
    if not isinstance(data, dict):
        logger.warning(
            "Digest manifest %s is not a JSON object; ignoring", manifest_path
        )
        return {}
    return {str(key): str(value) for key, value in data.items()}


def save_manifest(
    manifest: dict[str, str], manifest_path: Path = DEFAULT_MANIFEST_PATH
) -> None:
    """Persist a digest manifest to disk, creating parent directories as needed.

    Args:
        manifest: Mapping from file path (string) to sha256 digest.
        manifest_path: Location to write the manifest JSON file.
    """
    manifest_path.parent.mkdir(parents=True, exist_ok=True)
    manifest_path.write_text(
        json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8"
    )


def changed_files(paths: list[Path], manifest: dict[str, str]) -> list[Path]:
    """Return the subset of ``paths`` whose content differs from the manifest.

    A path is considered changed if it is absent from the manifest or its
    current digest differs from the recorded one. Unreadable paths fail
    open (are reported as changed) so a transient read error never causes a
    real change to be silently skipped.

    Args:
        paths: Candidate files to check.
        manifest: Digest manifest as returned by :func:`load_manifest`.

    Returns:
        The subset of ``paths`` considered changed, in input order.
    """
    changed: list[Path] = []
    for path in paths:
        key = str(path)
        try:
            digest = file_digest(path)
        except OSError as exc:
            logger.warning("Cannot digest %s: %s", path, exc)
            changed.append(path)
            continue
        if manifest.get(key) != digest:
            changed.append(path)
    return changed