commit_observer.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/git/commit_observer.py
 
"""
Purpose
-------
Detect silent mutations of tracked-but-unstaged files during the
``oct git commit`` pipeline (B1.0). Snapshots the working-tree state
of every tracked file outside the staged set at the start of the
commit, then compares again right before and right after the
underlying ``git commit`` invocation. Any byte-level change is
reported as either a warning or a hard abort, depending on the
``git.observe_unstaged_mutation`` octrc setting.
 
Responsibilities
----------------
- Provide :func:`snapshot_unstaged_tracked` — returns a mapping of
  project-relative path -> sha256 hex digest for every tracked file
  *not* in the staged set.
- Provide :func:`diff_snapshots` — returns the list of paths whose
  hash changed (new file, deleted file, or content drift) between two
  snapshots.
- Be defensive: large files (> :data:`_MAX_HASH_BYTES`) are recorded
  with the sentinel digest ``"OVERSIZE"`` so they don't dominate
  commit latency, and a missing/unreadable file maps to ``""`` so the
  diff path treats it consistently.
 
Diagnostics
-----------
Domain: GIT
Levels:
    L1 — errors: file IO failures during snapshot
    L2 — lifecycle: snapshot start / snapshot complete
    L3 — details: file count, total bytes hashed
    L4 — deep trace: per-file hash + size
 
Contracts
---------
- This module is pure: no Click, no subprocess of its own. It uses
  :func:`oct.core.git.git_run` to enumerate tracked files.
- Snapshot calls must never raise on individual-file IO errors —
  they record the file as unreadable and continue.
- The hash digest is content-only (no metadata). Renames are detected
  via the path key, not the digest.
"""
 
from __future__ import annotations
 
import hashlib
from pathlib import Path
 
from oct.core.diagnostics import _dbg
 
 
#: Files larger than this are recorded with sentinel ``"OVERSIZE"``
#: rather than hashed. Keeps the snapshot cost bounded on repos with
#: large generated artefacts checked into git.
_MAX_HASH_BYTES: int = 10 * 1024 * 1024  # 10 MB
 
#: Sentinel digest used when a file exceeds :data:`_MAX_HASH_BYTES`.
_OVERSIZE_SENTINEL: str = "OVERSIZE"
 
#: Sentinel digest used when a file is missing or unreadable. Empty
#: string is distinct from any valid sha256 hex digest.
_MISSING_SENTINEL: str = ""
 
 
def _hash_file(path: Path) -> str:
    """Return the sha256 hex digest of *path*, or a sentinel.
 
    Returns :data:`_OVERSIZE_SENTINEL` when the file is larger than
    :data:`_MAX_HASH_BYTES`, and :data:`_MISSING_SENTINEL` when the
    file is missing or unreadable. Never raises.
    """
    func = "_hash_file"
    try:
        size = path.stat().st_size
    except OSError as exc:
        _dbg("GIT", func, f"stat failed: {path} ({exc})", 1, override=True)
        return _MISSING_SENTINEL
 
    if size > _MAX_HASH_BYTES:
        _dbg("GIT", func, f"oversize: {path} ({size} bytes)", 4)
        return _OVERSIZE_SENTINEL
 
    try:
        h = hashlib.sha256()
        with path.open("rb") as fh:
            while True:
                chunk = fh.read(65536)
                if not chunk:
                    break
                h.update(chunk)
        digest = h.hexdigest()
        _dbg("GIT", func, f"hashed {path} -> {digest[:8]}…", 4)
        return digest
    except OSError as exc:
        _dbg("GIT", func, f"read failed: {path} ({exc})", 1, override=True)
        return _MISSING_SENTINEL
 
 
def snapshot_unstaged_tracked(
    repo_root: Path,
    staged_set: set[str] | None = None,
) -> dict[str, str]:
    """Return ``{rel_path: sha256_hex}`` for every tracked-but-unstaged file.
 
    *repo_root* must be the actual git repository root (not the OCT
    project root in monorepo layouts). *staged_set* is a set of
    repo-relative path strings already in the index; those files are
    excluded from the snapshot because their mutations are caller
    intent, not silent regression.
 
    The snapshot includes:
      - Tracked files that are clean (working tree == HEAD).
      - Tracked files modified but not staged (``M`` in
        ``git status -s`` second column).
 
    The snapshot excludes:
      - Files in *staged_set*.
      - Untracked files (``??``).
      - Submodules and worktree symlinks pointing outside *repo_root*.
    """
    func = "snapshot_unstaged_tracked"
    _dbg("GIT", func, f"repo_root={repo_root}", 2)
    from oct.core.git import git_run
 
    if staged_set is None:
        staged_set = set()
 
    proc = git_run(["ls-files"], cwd=Path(repo_root))
    snapshot: dict[str, str] = {}
    for raw in proc.stdout.splitlines():
        rel = raw.strip()
        if not rel or rel in staged_set:
            continue
        full = (repo_root / rel).resolve()
        try:
            full.relative_to(Path(repo_root).resolve())
        except ValueError:
            # Path escape — drop, don't hash.
            _dbg("GIT", func, f"path-escape skip: {rel}", 1, override=True)
            continue
        snapshot[rel] = _hash_file(full)
 
    _dbg(
        "GIT", func,
        f"snapshot complete: {len(snapshot)} file(s) "
        f"(staged={len(staged_set)} excluded)",
        3,
    )
    return snapshot
 
 
def diff_snapshots(
    before: dict[str, str],
    after: dict[str, str],
) -> list[str]:
    """Return the list of paths whose digest changed between snapshots.
 
    Includes:
      - Paths present in both with different digests (mutation).
      - Paths present in *before* but missing from *after* (deletion or
        unreadable).
      - Paths newly tracked in *after* (rare; would only appear if
        ``git add`` ran during the commit pipeline).
 
    A path with the same :data:`_OVERSIZE_SENTINEL` on both sides is
    NOT flagged — we cannot tell whether the content of an oversize
    file changed without a full hash. Document this caveat to the
    caller via the per-file payload (omitted from the simple list
    return; see :func:`diff_snapshots_detail` for the verbose form).
    """
    func = "diff_snapshots"
    keys = set(before.keys()) | set(after.keys())
    changed: list[str] = []
    for k in keys:
        b = before.get(k, _MISSING_SENTINEL)
        a = after.get(k, _MISSING_SENTINEL)
        if b == _OVERSIZE_SENTINEL and a == _OVERSIZE_SENTINEL:
            continue
        if b != a:
            changed.append(k)
    changed.sort()
    _dbg("GIT", func, f"changed={len(changed)}", 3)
    return changed