import os
import re
import subprocess
import sys
import tomllib
OURS = re.compile(r"(makenot\.work|git\.sr\.ht/~maxmj)", re.I)
SKIP_DIRS = {
"target", ".git", "node_modules", "dist", "vendor",
"_archive", "_scratch", "trash", "_meta", "vtebench",
}
MAX_DEPTH = 4
DEP_SECTIONS = ("dependencies", "dev-dependencies", "build-dependencies")
def manifests(root):
out = []
stack = [(root, 0)]
while stack:
d, depth = stack.pop()
try:
entries = list(os.scandir(d))
except OSError:
continue
for e in entries:
if e.is_file() and e.name == "Cargo.toml":
out.append(e.path)
elif e.is_dir() and e.name not in SKIP_DIRS and depth < MAX_DEPTH:
stack.append((e.path, depth + 1))
return out
def load(path):
try:
with open(path, "rb") as fh:
return tomllib.load(fh)
except (OSError, tomllib.TOMLDecodeError):
return None
def dep_tables(doc):
for section in DEP_SECTIONS:
table = doc.get(section)
if isinstance(table, dict):
yield table
for cfg in (doc.get("target") or {}).values():
if not isinstance(cfg, dict):
continue
for section in DEP_SECTIONS:
table = cfg.get(section)
if isinstance(table, dict):
yield table
ws = doc.get("workspace") or {}
table = ws.get("dependencies")
if isinstance(table, dict):
yield table
def parse_version(v):
core = str(v).split("+")[0].split("-")[0]
parts = []
for piece in core.split(".")[:3]:
try:
parts.append(int(piece))
except ValueError:
parts.append(0)
while len(parts) < 3:
parts.append(0)
return tuple(parts)
def satisfies(req, version):
req = req.strip()
if not req or any(c in req for c in "<>*,~"):
return None
if "-" in str(version).split("+")[0] and "-" not in req:
return False
exact = req.startswith("=")
req = req.lstrip("^=").strip()
if not req:
return None
given = req.split(".")
try:
r = [int(p) for p in given[:3]]
except ValueError:
return None
v = parse_version(version)
if exact:
return tuple(v[: len(r)]) == tuple(r)
if r[0] > 0:
return v[0] == r[0] and v[1:] >= tuple(r[1:] + [0] * (2 - len(r[1:])))
if len(r) == 1:
return v[0] == 0
if r[1] > 0:
return v[0] == 0 and v[1] == r[1] and v[2] >= (r[2] if len(r) > 2 else 0)
if len(r) > 2:
return v[:3] == (0, 0, r[2])
return v[0] == 0 and v[1] == 0
def git_lines(repo, *args):
try:
out = subprocess.run(
["git", "-C", repo, *args],
capture_output=True, text=True, check=True,
)
except (OSError, subprocess.CalledProcessError):
return None
return out.stdout.splitlines()
def git_manifests(repo, sha):
lines = git_lines(repo, "ls-tree", "-r", "--name-only", sha)
if lines is None:
return None
out = []
for rel in lines:
if os.path.basename(rel) != "Cargo.toml":
continue
if any(part in SKIP_DIRS for part in rel.split("/")):
continue
out.append(rel)
return out
def load_at(repo, sha, rel):
lines = git_lines(repo, "show", f"{sha}:{rel}")
if lines is None:
return None
try:
return tomllib.loads("\n".join(lines))
except tomllib.TOMLDecodeError:
return None
def pushed_view(docs, repo, sha):
rels = git_manifests(repo, sha)
if rels is None:
return None
out = {k: v for k, v in docs.items() if not k.startswith(repo + os.sep)}
for rel in rels:
out[os.path.join(repo, rel)] = load_at(repo, sha, rel)
return out
def repo_of(path, tree):
d = os.path.dirname(path)
while d.startswith(tree):
if os.path.exists(os.path.join(d, ".git")):
return d
if d == tree:
break
parent = os.path.dirname(d)
if parent == d:
break
d = parent
return None
def publishing_ref(repo, cache):
if repo in cache:
return cache[repo]
ref = None
lines = git_lines(repo, "remote", "-v") or []
urls = {}
for line in lines:
parts = line.split()
if len(parts) >= 2:
urls.setdefault(parts[0], parts[1])
order = [r for r in ("mnw",) if r in urls]
order += [r for r, u in urls.items() if r not in order and OURS.search(u)]
order += [r for r in ("origin",) if r in urls and r not in order]
for remote in order:
head = git_lines(repo, "symbolic-ref", "--quiet", f"refs/remotes/{remote}/HEAD")
candidates = []
if head:
candidates.append(head[0].rsplit("/", 1)[-1])
candidates += ["main", "master"]
for branch in candidates:
if git_lines(repo, "rev-parse", "--verify", "--quiet",
f"refs/remotes/{remote}/{branch}"):
ref = f"{remote}/{branch}"
break
if ref:
break
cache[repo] = ref
return ref
def version_at(repo, ref, rel, cache):
key = (repo, ref, rel)
if key in cache:
return cache[key]
version = None
doc = load_at(repo, ref, rel)
if doc:
pkg = doc.get("package")
if isinstance(pkg, dict):
v = pkg.get("version")
if isinstance(v, str):
version = v
elif isinstance(v, dict) and v.get("workspace") is True:
d = os.path.dirname(rel)
while True:
root_rel = os.path.join(d, "Cargo.toml") if d else "Cargo.toml"
root = load_at(repo, ref, root_rel) if root_rel != rel else None
inherited = (
((root or {}).get("workspace") or {}).get("package") or {}
).get("version")
if isinstance(inherited, str):
version = inherited
break
if not d:
break
d = os.path.dirname(d)
cache[key] = version
return version
def analyze_published(docs, disk_docs, tree, repo, sha):
ref_cache, version_cache, fetched = {}, {}, set()
versions_on_disk = crate_index(disk_docs, tree)
broken, graded, ungraded = [], 0, {}
for consumer_manifest, name, req in requirements(docs):
known = versions_on_disk.get(name)
if known is None:
continue provider_manifest = known[1]
provider_repo = repo_of(provider_manifest, tree)
if provider_repo is None:
ungraded.setdefault(os.path.dirname(provider_manifest), "not a git repo")
continue
if repo is not None and provider_repo == repo and sha:
continue
ref = publishing_ref(provider_repo, ref_cache)
if ref is None:
ungraded.setdefault(provider_repo, "no fetched remote to read")
continue
rel = os.path.relpath(provider_manifest, provider_repo)
have = version_at(provider_repo, ref, rel, version_cache)
if have is None:
ungraded.setdefault(provider_repo, f"{name} is not at {ref} yet")
continue
verdict = satisfies(req, have)
if verdict is None:
continue
if not verdict and provider_repo not in fetched:
fetched.add(provider_repo)
remote, branch = ref.split("/", 1)
git_lines(provider_repo, "fetch", "--quiet", remote, branch)
version_cache.pop((provider_repo, ref, rel), None)
have = version_at(provider_repo, ref, rel, version_cache) or have
verdict = satisfies(req, have)
graded += 1
if not verdict:
broken.append(
(consumer_manifest, name, req, have, provider_manifest, ref)
)
return broken, graded, ungraded
def crate_index(docs, tree):
ws_version = {}
for p, doc in docs.items():
if not doc:
continue
v = ((doc.get("workspace") or {}).get("package") or {}).get("version")
if isinstance(v, str):
ws_version[os.path.dirname(p)] = v
def resolve_version(manifest_path, pkg):
v = pkg.get("version")
if isinstance(v, str):
return v
d = os.path.dirname(manifest_path)
while d.startswith(tree):
if d in ws_version:
return ws_version[d]
parent = os.path.dirname(d)
if parent == d:
break
d = parent
return None
versions = {}
for p, doc in docs.items():
if not doc:
continue
pkg = doc.get("package")
if not isinstance(pkg, dict) or not isinstance(pkg.get("name"), str):
continue
v = resolve_version(p, pkg)
if v:
versions[pkg["name"]] = (v, p)
return versions
def requirements(docs):
for p, doc in docs.items():
if not doc:
continue
for table in dep_tables(doc):
for key, spec in table.items():
if not isinstance(spec, dict):
continue
git = spec.get("git")
req = spec.get("version")
if not isinstance(git, str) or not isinstance(req, str):
continue
if not OURS.search(git):
continue
name = spec.get("package") if isinstance(spec.get("package"), str) else key
yield p, name, req
def analyze(docs, tree):
versions = crate_index(docs, tree)
broken, unchecked, absent, graded = [], 0, set(), 0
for p, name, req in requirements(docs):
known = versions.get(name)
if known is None:
absent.add(name)
continue
verdict = satisfies(req, known[0])
if verdict is None:
unchecked += 1
continue
graded += 1
if not verdict:
broken.append((p, name, req, known[0], known[1], None))
return broken, unchecked, absent, graded
def split_blame(broken, repo):
ours, theirs = [], []
for item in broken:
consumer_manifest, _name, _req, _have, provider_manifest, _src = item
mine = repo is not None and (
consumer_manifest.startswith(repo + os.sep)
or provider_manifest.startswith(repo + os.sep)
)
(ours if mine else theirs).append(item)
return ours, theirs
def report(broken, repo, tree, label):
ours, theirs = split_blame(broken, repo)
def rel(path):
return os.path.relpath(path, tree)
for consumer_manifest, name, req, have, provider_manifest, src in ours + theirs:
where = (
f"{src} has {have} ({rel(provider_manifest)})"
if src
else f"the tree has {have} ({rel(provider_manifest)})"
)
print(
f" [{label}] {rel(consumer_manifest)}: requires {name} \"{req}\", {where}",
file=sys.stderr,
)
if repo is None:
return bool(broken)
if not ours:
if theirs:
print(
f"pre-push: [{label}] {len(theirs)} unresolvable requirements "
"elsewhere in the tree (listed above, not this push's).",
)
return False
return True
def main():
if len(sys.argv) < 2:
print(__doc__.strip(), file=sys.stderr)
return 2
tree = os.path.realpath(sys.argv[1])
repo = os.path.realpath(sys.argv[2]) if len(sys.argv) > 2 else None
sha = sys.argv[3] if len(sys.argv) > 3 else None
docs = {p: load(p) for p in manifests(tree)}
views = [("working copy", docs)]
skipped_push_view = False
if repo and sha:
pushed = pushed_view(docs, repo, sha)
if pushed is None:
skipped_push_view = True
else:
views.append(("as pushed", pushed))
refuse = False
summaries = []
for label, view in views:
broken, unchecked, absent, graded = analyze(view, tree)
summaries.append((label, graded, unchecked, absent, bool(broken)))
if broken and report(broken, repo, tree, label):
refuse = True
consumer_view = views[-1][1]
pub_broken, pub_graded, ungraded = analyze_published(
consumer_view, docs, tree, repo, sha
)
summaries.append(("as published", pub_graded, 0, set(), bool(pub_broken)))
if pub_broken and report(pub_broken, repo, tree, "as published"):
refuse = True
if refuse:
bad = {lbl for lbl, _g, _u, _a, broke in summaries if broke}
clean = [lbl for lbl, _g, _u, _a, broke in summaries if not broke]
print("", file=sys.stderr)
print(
"pre-push: this push leaves a dependency that cannot resolve.\n"
" A version requirement states which major a consumer was written against,\n"
" so bumping a library and fixing its consumers is one pass (CLAUDE.md,\n"
" \"a breaking bump of an in-house crate is forward-fixed, in the same pass\").\n"
" Fix: bump the requirement in the manifests above, make the consumers\n"
" compile, and push them with this one.",
file=sys.stderr,
)
if "as published" in bad and "working copy" not in bad:
print(
" The working copy is fine and the published tree is not, so the\n"
" difference is what has been PUSHED: a git dependency resolves against\n"
" the branch head at the URL, and ~/Code/.cargo/config.toml's [patch]\n"
" block hides that locally. Push the sibling named above first.",
file=sys.stderr,
)
elif "as pushed" in bad and "working copy" not in bad:
print(
" The working copy is fine and the commit is not, so the difference is\n"
" what is COMMITTED. An uncommitted manifest edit is the usual cause.",
file=sys.stderr,
)
elif clean:
print(
f" Note: the {clean[0]} view is clean, so the views disagree; the one\n"
" that failed is named on each line above.",
file=sys.stderr,
)
return 1
for label, graded, unchecked, absent, _bad in summaries:
print(
f"pre-push: internal deps coherent [{label}] ({graded} requirements"
+ (f", {unchecked} unchecked" if unchecked else "")
+ (f", {len(absent)} crates not in this tree" if absent else "")
+ ")."
)
for where, why in sorted(ungraded.items()):
print(
f"pre-push: [as published] {os.path.relpath(where, tree)} not graded ({why})."
)
if skipped_push_view:
print("pre-push: could not read the pushed commit; graded the working copy only.")
return 0
if __name__ == "__main__":
sys.exit(main())