class MemoryCommitter:
"""Batch memory-authored 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
(a path-scoped ``git commit -- <paths>`` stages and commits only the
requested paths, never the user's other staged changes) and auto-push only
fires when every unpushed commit is memory-authored — user WIP is never
pushed.
"""
def __init__(
self,
*,
author: bytes = MEMORY_AUTHOR,
debounce_sec: float = DEFAULT_DEBOUNCE_SEC,
auto_push: bool = True,
) -> None:
self._author = author
self._debounce_sec = debounce_sec
self._auto_push = auto_push
self._lock = threading.Lock()
self._pending: set[str] = set()
self._timer: threading.Timer | None = None
self._repo: DulwichRepo | 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 memory-mcp author
# regardless of the inherited environment.
self._name, self._email = _parse_author_identity(author)
self._git_env = _author_env(self._name, self._email)
def _get_repo(self) -> DulwichRepo | None:
"""Get or open the dulwich repo handle (cached).
Used only for the auto-push safety check; the commit itself goes through
subprocess git. A re-read of refs from disk picks up the commits the git
subprocess wrote, so the cached handle stays valid across commits.
"""
if self._repo is None:
try:
self._repo = DulwichRepo(".")
except Exception:
return None
return self._repo
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()
@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 (every git-redirection var
in :data:`_GIT_REDIRECT_VARS` removed) so an inherited ``GIT_DIR`` /
``GIT_WORK_TREE`` / ``GIT_INDEX_FILE`` / ``GIT_COMMON_DIR`` / object-store
/ namespace var 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 = MemoryCommitter._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 _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). ``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 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 the dulwich
snapshot-the-whole-index path produced). The ``memory-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 prior dulwich path, which ran no hooks either).
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, so
# a single ignored path can't poison the whole add for the batch.
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()}")
# 2b. 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 memory-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"[memory] 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 _commit_files_locked(self, files: list[str]) -> int:
"""Commit the given paths in path-scoped commits, grouped by repo.
Caller must hold ``self._lock``. The commit is **path-scoped** (see
:meth:`_git_partial_commit`): it commits ONLY the requested paths and
never touches the user's other staged or unstaged working-tree state, so
a separately ``git add``-ed unrelated file is never swept into a
``memory-mcp`` commit and remains staged afterwards. Secrets are dropped
before staging, and files not inside any git repo are silently skipped.
Returns the number of files committed (0 if nothing committable
remained). Raises on a git failure so the caller can decide whether to
re-queue and retry.
"""
# Never auto-commit secrets, even if a caller scheduled them.
sensitive = [f for f in files if is_sensitive_path(f)]
if sensitive:
logger.warning(
"Refusing to auto-commit sensitive file(s): %s",
", ".join(sorted(sensitive)),
)
candidates = [f for f in files if not is_sensitive_path(f)]
if not candidates:
return 0
# Group by repo root so a path outside the .memory/ repo never derails
# the batch (in practice all paths share one repo). Deleted paths are
# kept 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 candidates:
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 .memory/ file(s)", committed)
return committed
def _do_commit(self) -> None:
"""Commit all pending files in a single batch commit.
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()
repo = self._get_repo()
if repo is None:
# Couldn't open the repo (transient). Keep the files so a later
# schedule()/flush() retries rather than silently dropping them.
self._pending.update(files)
return
try:
committed = 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 and leave a stale git entry.
self._pending.update(files)
self._consecutive_failures += 1
logger.warning(
"Memory 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()
return
if committed and self._auto_push:
self._safe_auto_push(repo)
def commit_now(self, paths: list[str]) -> int:
"""Synchronously commit the given memory paths, then safe auto-push.
Unlike :meth:`schedule`/:meth:`flush`, this bypasses the debounce queue
and commits exactly the paths handed in. It is intended for the
git-hook / CLI path so a user's code commit also flushes any pending
``.memory/`` files that a separate (MCP server) process hasn't committed
yet. Returns the number of files committed.
"""
repo = self._get_repo()
if repo is None:
return 0
with self._lock:
try:
committed = self._commit_files_locked(list(paths))
except Exception as e:
logger.warning("memory commit_now failed: %s", e)
return 0
if committed and self._auto_push:
self._safe_auto_push(repo)
return committed
def _get_remote_url(self, repo: DulwichRepo, name: str = "origin") -> str | None:
"""Read a remote URL from the repo's git config."""
try:
config = repo.get_config()
url = config.get((b"remote", name.encode()), b"url")
return url.decode() if url else None
except (KeyError, Exception):
return None
def _safe_auto_push(self, repo: DulwichRepo) -> None:
"""Push only if all unpushed commits are authored by memory-mcp.
Never pushes user WIP commits. Uses local refs only (no network calls
for the safety check).
"""
try:
remote_url = self._get_remote_url(repo)
if remote_url is None:
return
try:
symrefs = repo.refs.get_symrefs()
head_ref = symrefs.get(b"HEAD")
if not head_ref or not head_ref.startswith(b"refs/heads/"):
return
branch_name = head_ref[len(b"refs/heads/"):]
except Exception:
return
local_head = repo.head()
remote_ref_name = b"refs/remotes/origin/" + branch_name
# dulwich's DiskRefsContainer has no ``.get`` method (the previous
# ``repo.refs.get(...)`` raised AttributeError, swallowed by the
# broad except → memory never auto-pushed). Read the ref the
# idiomatic dulwich way; a missing remote ref means there is nothing
# to compare against, so there is nothing to push → return.
try:
remote_head_ref = repo.refs[remote_ref_name]
except KeyError:
return
if local_head == remote_head_ref:
return
walker = repo.get_walker(include=[local_head], exclude=[remote_head_ref])
for walk_entry in walker:
commit = walk_entry.commit
if commit.author != self._author:
logger.info("Skipping auto-push: found non-memory commit %s", commit.id.decode()[:7])
return
refspec = head_ref + b":" + head_ref
porcelain.push(repo, remote_url, refspecs=[refspec])
logger.info("Auto-pushed memory commits to %s", branch_name.decode())
except Exception as e:
logger.debug("Auto-push skipped: %s", e)
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()