zettelkasten.commit

zettelkasten.commit

Debounced, path-scoped git commits for .zettelkasten/ files.

Literature-review manifests, notes, and other .zettelkasten/ files represent weeks of authoring effort, so writes must be version-controlled and crash-safe. This module owns a debounced committer that batches scheduled file paths and commits ONLY those paths — the user's unrelated working-tree changes are never swept in. It mirrors memory.commit.MemoryCommitter with two deliberate differences:

• a distinct git author (zettelkasten-mcp) so its commits are distinguishable from memory-mcp and user commits, and
• no auto-push — local version history is enough for the zettelkasten; the memory subsystem's auto-push-safe rule is not replicated here.

The committer discovers the git repo from the scheduled file path (so it commits into whichever repo contains .zettelkasten/) and is a no-op when the path is not inside a git repo — it never raises into the caller.

ZettelCommitter

Batch .zettelkasten/ file commits behind a debounce timer.

Files handed to :meth:schedule are collected and committed together once the debounce window elapses without further activity. Commits are isolated (they add only the requested paths, never the user's other staged changes). Unlike the memory committer, this never pushes.

Source code in zettelkasten/commit.py

class ZettelCommitter:
    """Batch ``.zettelkasten/`` file commits behind a debounce timer.

    Files handed to :meth:`schedule` are collected and committed together once
    the debounce window elapses without further activity. Commits are isolated
    (they ``add`` only the requested paths, never the user's other staged
    changes). Unlike the memory committer, this never pushes.
    """

    def __init__(
        self,
        *,
        author: bytes = ZETTEL_AUTHOR,
        debounce_sec: float = DEFAULT_DEBOUNCE_SEC,
    ) -> None:
        self._author = author
        self._debounce_sec = debounce_sec
        self._lock = threading.Lock()
        self._pending: set[str] = set()
        self._timer: threading.Timer | None = None
        self._consecutive_failures: int = 0
        # Pre-compute the identity name/email and the pinned-identity, scrubbed
        # environment once so every commit uses the same zettelkasten-mcp author
        # regardless of the inherited environment.
        self._name, self._email = _parse_author_identity(author)
        self._git_env = _author_env(self._name, self._email)

    @staticmethod
    def _run_git(
        repo_root: str | Path,
        args: list[str],
        *,
        timeout: float = GIT_TIMEOUT_SEC,
        env: dict[str, str] | None = None,
    ) -> subprocess.CompletedProcess:
        """Run ``git -C <repo_root> <args>`` with no shell, capturing output.

        Paths are passed as argv items (never interpolated into a shell string),
        so spaces, unicode, and newlines in filenames are handled safely. ``env``
        defaults to a scrubbed copy of the environment (git-redirection vars
        removed) so an inherited ``GIT_DIR`` / ``GIT_WORK_TREE`` / ``GIT_INDEX_FILE``
        can't redirect the invocation.
        """
        return subprocess.run(
            ["git", "-C", str(repo_root), *args],
            capture_output=True,
            text=True,
            timeout=timeout,
            env=env if env is not None else _scrubbed_environ(),
        )

    @staticmethod
    def _discover_repo_root(filepath: str) -> str | None:
        """Return the work-tree root of the git repo containing ``filepath``.

        Uses ``git rev-parse --show-toplevel`` from the file's directory. A
        scheduled delete may have removed the file, so we walk up to the nearest
        existing ancestor before asking git. Returns ``None`` (never raises) when
        the path is not inside a git repo or git is unavailable, so the committer
        can no-op instead of raising into the caller.
        """
        try:
            start = Path(filepath).resolve()
            start_dir = start if start.is_dir() else start.parent
            while not start_dir.exists() and start_dir != start_dir.parent:
                start_dir = start_dir.parent
            result = ZettelCommitter._run_git(
                start_dir, ["rev-parse", "--show-toplevel"]
            )
            if result.returncode != 0:
                return None
            root = result.stdout.strip()
            return root or None
        except Exception:
            # NotAGitRepo / git missing / timeout → no-op (never raise).
            return None

    def schedule(self, filepath: str) -> None:
        """Add a file to the pending commit set and reset the debounce timer.

        A fresh schedule is a new authoring event, so it restores the auto-retry
        budget: without resetting ``_consecutive_failures`` here, once a batch
        hits :data:`MAX_COMMIT_RETRIES` the internal retry path stops re-arming
        the timer and every later commit attempt starts already over the cap,
        so transient failures would never be retried again. Resetting the
        counter re-arms a full retry budget for the newly scheduled file.
        """
        with self._lock:
            self._pending.add(filepath)
            self._consecutive_failures = 0
            self._arm_timer_locked()

    def _arm_timer_locked(self) -> None:
        """(Re)start the debounce timer. Caller must hold ``self._lock``."""
        if self._timer is not None:
            self._timer.cancel()
        self._timer = threading.Timer(self._debounce_sec, self._do_commit)
        self._timer.daemon = True
        self._timer.start()

    def _commit_files_locked(self, files: list[str]) -> int:
        """Commit the given paths, grouped by their containing repo.

        Caller must hold ``self._lock``. The commit is **path-scoped**: it
        commits ONLY the requested paths and never touches the user's other
        staged or unstaged working-tree state (see :meth:`_git_partial_commit`),
        so a separately ``git add``-ed unrelated file is never swept into a
        ``zettelkasten-mcp`` commit and remains staged afterwards. Files not
        inside any git repo are silently skipped (no-op). Returns the number of
        files committed. Raises on a git failure so the caller can decide
        whether to re-queue and retry.
        """
        if not files:
            return 0

        # Group by repo root so a path outside .zettelkasten's repo never
        # derails the batch (in practice all paths share one repo — the local
        # store). Deleted paths are kept in the batch so a removal is committed
        # too; the per-repo committer filters to paths that still exist or are
        # tracked-but-deleted.
        by_repo: dict[str, list[str]] = {}
        for f in files:
            root = self._discover_repo_root(f)
            if root is None:
                logger.debug("Skipping commit of file outside a git repo: %s", f)
                continue
            by_repo.setdefault(root, []).append(f)

        committed = 0
        for root, paths in by_repo.items():
            committed += self._git_partial_commit(root, paths)
        if committed:
            logger.info("Committed %d .zettelkasten/ file(s)", committed)
        return committed

    def _drop_ignored(self, repo_root: str | Path, relpaths: list[str]) -> list[str]:
        """Return ``relpaths`` minus paths git would refuse to stage as ignored.

        Uses ``git check-ignore`` (index-aware by default, so a *tracked* file
        that merely matches an ignore pattern is NOT dropped — only genuinely
        un-addable, untracked, ignored paths are). On any git error (e.g. not a
        repo) the list is returned unchanged so this can only ever drop a known
        ignored path, never lose a legitimate one.
        """
        if not relpaths:
            return relpaths
        # Feed paths via ``--stdin -z`` (NUL-delimited): ``git check-ignore -z``
        # is rejected with pathnames as argv ("-z only makes sense with
        # --stdin"), and NUL-delimited stdin is the only form that is safe for
        # paths containing spaces/newlines.
        try:
            check = subprocess.run(
                ["git", "-C", str(repo_root), "check-ignore", "-z", "--stdin"],
                input="\0".join(relpaths),
                capture_output=True,
                text=True,
                timeout=GIT_TIMEOUT_SEC,
                env=self._git_env,
            )
        except Exception:
            return relpaths
        # returncode: 0 = some paths ignored, 1 = none ignored, other = error.
        if check.returncode not in (0, 1):
            return relpaths
        ignored = {r for r in check.stdout.split("\0") if r}
        if not ignored:
            return relpaths
        return [r for r in relpaths if r not in ignored]

    def _git_partial_commit(self, repo_root: str, abs_paths: list[str]) -> int:
        """Commit only ``abs_paths`` into the repo at ``repo_root`` via git.

        Delegates index/HEAD/ref/tree handling to ``git`` itself — which
        implements it correctly and atomically — instead of hand-rolling it:

        1. Filter to paths that still exist OR are tracked-but-deleted, relative
           to the repo root.
        2. ``git add -A -- <relpaths>`` stages exactly those paths (``-A`` so a
           deletion is staged too; the ``-- <pathspec>`` scopes it so NO other
           file is staged).
        3. ``git commit -m <msg> --no-verify -- <relpaths>`` does a PARTIAL
           commit: it commits HEAD + only those paths' working-tree state,
           leaves the user's other staged files untouched and still staged, and
           keeps HEAD↔index consistent for the committed paths (the desync bug
           the hand-rolled path used to produce). The ``zettelkasten-mcp``
           identity is passed via per-invocation ``-c user.name``/``-c
           user.email`` so the repo's stored config is never mutated, and
           ``--no-verify`` skips user hooks (matching the memory committer, which
           commits via dulwich and likewise runs no hooks).

        A nothing-to-commit batch is a normal no-op: it returns 0 without
        raising. A real git failure raises so the caller's retry path re-queues.
        Returns the number of paths actually committed.
        """
        root = Path(repo_root).resolve()

        # 1. Compute repo-relative paths, keeping only those that exist or are
        #    tracked-but-deleted (a scheduled delete). An untracked, missing
        #    path would make `git add` error on an unmatched pathspec.
        existing: list[str] = []
        missing: list[str] = []
        for ap in abs_paths:
            p = Path(ap).resolve()
            try:
                rel = p.relative_to(root)
            except ValueError:
                continue  # not under this repo root (shouldn't happen)
            rel_str = str(rel).replace(os.sep, "/")
            (existing if p.exists() else missing).append(rel_str)

        relpaths = list(existing)
        if missing:
            ls = self._run_git(root, ["ls-files", "-z", "--", *missing])
            if ls.returncode == 0:
                tracked = {r for r in ls.stdout.split("\0") if r}
                relpaths.extend(r for r in missing if r in tracked)

        if not relpaths:
            return 0

        # 1b. Drop paths git would refuse to stage because they are ignored
        #     (e.g. a gitignored ``_embeddings.json`` under ``.zettelkasten/``).
        #     ``git add`` is all-or-nothing: a single ignored path in the
        #     pathspec makes the whole add fail, so co-scheduled legit files
        #     would never commit. Dropping ignored paths up front keeps the batch
        #     making progress on the addable ones, while a genuinely transient
        #     failure (index.lock) still surfaces below and re-queues via retry.
        relpaths = self._drop_ignored(root, relpaths)
        if not relpaths:
            return 0

        # 2. Stage exactly those paths (and only those).
        add = self._run_git(root, ["add", "-A", "--", *relpaths], env=self._git_env)
        if add.returncode != 0:
            raise RuntimeError(f"git add failed: {add.stderr.strip()}")

        # 4. Nothing-to-commit pre-check: if staging produced no change to the
        #    scoped paths, this batch is a normal no-op.
        status = self._run_git(
            root, ["status", "--porcelain", "-z", "--", *relpaths], env=self._git_env
        )
        if status.returncode != 0:
            raise RuntimeError(f"git status failed: {status.stderr.strip()}")
        changed = [r for r in status.stdout.split("\0") if r.strip()]
        if not changed:
            return 0
        n = len(changed)

        # 3. Partial commit of exactly those paths. The zettelkasten-mcp identity
        #    is pinned both via per-invocation ``-c user.name/email`` AND via the
        #    ``GIT_AUTHOR_*``/``GIT_COMMITTER_*`` env vars in ``self._git_env``
        #    (the env vars take precedence over -c, so an inherited GIT_AUTHOR_*
        #    can't override the distinct author).
        msg = f"[zettelkasten] update {n} file{'s' if n != 1 else ''}"
        commit = self._run_git(
            root,
            [
                "-c", f"user.name={self._name}",
                "-c", f"user.email={self._email}",
                "-c", "commit.gpgsign=false",
                "commit", "-m", msg, "--no-verify", "--", *relpaths,
            ],
            env=self._git_env,
        )
        if commit.returncode != 0:
            combined = (commit.stdout + commit.stderr).lower()
            if "nothing to commit" in combined or "no changes added" in combined:
                return 0
            raise RuntimeError(f"git commit failed: {commit.stderr.strip()}")
        return n

    def _do_commit(self) -> None:
        """Commit all pending files in a single batch.

        Holds the lock for the entire operation to serialize commits within this
        process.
        """
        with self._lock:
            self._timer = None
            if not self._pending:
                return
            files = list(self._pending)
            self._pending.clear()

            try:
                self._commit_files_locked(files)
                self._consecutive_failures = 0
            except Exception as e:
                # The commit failed (e.g. a transient index.lock / repack race).
                # Re-queue ALL scheduled files so they are retried, not lost; the
                # pending set was already cleared above. A scheduled path that no
                # longer exists on disk is a valid scheduled DELETE (the committer
                # commits the removal), so it must be re-queued too — filtering by
                # ``Path(f).exists()`` would silently drop a pending deletion on a
                # transient failure.
                self._pending.update(files)
                self._consecutive_failures += 1
                logger.warning(
                    "Zettelkasten batch commit failed (attempt %d); re-queued file(s): %s",
                    self._consecutive_failures,
                    e,
                )
                if self._consecutive_failures <= MAX_COMMIT_RETRIES:
                    self._arm_timer_locked()

    def commit_now(self, paths: list[str]) -> int:
        """Synchronously commit the given paths, bypassing the debounce queue.

        Returns the number of files committed; never raises. A transient git
        failure (e.g. ``index.lock`` contention with a concurrent git process)
        is retried a bounded number of times with a short backoff rather than
        silently swallowed, so a momentary lock race doesn't lose the commit.
        The retry budget is small and capped (:data:`MAX_COMMIT_RETRIES`) so the
        synchronous caller is never blocked for long.
        """
        with self._lock:
            last_err: Exception | None = None
            for attempt in range(1, MAX_COMMIT_RETRIES + 1):
                try:
                    return self._commit_files_locked(list(paths))
                except Exception as e:
                    last_err = e
                    if attempt < MAX_COMMIT_RETRIES:
                        time.sleep(COMMIT_RETRY_BACKOFF_SEC * attempt)
            logger.warning(
                "zettelkasten commit_now failed after %d attempt(s): %s",
                MAX_COMMIT_RETRIES,
                last_err,
            )
            return 0

    def flush(self) -> None:
        """Flush any pending debounced commit immediately (e.g. on shutdown)."""
        with self._lock:
            if self._timer is not None:
                self._timer.cancel()
                self._timer = None
            if not self._pending:
                return
        self._do_commit()

schedule

schedule(filepath: str) -> None

Add a file to the pending commit set and reset the debounce timer.

A fresh schedule is a new authoring event, so it restores the auto-retry budget: without resetting _consecutive_failures here, once a batch hits :data:MAX_COMMIT_RETRIES the internal retry path stops re-arming the timer and every later commit attempt starts already over the cap, so transient failures would never be retried again. Resetting the counter re-arms a full retry budget for the newly scheduled file.

Source code in zettelkasten/commit.py

def schedule(self, filepath: str) -> None:
    """Add a file to the pending commit set and reset the debounce timer.

    A fresh schedule is a new authoring event, so it restores the auto-retry
    budget: without resetting ``_consecutive_failures`` here, once a batch
    hits :data:`MAX_COMMIT_RETRIES` the internal retry path stops re-arming
    the timer and every later commit attempt starts already over the cap,
    so transient failures would never be retried again. Resetting the
    counter re-arms a full retry budget for the newly scheduled file.
    """
    with self._lock:
        self._pending.add(filepath)
        self._consecutive_failures = 0
        self._arm_timer_locked()

commit_now

commit_now(paths: list[str]) -> int

Synchronously commit the given paths, bypassing the debounce queue.

Returns the number of files committed; never raises. A transient git failure (e.g. index.lock contention with a concurrent git process) is retried a bounded number of times with a short backoff rather than silently swallowed, so a momentary lock race doesn't lose the commit. The retry budget is small and capped (:data:MAX_COMMIT_RETRIES) so the synchronous caller is never blocked for long.

Source code in zettelkasten/commit.py

def commit_now(self, paths: list[str]) -> int:
    """Synchronously commit the given paths, bypassing the debounce queue.

    Returns the number of files committed; never raises. A transient git
    failure (e.g. ``index.lock`` contention with a concurrent git process)
    is retried a bounded number of times with a short backoff rather than
    silently swallowed, so a momentary lock race doesn't lose the commit.
    The retry budget is small and capped (:data:`MAX_COMMIT_RETRIES`) so the
    synchronous caller is never blocked for long.
    """
    with self._lock:
        last_err: Exception | None = None
        for attempt in range(1, MAX_COMMIT_RETRIES + 1):
            try:
                return self._commit_files_locked(list(paths))
            except Exception as e:
                last_err = e
                if attempt < MAX_COMMIT_RETRIES:
                    time.sleep(COMMIT_RETRY_BACKOFF_SEC * attempt)
        logger.warning(
            "zettelkasten commit_now failed after %d attempt(s): %s",
            MAX_COMMIT_RETRIES,
            last_err,
        )
        return 0

flush

flush() -> None

Flush any pending debounced commit immediately (e.g. on shutdown).

Source code in zettelkasten/commit.py

def flush(self) -> None:
    """Flush any pending debounced commit immediately (e.g. on shutdown)."""
    with self._lock:
        if self._timer is not None:
            self._timer.cancel()
            self._timer = None
        if not self._pending:
            return
    self._do_commit()

review_lock

review_lock(name: str) -> Lock

Return the in-process write lock for review name (lazily created).

This is layer 1 of :func:review_write_lock: it serializes threads WITHIN a single process. It provides no cross-process protection on its own — for the full guarantee (threads + separate OS processes) use :func:review_write_lock, which acquires this lock and then an OS advisory file lock.

Source code in zettelkasten/commit.py

def review_lock(name: str) -> threading.Lock:
    """Return the in-process write lock for review ``name`` (lazily created).

    This is layer 1 of :func:`review_write_lock`: it serializes threads WITHIN a
    single process. It provides no cross-process protection on its own — for the
    full guarantee (threads + separate OS processes) use
    :func:`review_write_lock`, which acquires this lock and then an OS advisory
    file lock.
    """
    with _review_locks_guard:
        lock = _review_locks.get(name)
        if lock is None:
            lock = threading.Lock()
            _review_locks[name] = lock
        return lock

review_write_lock

review_write_lock(name: str, *, graphs_dir: Path | str | None = None, timeout: float = 10.0) -> Iterator[None]

Two-layer (thread + cross-process) write lock for review name.

Acquires the in-process :func:review_lock first (serializes threads within this process), then an OS advisory file lock on a per-machine lockfile (serializes SEPARATE OS processes — the MCP server and dashboard backend). Both layers are released on exit, in reverse order, even on error.

graphs_dir is the resolved store base the caller writes to (it flows through to :func:_review_lock_path); pass the SAME base used to resolve the _reviews/<name>.yaml write so both processes derive one shared lock. Defaults to GRAPHS_DIR when omitted.

Raises :class:TimeoutError if EITHER layer cannot be acquired within timeout seconds — the in-process layer is acquired with the same timeout and, on failure, nothing is left held (the OS layer is never reached). On platforms without an advisory-lock primitive the OS layer degrades to a best-effort no-op while the in-process layer still holds.

Usage::

with review_write_lock(name, graphs_dir=base):
    data = load_review(name)
    ...  # mutate
    write_yaml_atomic(path, data)

Source code in zettelkasten/commit.py

@contextmanager
def review_write_lock(
    name: str, *, graphs_dir: Path | str | None = None, timeout: float = 10.0
) -> Iterator[None]:
    """Two-layer (thread + cross-process) write lock for review ``name``.

    Acquires the in-process :func:`review_lock` first (serializes threads within
    this process), then an OS advisory file lock on a per-machine lockfile
    (serializes SEPARATE OS processes — the MCP server and dashboard backend).
    Both layers are released on exit, in reverse order, even on error.

    ``graphs_dir`` is the resolved store base the caller writes to (it flows
    through to :func:`_review_lock_path`); pass the SAME base used to resolve the
    ``_reviews/<name>.yaml`` write so both processes derive one shared lock.
    Defaults to ``GRAPHS_DIR`` when omitted.

    Raises :class:`TimeoutError` if EITHER layer cannot be acquired within
    ``timeout`` seconds — the in-process layer is acquired with the same timeout
    and, on failure, nothing is left held (the OS layer is never reached). On
    platforms without an advisory-lock primitive the OS layer degrades to a
    best-effort no-op while the in-process layer still holds.

    Usage::

        with review_write_lock(name, graphs_dir=base):
            data = load_review(name)
            ...  # mutate
            write_yaml_atomic(path, data)
    """
    in_proc = review_lock(name)
    if not in_proc.acquire(timeout=timeout):
        raise TimeoutError(
            f"Could not acquire in-process review write lock for '{name}' "
            f"within {timeout}s"
        )
    fd: int | None = None
    try:
        lock_path = _review_lock_path(name, graphs_dir=graphs_dir)
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        fd = os.open(str(lock_path), os.O_RDWR | os.O_CREAT, 0o600)
        _acquire_os_lock(fd, timeout=timeout, name=name)
        try:
            yield
        finally:
            _release_os_lock(fd)
    finally:
        if fd is not None:
            try:
                os.close(fd)
            except OSError:
                pass
        in_proc.release()

box_write_lock

box_write_lock(graph_name: str, *, graphs_dir: Path | str | None = None, timeout: float = 10.0) -> Iterator[None]

Two-layer (thread + cross-process) write lock for a whole graph box.

Holding this lock serializes every write to graph_name across threads and separate OS processes (the MCP server and the dashboard backend). It is built on :func:review_write_lock with a box:: key namespace so a box named x never collides with a review/table/outline also named x.

graphs_dir is the resolved store base; pass the SAME base used to resolve the box's writes so all processes derive one shared lock. Raises :class:TimeoutError if the lock cannot be acquired within timeout.

Source code in zettelkasten/commit.py

@contextmanager
def box_write_lock(
    graph_name: str, *, graphs_dir: Path | str | None = None, timeout: float = 10.0
) -> Iterator[None]:
    """Two-layer (thread + cross-process) write lock for a whole graph box.

    Holding this lock serializes every write to ``graph_name`` across threads
    and separate OS processes (the MCP server and the dashboard backend). It is
    built on :func:`review_write_lock` with a ``box::`` key namespace so a box
    named ``x`` never collides with a review/table/outline also named ``x``.

    ``graphs_dir`` is the resolved store base; pass the SAME base used to resolve
    the box's writes so all processes derive one shared lock. Raises
    :class:`TimeoutError` if the lock cannot be acquired within ``timeout``.
    """
    with review_write_lock(f"box::{graph_name}", graphs_dir=graphs_dir, timeout=timeout):
        yield

boxes_write_lock

boxes_write_lock(graph_names: Iterable[str], *, graphs_dir: Path | str | None = None, timeout: float = 10.0) -> Iterator[None]

Acquire :func:box_write_lock for several boxes at once, deadlock-free.

Box names are de-duplicated, empty names dropped, and the rest acquired in SORTED order. Consistent ordering is what prevents an ABBA deadlock when two callers lock the same pair of boxes (e.g. a cross-graph link that touches a source box and _cross) from opposite directions.

Source code in zettelkasten/commit.py

@contextmanager
def boxes_write_lock(
    graph_names: Iterable[str],
    *,
    graphs_dir: Path | str | None = None,
    timeout: float = 10.0,
) -> Iterator[None]:
    """Acquire :func:`box_write_lock` for several boxes at once, deadlock-free.

    Box names are de-duplicated, empty names dropped, and the rest acquired in
    SORTED order. Consistent ordering is what prevents an ABBA deadlock when two
    callers lock the same pair of boxes (e.g. a cross-graph link that touches a
    source box and ``_cross``) from opposite directions.
    """
    from contextlib import ExitStack

    unique = sorted({g for g in graph_names if g})
    with ExitStack() as stack:
        for g in unique:
            stack.enter_context(
                box_write_lock(g, graphs_dir=graphs_dir, timeout=timeout)
            )
        yield