#!/usr/bin/env python3
"""landing_check.py -- is every file at this revision actually served by the forge?
WHY THIS EXISTS
On 2026-10-17 this lane committed `tools/census_row_delta.py` and did not push
it. The commit was real, `git log` showed it, the check-in quoted it -- and the
archive could not read it: the anonymous raw path returned HTTP 404 / 11 B.
A committed artifact the archive cannot read is not a landed artifact, and
nothing in this repo compared "bytes at this revision" against "bytes the
forge serves". The mismatch was found by hand, twice, in one night. This is
the gate for that sentence.
It is deliberately narrow: it does not grade whether the work is any good, it
does not read the queue, and it does not decide what should be committed. It
answers one question per tracked file, with the bytes on both sides.
WHAT IT MEASURES -- one row per tracked file at --rev, one verdict per row
OK the blob at `rev:path` and the bytes served at
`/branch//` have the same length and the
same sha256.
DIFFERS both sides returned bytes and they are not identical. This is
the unpushed-commit class: the local revision left the forge
behind, or the branch moved under the revision being checked.
MISSING the transport answered 404 (or, for file://, no such file).
A tracked path the archive cannot read at all.
HTTP_ERROR any other non-200 status. Not a verdict about content.
UNREACHABLE the transport failed before a status existed (DNS, refused,
TLS, timeout, unreadable fixture root). Distinct from MISSING
on purpose: "the forge did not answer" is not "the file is not
there", and collapsing them would let a network outage read as
a clean negative.
READ_FAILED `git cat-file blob rev:path` could not produce the local bytes.
An archive-side inconsistency, not a forge one.
SKIPPED_LARGE the local blob exceeds --max-bytes; not fetched, not graded.
ADVISORY, never fatal: a size policy is not a landing defect.
FATAL = DIFFERS, MISSING, HTTP_ERROR, UNREACHABLE, READ_FAILED. Exit 1 if any
row is fatal, else 0. Usage/precondition errors exit 2. A run in which every
fetched file matches prints `landing: N files, N OK, ...` and exits 0.
WHAT IT DOES NOT CLAIM
Byte equality is not provenance: a forge serving an identical copy says the
archive can read the same bytes, not that the bytes are correct, reviewed, or
reachable at another URL. It also cannot see a file that no revision tracks --
the census blindness this repo already paid for -- so untracked working-tree
files are out of scope by construction, and said so here rather than dropped.
The raw path is a forge convention (`raw/branch//`); --base-url
exists so the convention is a parameter, not an assumption baked into a URL.
READ-ONLY: the only write path is `--json PATH`. No file, branch or remote is
modified; every fetch is anonymous and GET.
USAGE:
python3 landing_check.py --repo /root/troupe-standup
python3 landing_check.py --repo . --rev HEAD --path tools/selftest_sweep.py
python3 landing_check.py --repo . --json /tmp/landing.json --quiet
python3 landing_check.py --selftest # fixture forge, no network, 26 cells
"""
import argparse
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from collections import Counter
DEFAULT_BASE = ("https://forge.alexanderwhitestone.com/git/stackchain/"
"troupe-standup/raw")
CLAIM_RE = re.compile(r"^[0-9a-f]{7,40}$")
FATAL = ("DIFFERS", "MISSING", "HTTP_ERROR", "UNREACHABLE", "READ_FAILED")
ADVISORY = ("SKIPPED_LARGE",)
class FetchError(Exception):
"""The transport failed before any status existed."""
def sha256(data):
return hashlib.sha256(data).hexdigest()
def git(repo, *args):
cp = subprocess.run(["git", "-C", repo, *args], capture_output=True)
if cp.returncode != 0:
raise RuntimeError("git %s failed: %s" % (
" ".join(args), cp.stderr.decode("utf-8", "replace").strip()))
return cp.stdout
def tracked_at(repo, rev):
"""Every path the revision tracks, NUL-separated so spaces are safe."""
out = git(repo, "ls-tree", "-r", "--name-only", "-z", rev)
return sorted(p for p in out.decode("utf-8", "surrogateescape").split("\0") if p)
def blob_at(repo, rev, path):
"""The exact bytes of rev:path. Binary-safe: no decode anywhere."""
return git(repo, "cat-file", "blob", "%s:%s" % (rev, path))
class Forge:
"""Anonymous GET of the raw path, or a file:// tree standing in for it."""
def __init__(self, base_url, timeout=20):
self.base = base_url.rstrip("/")
self.timeout = timeout
def url(self, branch, path):
from urllib.parse import quote
return "%s/branch/%s/%s" % (self.base, quote(branch, safe=""),
quote(path, safe="/"))
def fetch(self, branch, path):
"""-> (bytes, status). Raises FetchError when no status was reached."""
url = self.url(branch, path)
if url.startswith("file://"):
from urllib.parse import unquote
local = unquote(url[len("file://"):])
branch_root = self.base[len("file://"):].rstrip("/") + "/branch/" + branch
if not os.path.isdir(branch_root):
raise FetchError("fixture root absent: %s" % branch_root)
if os.path.isdir(local):
return b"", 500 # a directory is not a file
try:
with open(local, "rb") as fh:
return fh.read(), 200
except FileNotFoundError:
return b"", 404
except OSError as exc:
raise FetchError(str(exc))
import urllib.error
import urllib.request
try:
with urllib.request.urlopen(url, timeout=self.timeout) as resp:
return resp.read(), resp.status
except urllib.error.HTTPError as exc:
return b"", exc.code
except (urllib.error.URLError, OSError) as exc:
raise FetchError(str(exc))
def grade(local, remote, status):
"""One verdict from the two byte strings and the transport status."""
if status == 404:
return "MISSING", "served 404"
if status != 200:
return "HTTP_ERROR", "served HTTP %s" % status
if len(local) != len(remote):
return "DIFFERS", "length %d local vs %d served" % (len(local), len(remote))
if sha256(local) != sha256(remote):
return "DIFFERS", "same length, different sha256"
return "OK", "%d B, sha256 %s" % (len(local), sha256(local)[:16])
def build_rows(repo, rev, branch, forge, paths=None, max_bytes=4 * 1024 * 1024):
files = tracked_at(repo, rev)
if paths:
wanted = set(paths)
files = [p for p in files if p in wanted]
rows = []
for path in files:
row = {"path": path, "verdict": None, "detail": "",
"local_bytes": None, "remote_bytes": None,
"local_sha256": None, "remote_sha256": None, "status": None}
try:
local = blob_at(repo, rev, path)
except RuntimeError as exc:
row.update(verdict="READ_FAILED", detail=str(exc)[:120])
rows.append(row)
continue
row["local_bytes"] = len(local)
row["local_sha256"] = sha256(local)
if len(local) > max_bytes:
row.update(verdict="SKIPPED_LARGE",
detail="%d B over --max-bytes %d" % (len(local), max_bytes))
rows.append(row)
continue
try:
remote, status = forge.fetch(branch, path)
except FetchError as exc:
row.update(verdict="UNREACHABLE", detail=str(exc)[:120])
rows.append(row)
continue
row["status"] = status
row["remote_bytes"] = len(remote) if status == 200 else None
row["remote_sha256"] = sha256(remote) if status == 200 else None
row["verdict"], row["detail"] = grade(local, remote, status)
rows.append(row)
return rows
def summarise(rows):
"""Verdict census as a {verdict: n} mapping -- one pass via Counter,
replacing the original's per-call dict.get(k,0)+1 loop."""
return Counter(row["verdict"] for row in rows)
def summary_line(rows, counts=None):
"""One-line tally. Pass a precomputed census via `counts` to avoid
recomputing it; run() reuses its own census instead of re-scanning."""
counts = counts if counts is not None else summarise(rows)
order = ["OK", "DIFFERS", "MISSING", "HTTP_ERROR", "UNREACHABLE",
"READ_FAILED", "SKIPPED_LARGE"]
parts = ["%d %s" % (counts[k], k) for k in order if counts.get(k)]
fatal = sum(counts.get(k, 0) for k in FATAL)
return "landing: %d files, %s, %d fatal" % (len(rows), ", ".join(parts), fatal)
def run(args):
rows = build_rows(args.repo, args.rev, args.branch,
Forge(args.base_url, args.timeout),
args.path, args.max_bytes)
counts = summarise(rows) # single census pass
fatal = sum(counts.get(k, 0) for k in FATAL) # derived once
if not args.quiet:
width = max([len("path")] + [len(r["path"]) for r in rows]) if rows else 4
print("%-*s %-12s %s" % (width, "path", "verdict", "detail"))
for row in rows:
print("%-*s %-12s %s" % (width, row["path"], row["verdict"],
row["detail"]))
print()
print(summary_line(rows, counts)) # reuse the census
if args.json:
payload = {"rev": args.rev, "branch": args.branch,
"base_url": args.base_url, "rows": rows,
"summary": counts,
"fatal": fatal}
try:
with open(args.json, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=2, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
return 1 if fatal else 0
def selftest():
"""A fixture repo and a fixture forge: no network, every verdict reachable."""
import contextlib
import io
root = tempfile.mkdtemp(prefix="landing-check-selftest-")
cells = [0]
bad = [0]
def check(name, got, want):
cells[0] += 1
if got != want:
bad[0] += 1
print("FAIL %s\n got %r\n want %r" % (name, got, want))
try:
repo = os.path.join(root, "repo")
base_ok = os.path.join(root, "forge") # the raw path's parent
served = os.path.join(base_ok, "branch", "main")
os.makedirs(repo)
os.makedirs(os.path.join(served, "tools"))
os.makedirs(os.path.join(served, "docs"))
git(repo, "init", "-q", "-b", "main")
git(repo, "config", "user.email", "selftest@invalid")
git(repo, "config", "user.name", "selftest")
blobs = {"tools/a.py": b"print(1)\n",
"tools/bin.dat": bytes([0x00, 0xff, 0x80, 0x01, 0x00]),
"docs/two words.md": b"space in the name\n"}
for path, data in blobs.items():
full = os.path.join(repo, path)
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, "wb") as fh:
fh.write(data)
git(repo, "add", "-A")
git(repo, "commit", "-q", "-m", "rev one")
rev1 = git(repo, "rev-parse", "HEAD").decode().strip()
# serve rev one, then add a file and commit rev two locally only
for path, data in blobs.items():
dst = os.path.join(served, path)
os.makedirs(os.path.dirname(dst), exist_ok=True)
with open(dst, "wb") as fh:
fh.write(data)
with open(os.path.join(repo, "tools/late.py"), "w") as fh:
fh.write("late = True\n")
git(repo, "add", "-A")
git(repo, "commit", "-q", "-m", "rev two")
rev2 = git(repo, "rev-parse", "HEAD").decode().strip()
forge = Forge("file://" + base_ok)
# --- URL construction -------------------------------------------------
check("url: plain path",
forge.url("main", "tools/a.py"),
"file://" + base_ok + "/branch/main/tools/a.py")
check("url: space is percent-encoded, separators are not",
forge.url("main", "docs/two words.md"),
"file://" + base_ok + "/branch/main/docs/two%20words.md")
# --- git plumbing -----------------------------------------------------
check("discovery: rev one sees its three files",
tracked_at(repo, rev1), sorted(blobs))
check("discovery: rev-scoped, late.py absent at rev one",
"tools/late.py" in tracked_at(repo, rev1), False)
check("discovery: rev two sees late.py",
"tools/late.py" in tracked_at(repo, rev2), True)
with open(os.path.join(repo, "untracked.txt"), "w") as fh:
fh.write("not tracked\n")
check("discovery: an untracked worktree file is never a row",
"untracked.txt" in tracked_at(repo, rev2), False)
check("blob: text bytes are exact",
blob_at(repo, rev1, "tools/a.py"), b"print(1)\n")
check("blob: binary bytes survive cat-file",
blob_at(repo, rev1, "tools/bin.dat"), blobs["tools/bin.dat"])
# --- verdicts ---------------------------------------------------------
ok_rows = build_rows(repo, rev1, "main", forge)
check("all three tracked rows graded OK",
[r["verdict"] for r in ok_rows], ["OK", "OK", "OK"])
check("OK detail carries the served length",
[r["path"] for r in ok_rows if r["local_bytes"] == 9],
["tools/a.py"])
# one served byte changed -> DIFFERS, same length
target = os.path.join(served, "tools/a.py")
with open(target, "wb") as fh:
fh.write(b"print(2)\n")
rows = build_rows(repo, rev1, "main", forge, ["tools/a.py"])
check("mutation: one changed byte is DIFFERS", rows[0]["verdict"], "DIFFERS")
check("mutation: same length is reported as such",
rows[0]["detail"], "same length, different sha256")
# served file deleted -> MISSING
os.remove(target)
rows = build_rows(repo, rev1, "main", forge, ["tools/a.py"])
check("deleted served file is MISSING", rows[0]["verdict"], "MISSING")
check("MISSING carries the 404 status", rows[0]["status"], 404)
# a directory where a file should be -> HTTP_ERROR, not MISSING
os.makedirs(os.path.join(served, "tools/a.py"))
rows = build_rows(repo, rev1, "main", forge, ["tools/a.py"])
check("a directory at the raw path is HTTP_ERROR", rows[0]["verdict"],
"HTTP_ERROR")
# a served root that does not exist -> UNREACHABLE, not MISSING
gone = Forge("file://" + os.path.join(root, "no-such-forge"))
rows = build_rows(repo, rev1, "main", gone, ["tools/a.py"])
check("an absent forge root is UNREACHABLE", rows[0]["verdict"],
"UNREACHABLE")
# a transport that refuses the connection is UNREACHABLE too (no DNS,
# no external network: port 1 on loopback refuses immediately)
refused = Forge("http://127.0.0.1:1/raw", timeout=5)
rows = build_rows(repo, rev1, "main", refused, ["tools/a.py"])
check("a refused connection is UNREACHABLE", rows[0]["verdict"],
"UNREACHABLE")
# restore the good served file before the exit-code cells
shutil.rmtree(os.path.join(served, "tools/a.py"))
with open(os.path.join(served, "tools/a.py"), "wb") as fh:
fh.write(blobs["tools/a.py"])
# --- policy rows ------------------------------------------------------
rows = build_rows(repo, rev1, "main", Forge(base_ok), max_bytes=2)
check("over --max-bytes is SKIPPED_LARGE, never fetched",
sorted({r["verdict"] for r in rows}), ["SKIPPED_LARGE"])
check("SKIPPED_LARGE keeps the local length",
rows[0]["local_bytes"] is not None, True)
# --- exit codes, json, quiet, path filter -----------------------------
out = os.path.join(root, "out.json")
ns = argparse.Namespace(repo=repo, rev=rev1, branch="main",
base_url="file://" + base_ok, path=None,
max_bytes=4 * 1024 * 1024, timeout=5,
json=out, quiet=True)
rc = run(ns)
payload = json.loads(open(out, encoding="utf-8").read())
check("a clean run exits 0", rc, 0)
check("json rows and summary agree",
sum(payload["summary"].values()), len(payload["rows"]))
check("json fatal count is zero on a clean run", payload["fatal"], 0)
with open(os.path.join(served, "tools/a.py"), "wb") as fh:
fh.write(b"print(9)\n")
ns.path = ["tools/a.py"]
ns.json = None
buf1 = io.StringIO()
with contextlib.redirect_stdout(buf1):
rc_path = run(ns)
check("--path restricts the run to one tracked row, and DIFFERS exits 1",
rc_path, 1)
with open(os.path.join(served, "tools/a.py"), "wb") as fh:
fh.write(blobs["tools/a.py"])
buf2 = io.StringIO()
with contextlib.redirect_stdout(buf2):
rc_restored = run(ns)
check("restored bytes make the run clean again", rc_restored, 0)
ns.path = None
# quiet: the table is suppressed, the summary survives
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
run(ns)
printed = buf.getvalue()
check("--quiet prints the summary line and nothing else",
printed, "landing: 3 files, 3 OK, 0 fatal\n")
# --- precondition -----------------------------------------------------
check("a non-git --repo is a precondition error (2)",
_exit2(os.path.join(root, "not-a-repo")), 2)
finally:
shutil.rmtree(root, ignore_errors=True)
print("selftest: %d cells, %d passed, %d failed" % (cells[0], cells[0] - bad[0], bad[0]))
return 0 if bad[0] == 0 else 1
def _exit2(repo):
"""The code main() returns for a path that is not a git work tree."""
probe = subprocess.run(["git", "-C", repo, "rev-parse", "--git-dir"],
capture_output=True)
return 2 if probe.returncode != 0 else 0
def main(argv=None):
ap = argparse.ArgumentParser(
description="verify that the forge serves the bytes this revision tracks")
ap.add_argument("--repo", default=".")
ap.add_argument("--rev", default="HEAD")
ap.add_argument("--branch", default="main")
ap.add_argument("--base-url", default=DEFAULT_BASE)
ap.add_argument("--path", action="append", metavar="PATH",
help="restrict to this tracked path (repeatable)")
ap.add_argument("--max-bytes", type=int, default=4 * 1024 * 1024,
help="blobs larger than this are SKIPPED_LARGE (default 4 MiB)")
ap.add_argument("--timeout", type=float, default=20.0)
ap.add_argument("--json", metavar="PATH")
ap.add_argument("--quiet", action="store_true")
ap.add_argument("--selftest", action="store_true")
args = ap.parse_args(argv)
if args.selftest:
return selftest()
if not os.path.isdir(args.repo):
print("landing_check: --repo %s is not a directory" % args.repo,
file=sys.stderr)
return 2
if _exit2(args.repo) != 0:
print("landing_check: --repo %s is not a git work tree" % args.repo,
file=sys.stderr)
return 2
try:
return run(args)
except RuntimeError as exc:
print("landing_check: %s" % exc, file=sys.stderr)
return 2
if __name__ == "__main__":
sys.exit(main())