test(ci): propose generated-lock delta checks

.github/workflows/lock-delta.yml

@@ -0,0 +1,31 @@
+name: Generated lock delta
+on:
+  pull_request:
+    paths:
+      - 'requirements/**'
+      - 'tools/check_lock_delta.py'
+      - 'tools/check_changed_locks.py'
+      - 'tools/lock-delta-scope.json'
+      - 'tools/tests/test_lock_*.py'
+      - '.github/workflows/lock-delta.yml'
+permissions:
+  contents: read
+jobs:
+  lock-delta:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97
+        with:
+          python-version: '3.11'
+      - name: Guard regression controls
+        run: |
+          python -m unittest discover -s tools/tests -p 'test_lock_*.py'
+          python -O -m unittest discover -s tools/tests -p 'test_lock_*.py'
+      - name: Compare changed lock structure
+        env:
+          LOCK_BASE_SHA: ${{ github.event.pull_request.base.sha }}
+        run: python tools/check_changed_locks.py --base "$LOCK_BASE_SHA"

tools/LOCK-VALIDATION.md

@@ -0,0 +1,32 @@
+# Generated requirement lock review
+
+This draft port builds on agentrust-io/weight-custody-manifest#249. Source head `9fcf6bd6684dbd87473d1b0c670eaeb154749a75` still awaits independent acceptance; this port is not an accepted cross-project policy.
+
+The read-only PR job runs standard-library regression controls, then compares changed requirements/*.txt files against the PR base. It rejects removed package/extras/marker identities, malformed or absent SHA-256 hashes, duplicate identities and lock deletion. Renames are enumerated as a deletion and addition, so moving a lock (with or without dependency loss) requires explicit maintainer review instead of treating it only as a new baseline. Version and SHA-256 hash-set changes are reported for human review; hash ordering and duplicate identical hashes are ignored. New locks receive only a structural check because no prior dependency baseline exists.
+
+This recognizes the exact-pin/hash subset emitted by uv/pip-compile. It is not a general PEP 508 equivalence checker, dependency audit, package authentication or proof that an upgrade runs on Windows. Existing repository validation remains required. No SDK runtime or dependency pin changes.
+
+Manual cross-platform resolution is optional and separate: run tools/check_lock_delta.py --resolve --python-version 3.11 with a reviewed generated lock and uv available. It creates isolated Linux/Windows resolution contexts without installing their packages. Keep platform/version choices specific to each lock.
+
+Source: https://github.com/agentrust-io/weight-custody-manifest/pull/249 .
+
+
+Explicitly excluded non-lock files: {}. These remain under their existing compiler/composed-environment checks. The guard does not claim to validate them.
+
+Hash options must follow the marker (when present); every trailing token must be
+a complete lowercase SHA-256 hash option. Quoted marker literals are retained.
+Marker text is compared as an identity, not evaluated or fully validated as
+PEP 508 syntax; supported-platform resolution is a separate check.
+
+The shared checker is byte-identical to WCM source head
+`9fcf6bd6684dbd87473d1b0c670eaeb154749a75`. Its optional manual
+`--all-locks` CLI checks the union of current and base top-level
+`requirements/*.txt` paths; deleted/renamed paths and invalid bases fail, while
+new locks are explicitly identified and structurally checked. This low-level
+manual mode does not apply this port's exclusions. The PR workflow continues to
+use `check_changed_locks.py`: all changed requirement locks are discovered with
+`--no-renames`, and the existing exclusions above remain unchanged. The workflow
+does not install packages or run platform resolvers. Tests exercise hash-only
+changes through this real driver and valid/corrupted runtime, server and custom
+lock filenames. Independent WCM source acceptance and repository-specific
+review remain prerequisites; draft status is retained.

tools/check_changed_locks.py

@@ -0,0 +1,51 @@
+"""Compare changed generated requirement locks without installing dependencies."""
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+from pathlib import Path
+
+from check_lock_delta import check_delta, parse_lock
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", required=True)
+    args = parser.parse_args()
+    root = Path(__file__).resolve().parents[1]
+    scope_file = root / "tools/lock-delta-scope.json"
+    excluded = json.loads(scope_file.read_text()) if scope_file.exists() else {}
+    changed = subprocess.check_output(
+        ["git", "diff", "--no-renames", "--name-only", "-z", args.base, "HEAD", "--", "requirements"],
+        cwd=root,
+    ).decode().split("\0")
+    for relative in changed:
+        if not relative.endswith(".txt"):
+            continue
+        if relative in excluded:
+            print(f"OUT OF SCOPE {relative}: {excluded[relative]}")
+            continue
+        path = root / relative
+        component = root
+        for part in Path(relative).parts:
+            component = component / part
+            if component.is_symlink():
+                raise ValueError(f"lock path must not contain symlinks: {relative}")
+        if not path.is_file():
+            raise ValueError(f"deleted lock requires explicit maintainer review: {relative}")
+        after = path.read_text(encoding="utf-8")
+        previous = subprocess.run(
+            ["git", "show", f"{args.base}:{relative}"],
+            cwd=root, capture_output=True, text=True, check=False,
+        )
+        if previous.returncode:
+            parse_lock(after)
+            print(f"NEW {relative}: structure only; no prior dependency baseline")
+        else:
+            print(f"PASS {relative}: changed packages {check_delta(previous.stdout, after)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/check_lock_delta.py

@@ -0,0 +1,165 @@
+"""Check universal hash-lock structure and optionally resolve two target platforms.
+
+This checks a declared subset of uv/pip-compile output, not arbitrary pip syntax.
+Removal or marker changes require an explicit --allow-removed package review.
+It does not authenticate package publishers or decide whether upgrades are safe.
+"""
+from __future__ import annotations
+
+import argparse
+import re
+import subprocess
+import sys
+import tempfile
+import venv
+from pathlib import Path
+
+PIN = re.compile(r"^([A-Za-z0-9_.-]+)(\[[^]]+\])?==([^\s;]+)(.*)$")
+TOKEN = re.compile(r"\"[^\"]*\"|'[^']*'|[^\s'\"]+")
+HASH = re.compile(r"--hash=sha256:[0-9a-f]{64}(?=\s|$)")
+
+
+def parse_lock(text):
+    lines = []
+    current = ""
+    for raw in text.splitlines():
+        line = raw.strip()
+        if not line or line.startswith("#"):
+            continue
+        current += " " + line.removesuffix("\\").strip()
+        if not line.endswith("\\"):
+            lines.append(current.strip())
+            current = ""
+    if current:
+        raise ValueError("unterminated lock entry")
+    result = {}
+    for line in lines:
+        match = PIN.fullmatch(line)
+        if not match:
+            raise ValueError("expected an exact version pin with SHA-256 hashes")
+        name, extras, version, tail = match.groups()
+        name = re.sub(r"[-_.]+", "-", name).lower()
+        extras = tuple(sorted((extras or "").strip("[]").lower().split(",")))
+        # Keep quoted marker literals intact, then require every trailing
+        # option to be a complete hash token. Removing valid hashes first can
+        # hide a malformed second hash (or another option) inside the marker.
+        tokens = list(TOKEN.finditer(tail))
+        end = 0
+        option = None
+        for token in tokens:
+            if tail[end:token.start()].strip():
+                raise ValueError(f"{name}: unsupported lock option or malformed hash")
+            if option is None and token.group().startswith("--"):
+                option = token
+            end = token.end()
+        if tail[end:].strip():
+            raise ValueError(f"{name}: unsupported lock option or malformed hash")
+        if option is None:
+            raise ValueError(f"{name}: missing SHA-256 hash")
+        marker = tail[:option.start()].strip()
+        if ((marker and not marker.startswith(";"))
+                or not all(HASH.fullmatch(value)
+                           for value in tail[option.start():].split())):
+            raise ValueError(f"{name}: unsupported lock option or malformed hash")
+        # Collapse only unquoted whitespace: spaces inside a literal are part
+        # of marker identity, even when the resulting marker is never active.
+        marker = re.sub(r'''"[^"]*"|'[^']*'|\s+''',
+                        lambda part: " " if part.group().isspace() else part.group(), marker)
+        key = (name, extras, marker)
+        if key in result:
+            raise ValueError(f"duplicate pin: {name} {marker}")
+        result[key] = (version, frozenset(tail[option.start():].split()))
+    if not result:
+        raise ValueError("empty lock")
+    return result
+
+
+def check_delta(before, after, allowed=()):
+    old, new = parse_lock(before), parse_lock(after)
+    allowed = {re.sub(r"[-_.]+", "-", name).lower() for name in allowed}
+    missing = sorted(key for key in old.keys() - new.keys() if key[0] not in allowed)
+    if missing:
+        raise ValueError(f"removed pins or changed platform markers: {missing}")
+    return sorted({key[0] for key in new if old.get(key) != new[key]})
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", help="existing git revision to compare")
+    parser.add_argument("--allow-removed", action="append", default=[])
+    parser.add_argument("--resolve", action="store_true", help="requires uv and package network access")
+    parser.add_argument("--python-version", default="3.11")
+    parser.add_argument("--all-locks", action="store_true",
+                        help="check current and base requirements/*.txt inventory")
+    parser.add_argument("locks", nargs="*", type=Path)
+    args = parser.parse_args(argv)
+    root = Path(__file__).resolve().parents[1]
+    if args.all_locks and args.locks:
+        parser.error("--all-locks cannot be combined with explicit paths")
+    if not args.all_locks and not args.locks:
+        parser.error("provide lock paths or --all-locks")
+    base_paths = set()
+    paths = args.locks
+    if args.all_locks:
+        paths = {path.relative_to(root) for path in (root / "requirements").glob("*.txt")}
+        if args.base:
+            listing = subprocess.run(["git", "ls-tree", "-r", "--name-only", "-z",
+                                      args.base, "--", "requirements/"], cwd=root,
+                                     capture_output=True, text=True, check=True)
+            base_paths = {Path(name) for name in listing.stdout.split("\0")
+                          if Path(name).parent == Path("requirements") and name.endswith(".txt")}
+            paths |= base_paths
+        if not paths:
+            raise ValueError("no requirements locks found")
+        paths = sorted(paths)
+    for path in paths:
+        # Preserve the inventory's lexical name for the Git baseline. Resolving
+        # a symlink first can silently compare server.txt against dev.txt.
+        absolute = root / path
+        lexical = absolute.relative_to(root)
+        if ".." in lexical.parts:
+            raise ValueError(f"lock path must stay inside repository: {path}")
+        component = root
+        for part in lexical.parts:
+            component /= part
+            if component.is_symlink():
+                raise ValueError(f"lock path must not contain symlinks: {path}")
+        if not absolute.is_file():
+            raise ValueError(f"lock must be a regular file: {path}")
+        relative = lexical.as_posix()
+        after = absolute.read_text(encoding="utf-8")
+        parse_lock(after)
+        if args.base:
+            if args.all_locks and path not in base_paths:
+                # Only a successful base inventory can establish a new path.
+                # New locks still receive structural and requested resolver checks.
+                changed = sorted({key[0] for key in parse_lock(after)})
+                print(f"{relative}: new lock; requires dependency review")
+            else:
+                previous = subprocess.run(["git", "show", f"{args.base}:{relative}"], cwd=root,
+                                          capture_output=True, text=True, check=False)
+                if previous.returncode:
+                    raise ValueError(f"cannot read base lock {relative}; review new locks separately")
+                changed = check_delta(previous.stdout, after, args.allow_removed)
+            print(f"{relative}: changed packages {changed}")
+        if args.resolve:
+            # Each resolution uses an isolated interpreter context and ignores
+            # the reviewer's already-installed packages. No package is installed.
+            with tempfile.TemporaryDirectory(prefix="lock-resolution-") as directory:
+                venv.EnvBuilder(with_pip=False).create(directory)
+                python = Path(directory) / ("Scripts/python.exe" if sys.platform == "win32" else "bin/python")
+                for platform in ("x86_64-unknown-linux-gnu", "x86_64-pc-windows-msvc"):
+                    subprocess.run(["uv", "pip", "install", "--python", str(python), "--dry-run",
+                                    "--require-hashes", "--python-version",
+                                    args.python_version, "--python-platform", platform,
+                                    "-r", str(absolute)], check=True, cwd=root)
+        print(f"PASS {relative}")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (ValueError, OSError, subprocess.CalledProcessError) as exc:
+        print(f"FAIL {exc}", file=sys.stderr)
+        raise SystemExit(1) from None

tools/lock-delta-scope.json

@@ -0,0 +1 @@
+{}

tools/tests/test_lock_delta.py

@@ -0,0 +1,102 @@
+"""Synthetic controls for the generated-lock subset; no package installation."""
+import importlib.util
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+SPEC = importlib.util.spec_from_file_location("check_lock_delta", ROOT / "tools/check_lock_delta.py")
+guard = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(guard)
+HASH = " --hash=sha256:" + "a" * 64
+BASE = "core==1" + HASH + "\ncolorama==1 ; sys_platform == 'win32'" + HASH + "\n"
+
+
+class TestLockDelta(unittest.TestCase):
+    def test_version_upgrade_preserves_other_pins(self):
+        self.assertEqual(guard.check_delta(BASE, BASE.replace("core==1", "core==2")), ["core"])
+
+    def test_independent_corruptions(self):
+        cases = [
+            "core==1" + HASH,
+            BASE.replace(" ; sys_platform == 'win32'", ""),
+            BASE.replace(HASH, ""),
+            BASE.replace("a" * 64, "a" * 63),
+            BASE + "core==2" + HASH,
+            "core>=1" + HASH,
+            "",
+        ]
+        for index, after in enumerate(cases):
+            with self.subTest(index=index), self.assertRaises(ValueError):
+                guard.check_delta(BASE, after)
+
+    def test_extra_loss_and_explicit_scoped_removal(self):
+        with self.assertRaises(ValueError):
+            guard.check_delta("coverage[toml]==1" + HASH, "coverage==1" + HASH)
+        self.assertEqual(guard.check_delta(BASE, "core==1" + HASH, ["colorama"]), [])
+        with self.assertRaises(ValueError):
+            guard.check_delta(BASE, "colorama==1" + HASH, ["colorama"])
+
+    def test_optimized_cli_still_refuses_missing_hash(self):
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", dir=ROOT / "tools") as handle:
+            handle.write("unhashed==1\n")
+            handle.flush()
+            result = subprocess.run(
+                [sys.executable, "-O", str(ROOT / "tools/check_lock_delta.py"), handle.name],
+                capture_output=True, text=True, check=False,
+            )
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("missing SHA-256", result.stderr)
+
+
+
+class TestChangedLockCLI(unittest.TestCase):
+    def test_actual_driver_refuses_regeneration_loss_and_accepts_upgrade(self):
+        import shutil
+
+        cases = [
+            ("upgrade", BASE.replace("core==1", "core==2"), True),
+            ("removed-marker", BASE.replace(" ; sys_platform == 'win32'", ""), False),
+            ("deleted", None, False),
+            ("missing-hash", "core==2\n", False),
+        ]
+        for name, after, expected in cases:
+            with self.subTest(case=name), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                (root / "tools").mkdir()
+                (root / "requirements").mkdir()
+                for source in ("check_lock_delta.py", "check_changed_locks.py"):
+                    shutil.copy2(ROOT / "tools" / source, root / "tools" / source)
+                lock = root / "requirements/test.txt"
+                lock.write_text(BASE)
+
+                def git(*args):
+                    return subprocess.check_output(
+                        ["git", "-c", "user.name=Lock test", "-c",
+                         "user.email=lock-test@example.invalid", "-c",
+                         "commit.gpgsign=false", *args],
+                        cwd=root, stderr=subprocess.DEVNULL, text=True,
+                    ).strip()
+
+                git("init")
+                git("add", ".")
+                git("commit", "-m", "synthetic baseline")
+                base = git("rev-parse", "HEAD")
+                if after is None:
+                    lock.unlink()
+                else:
+                    lock.write_text(after)
+                git("add", "-A")
+                git("commit", "-m", "synthetic candidate")
+                result = subprocess.run(
+                    [sys.executable, str(root / "tools/check_changed_locks.py"),
+                     "--base", base], cwd=root, capture_output=True, text=True,
+                    check=False,
+                )
+                self.assertEqual(result.returncode == 0, expected, result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()