From c4c4d699988d546b2db895640ded6fe04f672789 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 06:21:34 -0500 Subject: [PATCH 1/7] tools: GB-6 proposal graph_binding_check (fails on unbound nodes/artifacts; no CI wiring) --- tests/test_graph_binding_check.py | 126 ++++++++++++++++++++ tools/graph_binding_check.py | 188 ++++++++++++++++++++++++++++++ 2 files changed, 314 insertions(+) create mode 100644 tests/test_graph_binding_check.py create mode 100644 tools/graph_binding_check.py diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py new file mode 100644 index 000000000..78f088ec8 --- /dev/null +++ b/tests/test_graph_binding_check.py @@ -0,0 +1,126 @@ +"""Tests for tools/graph_binding_check.py (GB-6 proposal). Synthetic fixtures only.""" +import hashlib +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +TOOL = Path(__file__).resolve().parents[1] / "tools" / "graph_binding_check.py" + + +def sha(b): + return hashlib.sha256(b).hexdigest() + + +class GraphBindingCheckTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + (self.root / "art").mkdir() + (self.root / "art" / "a.json").write_bytes(b"A") + (self.root / "art" / "b.json").write_bytes(b"B") + self.nodes = self.root / "nodes.json" + self.nodes.write_text(json.dumps(["TEST.n1", "TEST.n2"])) + + def tearDown(self): + self.tmp.cleanup() + + def run_tool(self, bindings, *extra, raw=None): + bf = self.root / "bindings.json" + bf.write_text(raw if raw is not None else json.dumps(bindings)) + cmd = [sys.executable, "-B", str(TOOL), "--root", str(self.root), + "--bindings", str(bf), "--graph-nodes", str(self.nodes), + "--scope", "art/*.json", *extra] + p = subprocess.run(cmd, capture_output=True, text=True) + rep = json.loads(p.stdout) if p.stdout else None + return p.returncode, rep, p.stderr + + def full(self): + return [ + {"node_id": "TEST.n1", "artifact_path": "art/a.json", "sha256": sha(b"A")}, + {"node_id": "TEST.n2", "artifact_path": "art/b.json", "sha256": sha(b"B")}, + ] + + def test_all_bound_passes(self): + rc, rep, _ = self.run_tool(self.full()) + self.assertEqual(rc, 0) + self.assertEqual(rep["status"], "ALL_BOUND") + self.assertEqual(rep["scientific_effect"], "NONE") + + def test_empty_bindings_fail_both_directions(self): + rc, rep, _ = self.run_tool([]) + self.assertEqual(rc, 1) + self.assertEqual(rep["unbound_nodes"], ["TEST.n1", "TEST.n2"]) + self.assertEqual(rep["unbound_artifacts"], ["art/a.json", "art/b.json"]) + + def test_report_only_exits_zero(self): + rc, rep, _ = self.run_tool([], "--report-only") + self.assertEqual(rc, 0) + self.assertEqual(rep["status"], "UNBOUND_ITEMS") + + def test_stale_sha_fails(self): + b = self.full() + b[0]["sha256"] = sha(b"old") + rc, rep, _ = self.run_tool(b) + self.assertEqual(rc, 1) + self.assertEqual(rep["bindings_stale_sha"][0]["actual_sha256"], sha(b"A")) + self.assertIn("TEST.n1", rep["unbound_nodes"]) + + def test_missing_artifact_and_unknown_node_fail(self): + b = self.full() + [ + {"node_id": "TEST.n1", "artifact_path": "art/gone.json", "sha256": sha(b"x")}, + {"node_id": "TEST.ghost", "artifact_path": "art/a.json", "sha256": sha(b"A")}, + ] + rc, rep, _ = self.run_tool(b) + self.assertEqual(rc, 1) + self.assertEqual(len(rep["bindings_missing_artifact"]), 1) + self.assertEqual(len(rep["bindings_unknown_node"]), 1) + + def test_symlink_escape_is_outside_root(self): + outside = Path(self.tmp.name).parent / ("gb6_out_" + Path(self.tmp.name).name) + outside.write_bytes(b"X") + try: + (self.root / "art" / "link.txt").symlink_to(outside) + b = self.full() + [{"node_id": "TEST.n1", "artifact_path": "art/link.txt", + "sha256": sha(b"X")}] + rc, rep, _ = self.run_tool(b) + self.assertEqual(rc, 1) + self.assertEqual(len(rep["bindings_outside_root"]), 1) + finally: + outside.unlink() + + def test_malformed_inputs_exit_2(self): + cases = [ + '{"not": "a list"}', + '[{"node_id": "TEST.n1", "artifact_path": "../x", "sha256": "%s"}]' % sha(b"A"), + '[{"node_id": "TEST.n1", "artifact_path": "/abs", "sha256": "%s"}]' % sha(b"A"), + '[{"node_id": "TEST.n1", "artifact_path": "art/a.json", "sha256": "ABC"}]', + '[{"node_id": "TEST.n1", "node_id": "TEST.n2", "artifact_path": "a", "sha256": "%s"}]' % sha(b"A"), + '[{"node_id": "TEST.n1", "artifact_path": "art/a.json", "sha256": "%s", "extra": 1}]' % sha(b"A"), + '[NaN]', + ] + for raw in cases: + with self.subTest(raw=raw): + rc, rep, err = self.run_tool(None, raw=raw) + self.assertEqual(rc, 2) + self.assertIsNone(rep) + self.assertIn("GRAPH_BINDING_INPUT_ERROR", err) + + def test_site_nodes_dir(self): + site = self.root / "site" + site.mkdir() + (site / "TEST.n3.html").write_text("x") + rc, rep, _ = self.run_tool(self.full(), "--site-nodes", str(site)) + self.assertEqual(rc, 1) + self.assertEqual(rep["unbound_nodes"], ["TEST.n3"]) + + def test_deterministic_output(self): + a = self.run_tool([], "--report-only") + b = self.run_tool([], "--report-only") + self.assertEqual(a, b) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py new file mode 100644 index 000000000..c04d03e08 --- /dev/null +++ b/tools/graph_binding_check.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +"""GB-6 proposal: fail on unbound graph nodes and unbound artifacts. + +Engineering hygiene only; scientific_effect NONE. A passing check means every +declared graph node has a binding to an existing artifact whose bytes match +the pinned sha256, and every in-scope artifact is bound to some node. It does +not claim review, acceptance, custody or any mathematical status. + +Inputs (all explicit; no network, no writes): + --root DIR repository root that artifact paths are relative to + --bindings FILE JSON list of {"node_id", "artifact_path", "sha256"} + --graph-nodes FILE JSON list of node IDs (repeatable), and/or + --site-nodes DIR derive node IDs from DIR/*.html file stems + --scope GLOB in-scope artifact glob relative to root (repeatable) + --report-only always exit 0 after printing the report + +Exit codes: 0 all bound; 1 unbound/stale items found; 2 malformed input. +The report is deterministic JSON on stdout (sorted keys, one trailing newline). +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from pathlib import Path, PurePosixPath + +NODE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:\-]{0,191}") +HEX64 = re.compile(r"[0-9a-f]{64}") +BINDING_KEYS = {"node_id", "artifact_path", "sha256"} +LIMIT = 16 * 1024 * 1024 + + +class InputError(ValueError): + pass + + +def _strict_load(path: Path): + raw = path.read_bytes() + if len(raw) > LIMIT: + raise InputError(f"{path.name}: input too large") + + def pairs(items): + out = {} + for key, value in items: + if key in out: + raise InputError(f"{path.name}: duplicate key {key!r}") + out[key] = value + return out + + def bad_constant(token): + raise InputError(f"{path.name}: nonfinite number {token}") + + try: + return json.loads(raw.decode("utf-8"), object_pairs_hook=pairs, + parse_constant=bad_constant) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise InputError(f"{path.name}: invalid JSON ({exc.__class__.__name__})") + + +def _safe_relpath(value: str) -> str: + if not isinstance(value, str) or not value or "\\" in value or "\x00" in value: + raise InputError("artifact_path must be a nonempty POSIX string") + p = PurePosixPath(value) + if p.is_absolute() or any(part in ("..", "") for part in value.split("/")): + raise InputError(f"artifact_path not repo-relative: {value!r}") + return str(p) + + +def load_nodes(files, site_dirs): + nodes = set() + for f in files: + data = _strict_load(Path(f)) + if not isinstance(data, list): + raise InputError(f"{Path(f).name}: graph nodes must be a JSON list") + for n in data: + if not isinstance(n, str) or not NODE_ID.fullmatch(n): + raise InputError(f"invalid node id {n!r}") + if n in nodes: + raise InputError(f"duplicate node id {n!r}") + nodes.add(n) + for d in site_dirs: + for html in sorted(Path(d).glob("*.html")): + n = html.name[:-len(".html")] + if not NODE_ID.fullmatch(n): + raise InputError(f"invalid site node id {n!r}") + nodes.add(n) + return nodes + + +def load_bindings(path): + data = _strict_load(Path(path)) + if not isinstance(data, list): + raise InputError("bindings must be a JSON list") + seen = set() + out = [] + for b in data: + if not isinstance(b, dict) or set(b) != BINDING_KEYS: + raise InputError(f"binding must have exactly keys {sorted(BINDING_KEYS)}") + node, rel, sha = b["node_id"], _safe_relpath(b["artifact_path"]), b["sha256"] + if not isinstance(node, str) or not NODE_ID.fullmatch(node): + raise InputError(f"invalid binding node id {node!r}") + if not isinstance(sha, str) or not HEX64.fullmatch(sha): + raise InputError(f"sha256 must be lower-case 64 hex for {rel}") + key = (node, rel) + if key in seen: + raise InputError(f"duplicate binding {node} -> {rel}") + seen.add(key) + out.append({"node_id": node, "artifact_path": rel, "sha256": sha}) + return out + + +def check(root: Path, nodes, bindings, scopes): + root = root.resolve() + unknown_node, missing, stale, outside = [], [], [], [] + bound_nodes, bound_paths = set(), set() + for b in sorted(bindings, key=lambda x: (x["node_id"], x["artifact_path"])): + if b["node_id"] not in nodes: + unknown_node.append(b) + continue + target = (root / b["artifact_path"]) + try: + resolved = target.resolve() + resolved.relative_to(root) + except ValueError: + outside.append(b) + continue + if not resolved.is_file(): + missing.append(b) + continue + actual = hashlib.sha256(resolved.read_bytes()).hexdigest() + if actual != b["sha256"]: + stale.append(dict(b, actual_sha256=actual)) + continue + bound_nodes.add(b["node_id"]) + bound_paths.add(b["artifact_path"]) + in_scope = set() + for g in scopes: + for p in root.glob(g): + if p.is_file(): + in_scope.add(p.relative_to(root).as_posix()) + report = { + "schema_version": 1, + "scientific_effect": "NONE", + "counts": { + "graph_nodes": len(nodes), + "bindings": len(bindings), + "in_scope_artifacts": len(in_scope), + }, + "unbound_nodes": sorted(nodes - bound_nodes), + "unbound_artifacts": sorted(in_scope - bound_paths), + "bindings_unknown_node": unknown_node, + "bindings_missing_artifact": missing, + "bindings_stale_sha": stale, + "bindings_outside_root": outside, + } + failing = any(report[k] for k in ( + "unbound_nodes", "unbound_artifacts", "bindings_unknown_node", + "bindings_missing_artifact", "bindings_stale_sha", "bindings_outside_root")) + report["status"] = "UNBOUND_ITEMS" if failing else "ALL_BOUND" + return report, failing + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(allow_abbrev=False, description=__doc__.splitlines()[0]) + ap.add_argument("--root", required=True) + ap.add_argument("--bindings", required=True) + ap.add_argument("--graph-nodes", action="append", default=[]) + ap.add_argument("--site-nodes", action="append", default=[]) + ap.add_argument("--scope", action="append", default=[]) + ap.add_argument("--report-only", action="store_true") + args = ap.parse_args(argv) + try: + if not args.graph_nodes and not args.site_nodes: + raise InputError("need --graph-nodes and/or --site-nodes") + nodes = load_nodes(args.graph_nodes, args.site_nodes) + bindings = load_bindings(args.bindings) + report, failing = check(Path(args.root), nodes, bindings, args.scope) + except (InputError, OSError) as exc: + print(f"GRAPH_BINDING_INPUT_ERROR: {exc}", file=sys.stderr) + return 2 + sys.stdout.write(json.dumps(report, sort_keys=True, indent=1) + "\n") + return 0 if (args.report_only or not failing) else 1 + + +if __name__ == "__main__": + sys.exit(main()) From c1044157cd024ac7c66e3815e021beab02e5a0ba Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 06:27:02 -0500 Subject: [PATCH 2/7] tools: GB-6 fix review F1-F4 (empty node set fails closed; scope must be repo-relative; tests) --- tests/test_graph_binding_check.py | 59 +++++++++++++++++++++++++++++++ tools/graph_binding_check.py | 45 +++++++++++++++++++---- 2 files changed, 98 insertions(+), 6 deletions(-) diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py index 78f088ec8..8607000dd 100644 --- a/tests/test_graph_binding_check.py +++ b/tests/test_graph_binding_check.py @@ -116,6 +116,65 @@ def test_site_nodes_dir(self): self.assertEqual(rc, 1) self.assertEqual(rep["unbound_nodes"], ["TEST.n3"]) + def _raw(self, *args): + bf = self.root / "bindings.json" + bf.write_text(json.dumps(self.full())) + cmd = [sys.executable, "-B", str(TOOL), "--root", str(self.root), + "--bindings", str(bf), *args] + p = subprocess.run(cmd, capture_output=True, text=True) + return p.returncode, p.stdout, p.stderr + + # F1: empty or missing node sets fail closed. + def test_f1_missing_site_nodes_dir_exits_2(self): + rc, out, err = self._raw("--site-nodes", str(self.root / "nope")) + self.assertEqual((rc, out), (2, "")) + self.assertIn("GRAPH_BINDING_INPUT_ERROR", err) + + def test_f1_empty_graph_nodes_exits_2_even_report_only(self): + empty = self.root / "empty.json" + empty.write_text("[]") + for extra in ([], ["--report-only"]): + with self.subTest(extra=extra): + rc, out, _ = self._raw("--graph-nodes", str(empty), *extra) + self.assertEqual((rc, out), (2, "")) + + def test_f1_empty_site_nodes_dir_exits_2(self): + (self.root / "emptysite").mkdir() + rc, out, _ = self._raw("--site-nodes", str(self.root / "emptysite")) + self.assertEqual((rc, out), (2, "")) + + def test_f1_scope_matching_nothing_fails(self): + rc, rep, _ = self.run_tool(self.full(), "--scope", "nomatch/*.json") + self.assertEqual(rc, 1) + self.assertEqual(rep["empty_scopes"], ["nomatch/*.json"]) + self.assertEqual(rep["status"], "UNBOUND_ITEMS") + + # F2/F3: scope patterns must stay inside root and exit 2 when malformed. + def test_f2_f3_bad_scope_exits_2_without_traceback(self): + (self.root.parent / ("gb6_sib_" + self.root.name)).write_text("x") + try: + for pat in ("../*", "art/../../*", "/etc/*", "", "art//*", "art\\*"): + with self.subTest(pat=pat): + rc, out, err = self._raw("--graph-nodes", str(self.nodes), + "--scope", pat) + self.assertEqual((rc, out), (2, "")) + self.assertIn("GRAPH_BINDING_INPUT_ERROR", err) + self.assertNotIn("Traceback", err) + finally: + (self.root.parent / ("gb6_sib_" + self.root.name)).unlink() + + def test_f2_scope_symlink_escape_reported(self): + outside = self.root.parent / ("gb6_scope_out_" + self.root.name) + outside.write_bytes(b"X") + try: + (self.root / "art" / "esc.json").symlink_to(outside) + rc, rep, _ = self.run_tool(self.full()) + self.assertEqual(rc, 1) + self.assertEqual(rep["scope_outside_root"], ["art/esc.json"]) + self.assertNotIn("art/esc.json", rep["unbound_artifacts"]) + finally: + outside.unlink() + def test_deterministic_output(self): a = self.run_tool([], "--report-only") b = self.run_tool([], "--report-only") diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py index c04d03e08..49852ea51 100644 --- a/tools/graph_binding_check.py +++ b/tools/graph_binding_check.py @@ -14,7 +14,10 @@ --scope GLOB in-scope artifact glob relative to root (repeatable) --report-only always exit 0 after printing the report -Exit codes: 0 all bound; 1 unbound/stale items found; 2 malformed input. +Exit codes: 0 all bound; 1 unbound/stale items or a scope matching nothing; +2 malformed input, including an empty node set, a missing --site-nodes +directory, or an absolute/'..' --scope (fail closed; --report-only does not +override exit 2). The report is deterministic JSON on stdout (sorted keys, one trailing newline). """ from __future__ import annotations @@ -81,14 +84,28 @@ def load_nodes(files, site_dirs): raise InputError(f"duplicate node id {n!r}") nodes.add(n) for d in site_dirs: + if not Path(d).is_dir(): + raise InputError(f"--site-nodes is not a directory: {Path(d).name!r}") for html in sorted(Path(d).glob("*.html")): n = html.name[:-len(".html")] if not NODE_ID.fullmatch(n): raise InputError(f"invalid site node id {n!r}") nodes.add(n) + if not nodes: + # F1: an empty node set must fail closed, never read as ALL_BOUND. + raise InputError("graph node set is empty") return nodes +def validate_scope(pattern: str) -> str: + """F2/F3: scope globs must be repo-relative; no absolute, '..' or empty parts.""" + if (not isinstance(pattern, str) or not pattern or "\\" in pattern + or "\x00" in pattern or pattern.startswith("/") + or any(part in ("..", "") for part in pattern.split("/"))): + raise InputError(f"--scope not repo-relative: {pattern!r}") + return pattern + + def load_bindings(path): data = _strict_load(Path(path)) if not isinstance(data, list): @@ -135,11 +152,22 @@ def check(root: Path, nodes, bindings, scopes): continue bound_nodes.add(b["node_id"]) bound_paths.add(b["artifact_path"]) - in_scope = set() + in_scope, empty_scopes, scope_outside = set(), [], [] for g in scopes: - for p in root.glob(g): - if p.is_file(): - in_scope.add(p.relative_to(root).as_posix()) + matched = False + for p in root.glob(validate_scope(g)): + if not p.is_file(): + continue + rel = p.relative_to(root).as_posix() + try: + p.resolve().relative_to(root) + except ValueError: + scope_outside.append(rel) # symlink escaping root + continue + matched = True + in_scope.add(rel) + if not matched: + empty_scopes.append(g) report = { "schema_version": 1, "scientific_effect": "NONE", @@ -154,10 +182,13 @@ def check(root: Path, nodes, bindings, scopes): "bindings_missing_artifact": missing, "bindings_stale_sha": stale, "bindings_outside_root": outside, + "empty_scopes": sorted(empty_scopes), + "scope_outside_root": sorted(scope_outside), } failing = any(report[k] for k in ( "unbound_nodes", "unbound_artifacts", "bindings_unknown_node", - "bindings_missing_artifact", "bindings_stale_sha", "bindings_outside_root")) + "bindings_missing_artifact", "bindings_stale_sha", "bindings_outside_root", + "empty_scopes", "scope_outside_root")) report["status"] = "UNBOUND_ITEMS" if failing else "ALL_BOUND" return report, failing @@ -174,6 +205,8 @@ def main(argv=None) -> int: try: if not args.graph_nodes and not args.site_nodes: raise InputError("need --graph-nodes and/or --site-nodes") + for g in args.scope: + validate_scope(g) nodes = load_nodes(args.graph_nodes, args.site_nodes) bindings = load_bindings(args.bindings) report, failing = check(Path(args.root), nodes, bindings, args.scope) From 5c741762f1239a8b1c55f46e44d423175220fee5 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:17:40 -0500 Subject: [PATCH 3/7] GB-6: v2 report mode aligned with the GB-5 contract (e1d1ebc2), plus open-policy report fields - --v2 FILE: strict v2 root/record parsing (kind, repository, artifact_path, commit, blob_sha, sha256, roles, artifact_id, reason, last_seen_commit, url) - compact refs owner/repo:path@commit#blob; git blob sha + sha256 at checked ref - states: bound/stale/missing/historical/absent/external_unverified/ other_repository/tree_unverifiable/unknown_node, plus unbound nodes/artifacts - report only: exit 0 on well-formed input, 2 on malformed input - policy_open.{directory_sources,hist_nodes_without_source,url_or_external_sources} are listed only (475/6364544516), not pass/fail - v1 behaviour unchanged; tests 15 -> 23 --- tests/test_graph_binding_check.py | 138 +++++++++++++++ tools/graph_binding_check.py | 282 +++++++++++++++++++++++++++++- 2 files changed, 418 insertions(+), 2 deletions(-) diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py index 8607000dd..d383a0054 100644 --- a/tests/test_graph_binding_check.py +++ b/tests/test_graph_binding_check.py @@ -181,5 +181,143 @@ def test_deterministic_output(self): self.assertEqual(a, b) +def blob(b): + return hashlib.sha1(b"blob %d\x00" % len(b) + b).hexdigest() + + +C = "1" * 40 +REPO = "d6g8k5htny-coder/main" + + +class GraphBindingV2ReportTest(unittest.TestCase): + """v2 contract (GB-5, e1d1ebc2): report mode only.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + (self.root / "art").mkdir() + (self.root / "art" / "a.json").write_bytes(b"A") + (self.root / "art" / "b.json").write_bytes(b"B") + (self.root / "pkg").mkdir() + (self.root / "pkg" / "x.txt").write_bytes(b"x") + self.graph = {"graph": {"data": {"nodes": { + "TEST.n1": {"source": "art/a.json"}, + "TEST.n2": {"source": "art/b.json"}, + "TEST.dir": {"source": "pkg/"}, + "TEST.url": {"source": "https://github.com/o/r/pull/1"}, + "TEST.ext": {"source": "external:Math-PR7"}, + "hist.old": {}, + "hist.carried": {"source": "art/a.json"}, + }}}} + (self.root / "g.json").write_text(json.dumps(self.graph)) + + def tearDown(self): + self.tmp.cleanup() + + def rec(self, node, path, data, **kw): + r = {"node_id": node, "kind": "file", "repository": REPO, "artifact_path": path, + "commit": C, "blob_sha": blob(data), "sha256": sha(data), "roles": ["source"]} + r.update(kw) + return r + + def doc(self, bindings, scope=("art/*.json",)): + return {"schema_version": 2, + "node_source": {"kind": "site-projection", "repository": REPO, + "path": "g.json", "commit": C}, + "scope": sorted(scope), "bindings": bindings} + + def run_v2(self, doc, raw=None, *extra): + f = self.root / "v2.json" + f.write_text(raw if raw is not None else json.dumps(doc)) + p = subprocess.run([sys.executable, "-B", str(TOOL), "--root", str(self.root), + "--v2", str(f), *extra], capture_output=True, text=True) + return p.returncode, (json.loads(p.stdout) if p.stdout else None), p.stderr + + def test_bound_and_ref_format(self): + rc, rep, _ = self.run_v2(self.doc([self.rec("TEST.n1", "art/a.json", b"A")])) + self.assertEqual(rc, 0) + self.assertEqual(rep["mode"], "report") + self.assertEqual(rep["bound"][0]["ref"], f"{REPO}:art/a.json@{C}#{blob(b'A')}") + self.assertEqual(rep["unbound_artifacts"], ["art/b.json"]) + self.assertIn("TEST.n2", rep["unbound_nodes"]) + + def test_stale_still_exit_zero(self): + r = self.rec("TEST.n1", "art/a.json", b"OLD") + rc, rep, _ = self.run_v2(self.doc([r])) + self.assertEqual(rc, 0) + self.assertEqual(rep["counts"]["stale"], 1) + self.assertEqual(rep["stale"][0]["actual_blob_sha"], blob(b"A")) + + def test_missing_unknown_and_other_kinds(self): + recs = [ + self.rec("TEST.n1", "art/gone.json", b"A"), + self.rec("TEST.n2", "art/b.json", b"B", repository="o/other"), + {"node_id": "TEST.url", "kind": "external_unverified", "url": "https://x/1", + "roles": ["review"]}, + self.rec("ZZ.nope", "art/a.json", b"A"), + {"node_id": "hist.carried", "kind": "absent", "repository": REPO, + "artifact_path": "old.json", "reason": "deleted", "last_seen_commit": C, + "roles": ["source"]}, + self.rec("hist.old", "art/a.json", b"A", kind="historical"), + ] + recs.sort(key=lambda r: (r["node_id"], r.get("repository", ""), + r.get("artifact_path", r.get("url", "")))) + rc, rep, _ = self.run_v2(self.doc(recs)) + self.assertEqual(rc, 0) + c = rep["counts"] + self.assertEqual((c["missing"], c["other_repository"], c["external_unverified"], + c["unknown_node"], c["absent"], c["historical"], c["bound"]), + (1, 1, 1, 1, 1, 1, 0)) + + def test_policy_open_fields_listed_not_judged(self): + rc, rep, _ = self.run_v2(self.doc([])) + self.assertEqual(rc, 0) + pol = rep["policy_open"] + self.assertEqual([d["node_id"] for d in pol["directory_sources"]], ["TEST.dir"]) + self.assertEqual(pol["hist_nodes_without_source"], ["hist.old"]) + self.assertEqual([d["node_id"] for d in pol["url_or_external_sources"]], + ["TEST.ext", "TEST.url"]) + # listing does not change bound/unbound state: all still unbound + for n in ("TEST.dir", "TEST.url", "TEST.ext", "hist.old"): + self.assertIn(n, rep["unbound_nodes"]) + self.assertEqual(rep["counts"]["bound"], 0) + + def test_tree_kind_without_git_is_unverifiable(self): + r = {"node_id": "TEST.dir", "kind": "tree", "repository": REPO, + "artifact_path": "pkg", "commit": C, "blob_sha": C, "roles": ["source"]} + rc, rep, _ = self.run_v2(self.doc([r])) + self.assertEqual(rc, 0) + self.assertEqual(rep["counts"]["tree_unverifiable"], 1) + + def test_malformed_exit_two(self): + bad = [ + dict(self.doc([]), schema_version=1), + dict(self.doc([]), extra=1), + self.doc([self.rec("TEST.n1", "../a.json", b"A")]), + self.doc([self.rec("TEST.n1", "art/a.json", b"A", roles=[])]), + self.doc([self.rec("TEST.n1", "art/a.json", b"A", roles=["boss"])]), + self.doc([self.rec("TEST.n1", "art/a.json", b"A", blob_sha="ABC")]), + self.doc([self.rec("TEST.n2", "art/b.json", b"B"), + self.rec("TEST.n1", "art/a.json", b"A")]), + self.doc([self.rec("TEST.n1", "art/a.json", b"A")] * 2), + self.doc([], scope=("/etc/*",)), + ] + for d in bad: + rc, rep, err = self.run_v2(d) + self.assertEqual(rc, 2, d) + self.assertIn("GRAPH_BINDING_INPUT_ERROR", err) + rc, _, _ = self.run_v2(None, raw='{"schema_version":2,"schema_version":2}') + self.assertEqual(rc, 2) + + def test_empty_node_source_fails_closed(self): + (self.root / "g.json").write_text(json.dumps({"nodes": {}})) + rc, _, _ = self.run_v2(self.doc([])) + self.assertEqual(rc, 2) + + def test_v2_rejects_mixed_v1_args(self): + rc, _, err = self.run_v2(self.doc([]), None, "--scope", "art/*.json") + self.assertEqual(rc, 2) + + if __name__ == "__main__": unittest.main() diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py index 49852ea51..5ca8542dd 100644 --- a/tools/graph_binding_check.py +++ b/tools/graph_binding_check.py @@ -14,7 +14,13 @@ --scope GLOB in-scope artifact glob relative to root (repeatable) --report-only always exit 0 after printing the report -Exit codes: 0 all bound; 1 unbound/stale items or a scope matching nothing; +v2 mode (GB-5 contract in docs/plans/2026-10-08-artifact-index-contract.md): + --v2 FILE v2 root {schema_version:2,node_source,scope,bindings}; + records use kind/repository/artifact_path/commit/ + blob_sha/sha256 and refs owner/repo:path@commit#blob. + REPORT ONLY: exit 0 on well-formed input, 2 otherwise. + +Exit codes (v1): 0 all bound; 1 unbound/stale items or a scope matching nothing; 2 malformed input, including an empty node set, a missing --site-nodes directory, or an absolute/'..' --scope (fail closed; --report-only does not override exit 2). @@ -193,16 +199,288 @@ def check(root: Path, nodes, bindings, scopes): return report, failing +# --------------------------------------------------------------------------- +# v2 report mode (GB-5 contract, main e1d1ebc2). REPORT ONLY: exits 0 on any +# well-formed input and 2 on malformed input; there is no enforce mode. +# --------------------------------------------------------------------------- +V2_ROOT_KEYS = {"schema_version", "node_source", "scope", "bindings"} +V2_NODE_SOURCE_KEYS = {"kind", "repository", "path", "commit"} +V2_NODE_SOURCE_KINDS = {"graph-readback", "site-projection"} +V2_KINDS = {"file", "tree", "historical", "absent", "external_unverified"} +V2_ROLES = {"source", "review", "kernel", "computation", "alignment", "teaching-source"} +V2_REQUIRED = {"node_id", "kind", "roles"} +V2_KIND_FIELDS = { + "file": ({"repository", "artifact_path", "commit", "blob_sha", "sha256"}, set()), + "historical": ({"repository", "artifact_path", "commit", "blob_sha", "sha256"}, set()), + "tree": ({"repository", "artifact_path", "commit", "blob_sha"}, set()), + "absent": ({"repository", "artifact_path", "reason", "last_seen_commit"}, set()), + "external_unverified": ({"url"}, set()), +} +V2_OPTIONAL = {"artifact_id"} +HEX40 = re.compile(r"[0-9a-f]{40}") +REPO = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+") +CONTROL = re.compile(r"[\x00-\x1f\x7f]") + + +def git_blob_sha(data: bytes) -> str: + return hashlib.sha1(b"blob %d\x00" % len(data) + data).hexdigest() + + +def v2_ref(repository: str, path: str, commit: str, blob: str) -> str: + """Compact reference form from the contract: owner/repo:path@commit#blob.""" + return f"{repository}:{path}@{commit}#{blob}" + + +def _v2_path(value) -> str: + if (not isinstance(value, str) or not value or len(value.encode("utf-8")) > 1024 + or value.startswith("/") or "\\" in value or CONTROL.search(value) + or any(part in ("", ".", "..") for part in value.split("/"))): + raise InputError(f"v2 artifact_path not repo-relative: {value!r}") + return value + + +def _hex(value, pattern, name): + if not isinstance(value, str) or not pattern.fullmatch(value): + raise InputError(f"v2 {name} must be full lower-case hex") + return value + + +def load_v2(path): + doc = _strict_load(Path(path)) + if not isinstance(doc, dict) or set(doc) != V2_ROOT_KEYS: + raise InputError(f"v2 root must have exactly keys {sorted(V2_ROOT_KEYS)}") + if type(doc["schema_version"]) is not int or doc["schema_version"] != 2: + raise InputError("v2 schema_version must be integer 2") + ns = doc["node_source"] + if not isinstance(ns, dict) or set(ns) != V2_NODE_SOURCE_KEYS: + raise InputError("v2 node_source must have exactly kind, repository, path, commit") + if ns["kind"] not in V2_NODE_SOURCE_KINDS: + raise InputError("v2 node_source.kind unknown") + if not isinstance(ns["repository"], str) or not REPO.fullmatch(ns["repository"]): + raise InputError("v2 node_source.repository must be owner/name") + _v2_path(ns["path"]) + _hex(ns["commit"], HEX40, "node_source.commit") + scope = doc["scope"] + if (not isinstance(scope, list) or scope != sorted(set(scope)) + or not all(isinstance(g, str) for g in scope)): + raise InputError("v2 scope must be a sorted unique list of strings") + for g in scope: + validate_scope(g) + recs = doc["bindings"] + if not isinstance(recs, list): + raise InputError("v2 bindings must be a list") + seen, keys = set(), [] + for r in recs: + if not isinstance(r, dict) or not V2_REQUIRED <= set(r): + raise InputError("v2 binding missing node_id/kind/roles") + kind = r["kind"] + if kind not in V2_KINDS: + raise InputError(f"v2 binding kind unknown: {kind!r}") + need, _ = V2_KIND_FIELDS[kind] + allowed = V2_REQUIRED | need | V2_OPTIONAL + if not need <= set(r) or not set(r) <= allowed: + raise InputError(f"v2 {kind} binding fields must be {sorted(V2_REQUIRED | need)} (+artifact_id)") + if not isinstance(r["node_id"], str) or not NODE_ID.fullmatch(r["node_id"]): + raise InputError("v2 invalid node_id") + roles = r["roles"] + if (not isinstance(roles, list) or not roles or roles != sorted(set(roles)) + or not set(roles) <= V2_ROLES): + raise InputError("v2 roles must be a nonempty sorted unique subset of the v1 roles") + if "artifact_id" in r and (not isinstance(r["artifact_id"], str) + or not NODE_ID.fullmatch(r["artifact_id"])): + raise InputError("v2 invalid artifact_id") + if "repository" in r and (not isinstance(r["repository"], str) + or not REPO.fullmatch(r["repository"])): + raise InputError("v2 repository must be owner/name") + if "artifact_path" in r: + _v2_path(r["artifact_path"]) + for f in ("commit", "blob_sha", "last_seen_commit"): + if f in r: + _hex(r[f], HEX40, f) + if "sha256" in r: + _hex(r["sha256"], HEX64, "sha256") + for f in ("reason", "url"): + if f in r and (not isinstance(r[f], str) or not r[f] or CONTROL.search(r[f])): + raise InputError(f"v2 {f} must be a nonempty string") + key = (r["node_id"], r.get("repository", ""), r.get("artifact_path", r.get("url", ""))) + if key in seen: + raise InputError(f"v2 duplicate binding {key}") + seen.add(key) + keys.append(key) + if keys != sorted(keys): + raise InputError("v2 bindings must sort by node_id, repository, artifact_path") + return doc + + +def _v2_nodes(root: Path, ns): + """Node universe only from the declared node_source file (graph.data.nodes, + or a top-level nodes dict/list). Returns {node_id: node_object}.""" + target = (root / ns["path"]).resolve() + try: + target.relative_to(root) + except ValueError: + raise InputError("v2 node_source.path escapes root") + if not target.is_file(): + raise InputError("v2 node_source.path missing at checked ref") + data = _strict_load(target) + nodes = data + if isinstance(nodes, dict) and "graph" in nodes: + nodes = nodes["graph"] + if isinstance(nodes, dict) and "data" in nodes: + nodes = nodes["data"] + if isinstance(nodes, dict) and "nodes" in nodes: + nodes = nodes["nodes"] + if isinstance(nodes, list): + nodes = {n.get("id") if isinstance(n, dict) else n: + (n if isinstance(n, dict) else {}) for n in nodes} + if not isinstance(nodes, dict) or not nodes: + raise InputError("v2 node_source has no nodes") # F1: fail closed + for k in nodes: + if not isinstance(k, str) or not NODE_ID.fullmatch(k): + raise InputError(f"v2 invalid node id in node_source {k!r}") + return nodes + + +def _git_tree_sha(root: Path, rel: str): + import subprocess + try: + out = subprocess.run(["git", "-C", str(root), "rev-parse", f"HEAD:{rel}"], + capture_output=True, text=True, timeout=30) + except (OSError, subprocess.SubprocessError): + return None + sha = out.stdout.strip() + return sha if out.returncode == 0 and HEX40.fullmatch(sha) else None + + +def check_v2(root: Path, doc, repository: str): + root = root.resolve() + nodes = _v2_nodes(root, doc["node_source"]) + st = {k: [] for k in ("bound", "stale", "missing", "historical", "absent", + "external_unverified", "other_repository", + "tree_unverifiable", "unknown_node")} + nodes_with_record, file_bound_paths = set(), set() + for r in doc["bindings"]: + item = {"node_id": r["node_id"], "kind": r["kind"]} + if "artifact_path" in r and "repository" in r: + item["artifact_path"] = r["artifact_path"] + if r["kind"] in ("file", "historical", "tree"): + item["ref"] = v2_ref(r["repository"], r["artifact_path"], r["commit"], r["blob_sha"]) + if r["node_id"] not in nodes: + st["unknown_node"].append(item) + continue + nodes_with_record.add(r["node_id"]) + kind = r["kind"] + if kind in ("historical", "absent", "external_unverified"): + if kind == "external_unverified": + item["url"] = r["url"] + st[kind].append(item) + continue + if r["repository"] != repository: + st["other_repository"].append(item) + continue + target = root / r["artifact_path"] + try: + target.resolve().relative_to(root) + except ValueError: + st["missing"].append(dict(item, detail="escapes_root")) + continue + if kind == "tree": + if not target.is_dir(): + st["missing"].append(item) + continue + actual = _git_tree_sha(root, r["artifact_path"]) + if actual is None: + st["tree_unverifiable"].append(item) + elif actual == r["blob_sha"]: + st["bound"].append(item) + else: + st["stale"].append(dict(item, actual_blob_sha=actual)) + continue + if not target.is_file(): + st["missing"].append(item) + continue + data = target.resolve().read_bytes() + blob, digest = git_blob_sha(data), hashlib.sha256(data).hexdigest() + if blob == r["blob_sha"] and digest == r["sha256"]: + st["bound"].append(item) + file_bound_paths.add(r["artifact_path"]) + else: + st["stale"].append(dict(item, actual_blob_sha=blob, actual_sha256=digest)) + in_scope, empty_scopes = set(), [] + for g in doc["scope"]: + matched = False + for p in root.glob(g): + if p.is_file(): + try: + p.resolve().relative_to(root) + except ValueError: + continue + matched = True + in_scope.add(p.relative_to(root).as_posix()) + if not matched: + empty_scopes.append(g) + + # Agent 16's open policy questions (475/6364544516). Listed only: these + # fields never pass or fail anything, and they do not change any node's + # bound/unbound state, which is computed from binding records as above. + pol_dir, pol_hist, pol_ext = [], [], [] + for nid in sorted(nodes): + src = nodes[nid].get("source") if isinstance(nodes[nid], dict) else None + if isinstance(src, str) and (src.startswith(("http://", "https://")) + or src.startswith("external:")): + pol_ext.append({"node_id": nid, "source": src}) + elif isinstance(src, str) and (src.endswith("/") or (root / src).is_dir()): + pol_dir.append({"node_id": nid, "source": src}) + if nid.startswith("hist.") and not src: + pol_hist.append(nid) + for k in st: + st[k].sort(key=lambda x: (x["node_id"], x.get("artifact_path", ""), x["kind"])) + return { + "schema_version": 2, + "mode": "report", + "scientific_effect": "NONE", + "node_source": doc["node_source"], + "counts": {"graph_nodes": len(nodes), "bindings": len(doc["bindings"]), + "in_scope_artifacts": len(in_scope), + **{k: len(v) for k, v in st.items()}}, + **st, + "unbound_nodes": sorted(set(nodes) - nodes_with_record), + "unbound_artifacts": sorted(in_scope - file_bound_paths), + "empty_scopes": empty_scopes, + "policy_open": { + "note": "open policy questions 475/6364544516; listed only, not pass/fail", + "directory_sources": pol_dir, + "hist_nodes_without_source": pol_hist, + "url_or_external_sources": pol_ext, + }, + } + + def main(argv=None) -> int: ap = argparse.ArgumentParser(allow_abbrev=False, description=__doc__.splitlines()[0]) ap.add_argument("--root", required=True) - ap.add_argument("--bindings", required=True) + ap.add_argument("--bindings") + ap.add_argument("--v2", help="v2 bindings file (report mode only)") + ap.add_argument("--repository", default="d6g8k5htny-coder/main", + help="owner/name of --root, for v2 records") ap.add_argument("--graph-nodes", action="append", default=[]) ap.add_argument("--site-nodes", action="append", default=[]) ap.add_argument("--scope", action="append", default=[]) ap.add_argument("--report-only", action="store_true") args = ap.parse_args(argv) + if args.v2 is not None: + try: + if args.bindings or args.graph_nodes or args.site_nodes or args.scope: + raise InputError("--v2 takes node source and scope from the v2 file only") + report = check_v2(Path(args.root), load_v2(args.v2), args.repository) + except (InputError, OSError) as exc: + print(f"GRAPH_BINDING_INPUT_ERROR: {exc}", file=sys.stderr) + return 2 + sys.stdout.write(json.dumps(report, sort_keys=True, indent=1) + "\n") + return 0 # v2 is report-only by contract until Dylan approves enforcement try: + if not args.bindings: + raise InputError("--bindings is required without --v2") if not args.graph_nodes and not args.site_nodes: raise InputError("need --graph-nodes and/or --site-nodes") for g in args.scope: From 51989623080310070dad9d172b59c4ea939b6a80 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:22:57 -0500 Subject: [PATCH 4/7] GB-6 v2: encode GB-5 contract source rules Q2 (directory) and Q3 (URL/external); hist.* stays open - source_rules.directory_sources (contract L101, L87, L94): a directory is pinned by tree or file records; files under it stay unbound artifacts - source_rules.url_or_external_sources (L102, L92): external_unverified, never bound - policy_open.hist_nodes_without_source kept: contract silent on hist.* nodes - report only; tests 23 -> 26 --- tests/test_graph_binding_check.py | 41 ++++++++++++++++++++++++------- tools/graph_binding_check.py | 40 +++++++++++++++++++++++------- 2 files changed, 63 insertions(+), 18 deletions(-) diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py index d383a0054..b1e858485 100644 --- a/tests/test_graph_binding_check.py +++ b/tests/test_graph_binding_check.py @@ -269,18 +269,41 @@ def test_missing_unknown_and_other_kinds(self): c["unknown_node"], c["absent"], c["historical"], c["bound"]), (1, 1, 1, 1, 1, 1, 0)) - def test_policy_open_fields_listed_not_judged(self): + def test_hist_nodes_stay_open_listed_only(self): rc, rep, _ = self.run_v2(self.doc([])) self.assertEqual(rc, 0) - pol = rep["policy_open"] - self.assertEqual([d["node_id"] for d in pol["directory_sources"]], ["TEST.dir"]) - self.assertEqual(pol["hist_nodes_without_source"], ["hist.old"]) - self.assertEqual([d["node_id"] for d in pol["url_or_external_sources"]], - ["TEST.ext", "TEST.url"]) - # listing does not change bound/unbound state: all still unbound - for n in ("TEST.dir", "TEST.url", "TEST.ext", "hist.old"): - self.assertIn(n, rep["unbound_nodes"]) + self.assertEqual(rep["policy_open"]["hist_nodes_without_source"], ["hist.old"]) + self.assertIn("hist.old", rep["unbound_nodes"]) + self.assertNotIn("directory_sources", rep["policy_open"]) + + def test_directory_source_loose_is_unbound(self): + rc, rep, _ = self.run_v2(self.doc([], scope=("pkg/*",))) + d = rep["source_rules"]["directory_sources"] + self.assertEqual([(x["node_id"], x["status"], x["bound"]) for x in d], + [("TEST.dir", "loose_directory_unbound", False)]) + self.assertIn("TEST.dir", rep["unbound_nodes"]) + + def test_directory_tree_record_files_still_unbound(self): + r = {"node_id": "TEST.dir", "kind": "tree", "repository": REPO, + "artifact_path": "pkg", "commit": C, "blob_sha": C, "roles": ["source"]} + rc, rep, _ = self.run_v2(self.doc([r], scope=("pkg/*",))) + self.assertEqual(rc, 0) + d = rep["source_rules"]["directory_sources"][0] + self.assertEqual((d["status"], d["record_kinds"]), ("pinned_by_tree_or_files", ["tree"])) + self.assertEqual(rep["unbound_artifacts"], ["pkg/x.txt"]) # L94: only file records bind + + def test_external_source_never_bound(self): + recs = [{"node_id": "TEST.ext", "kind": "external_unverified", "url": "external:Math-PR7", + "roles": ["source"]}, + {"node_id": "TEST.url", "kind": "external_unverified", + "url": "https://github.com/o/r/pull/1", "roles": ["source"]}] + rc, rep, _ = self.run_v2(self.doc(recs)) + self.assertEqual(rc, 0) + e = rep["source_rules"]["url_or_external_sources"] + self.assertEqual([(x["node_id"], x["has_external_unverified_record"], x["bound"]) + for x in e], [("TEST.ext", True, False), ("TEST.url", True, False)]) self.assertEqual(rep["counts"]["bound"], 0) + self.assertEqual(rep["counts"]["external_unverified"], 2) def test_tree_kind_without_git_is_unverifiable(self): r = {"node_id": "TEST.dir", "kind": "tree", "repository": REPO, diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py index 5ca8542dd..ff8d93a6b 100644 --- a/tools/graph_binding_check.py +++ b/tools/graph_binding_check.py @@ -420,17 +420,36 @@ def check_v2(root: Path, doc, repository: str): if not matched: empty_scopes.append(g) - # Agent 16's open policy questions (475/6364544516). Listed only: these - # fields never pass or fail anything, and they do not change any node's - # bound/unbound state, which is computed from binding records as above. - pol_dir, pol_hist, pol_ext = [], [], [] + # Node `source` rules from the GB-5 contract, fail-closed answers (e1d1ebc2 + # docs/plans/2026-10-08-artifact-index-contract.md). Report only. + # Q2 (L101) directory source: "Bind each file, or use kind: tree"; a tree + # record is evaluated per L87, and files under it stay unbound + # artifacts unless a `file` record names them (L94). + # Q3 (L102, L92) URL/external: source: external_unverified, "Never bound". + # hist.* nodes: the doc has no rule for hist.* nodes (Q1/L100 covers deleted + # files only), so that question stays OPEN and is listed only. + kinds_by_node = {} + bound_nodes = set() + for r in doc["bindings"]: + kinds_by_node.setdefault(r["node_id"], set()).add(r["kind"]) + for item in st["bound"]: + bound_nodes.add(item["node_id"]) + dir_rule, ext_rule, pol_hist = [], [], [] for nid in sorted(nodes): src = nodes[nid].get("source") if isinstance(nodes[nid], dict) else None + kinds = sorted(kinds_by_node.get(nid, ())) if isinstance(src, str) and (src.startswith(("http://", "https://")) or src.startswith("external:")): - pol_ext.append({"node_id": nid, "source": src}) + ext_rule.append({ + "node_id": nid, "source": src, "record_kinds": kinds, + "has_external_unverified_record": "external_unverified" in kinds, + "bound": nid in bound_nodes}) elif isinstance(src, str) and (src.endswith("/") or (root / src).is_dir()): - pol_dir.append({"node_id": nid, "source": src}) + ok = bool(set(kinds) & {"tree", "file"}) + dir_rule.append({ + "node_id": nid, "source": src, "record_kinds": kinds, + "status": "pinned_by_tree_or_files" if ok else "loose_directory_unbound", + "bound": nid in bound_nodes}) if nid.startswith("hist.") and not src: pol_hist.append(nid) for k in st: @@ -447,11 +466,14 @@ def check_v2(root: Path, doc, repository: str): "unbound_nodes": sorted(set(nodes) - nodes_with_record), "unbound_artifacts": sorted(in_scope - file_bound_paths), "empty_scopes": empty_scopes, + "source_rules": { + "contract": "docs/plans/2026-10-08-artifact-index-contract.md@e1d1ebc2", + "directory_sources": dir_rule, # Q2, L101 + L87 + L94 + "url_or_external_sources": ext_rule, # Q3, L102 + L92 + }, "policy_open": { - "note": "open policy questions 475/6364544516; listed only, not pass/fail", - "directory_sources": pol_dir, + "note": "contract silent on hist.* nodes (475/6364544516 Q2); listed only, not pass/fail", "hist_nodes_without_source": pol_hist, - "url_or_external_sources": pol_ext, }, } From b988ac082d02170de43fa8019de8a13a26698b85 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:25:02 -0500 Subject: [PATCH 5/7] GB-6 v2: hist.* nodes reported unbound_historical, never bound (CoS ruling 475 12:24Z, contract L90) - source_rules.hist_nodes replaces policy_open.hist_nodes_without_source - a bound record on a hist.* node is moved out of bound (demoted_from_bound) - report only; tests 26 -> 27 --- tests/test_graph_binding_check.py | 23 +++++++++++++++++++---- tools/graph_binding_check.py | 23 ++++++++++++++--------- 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py index b1e858485..938b431bd 100644 --- a/tests/test_graph_binding_check.py +++ b/tests/test_graph_binding_check.py @@ -269,12 +269,27 @@ def test_missing_unknown_and_other_kinds(self): c["unknown_node"], c["absent"], c["historical"], c["bound"]), (1, 1, 1, 1, 1, 1, 0)) - def test_hist_nodes_stay_open_listed_only(self): + def test_hist_nodes_unbound_historical_without_record(self): rc, rep, _ = self.run_v2(self.doc([])) self.assertEqual(rc, 0) - self.assertEqual(rep["policy_open"]["hist_nodes_without_source"], ["hist.old"]) - self.assertIn("hist.old", rep["unbound_nodes"]) - self.assertNotIn("directory_sources", rep["policy_open"]) + self.assertNotIn("policy_open", rep) + h = {x["node_id"]: x for x in rep["source_rules"]["hist_nodes"]} + self.assertEqual(sorted(h), ["hist.carried", "hist.old"]) + for x in h.values(): + self.assertEqual((x["status"], x["bound"]), ("unbound_historical", False)) + + def test_hist_node_never_bound_even_with_records(self): + recs = [self.rec("hist.carried", "art/a.json", b"A"), + self.rec("hist.old", "art/a.json", b"A", kind="historical")] + rc, rep, _ = self.run_v2(self.doc(recs)) + self.assertEqual(rc, 0) + self.assertEqual(rep["counts"]["bound"], 0) + self.assertEqual(rep["counts"]["historical"], 1) + h = {x["node_id"]: x for x in rep["source_rules"]["hist_nodes"]} + self.assertTrue(h["hist.old"]["has_historical_record"]) + self.assertEqual(h["hist.carried"]["demoted_from_bound"], 1) + for x in h.values(): + self.assertEqual((x["status"], x["bound"]), ("unbound_historical", False)) def test_directory_source_loose_is_unbound(self): rc, rep, _ = self.run_v2(self.doc([], scope=("pkg/*",))) diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py index ff8d93a6b..eb825a0f5 100644 --- a/tools/graph_binding_check.py +++ b/tools/graph_binding_check.py @@ -426,15 +426,19 @@ def check_v2(root: Path, doc, repository: str): # record is evaluated per L87, and files under it stay unbound # artifacts unless a `file` record names them (L94). # Q3 (L102, L92) URL/external: source: external_unverified, "Never bound". - # hist.* nodes: the doc has no rule for hist.* nodes (Q1/L100 covers deleted - # files only), so that question stays OPEN and is listed only. + # hist.* nodes: CoS ruling 2026-10-23 12:24Z (475), fail-closed, in force + # until Dylan or the GB-5 owner changes it: a hist.* node may carry a + # historical record pinned to an old commit, but per L90 that never + # counts as bound -> reported as unbound_historical, never bound. + hist_bound = [i for i in st["bound"] if i["node_id"].startswith("hist.")] + st["bound"] = [i for i in st["bound"] if not i["node_id"].startswith("hist.")] kinds_by_node = {} bound_nodes = set() for r in doc["bindings"]: kinds_by_node.setdefault(r["node_id"], set()).add(r["kind"]) for item in st["bound"]: bound_nodes.add(item["node_id"]) - dir_rule, ext_rule, pol_hist = [], [], [] + dir_rule, ext_rule, hist_rule = [], [], [] for nid in sorted(nodes): src = nodes[nid].get("source") if isinstance(nodes[nid], dict) else None kinds = sorted(kinds_by_node.get(nid, ())) @@ -450,8 +454,12 @@ def check_v2(root: Path, doc, repository: str): "node_id": nid, "source": src, "record_kinds": kinds, "status": "pinned_by_tree_or_files" if ok else "loose_directory_unbound", "bound": nid in bound_nodes}) - if nid.startswith("hist.") and not src: - pol_hist.append(nid) + if nid.startswith("hist."): + hist_rule.append({ + "node_id": nid, "source": src, "record_kinds": kinds, + "has_historical_record": "historical" in kinds, + "status": "unbound_historical", "bound": False, + "demoted_from_bound": sum(1 for i in hist_bound if i["node_id"] == nid)}) for k in st: st[k].sort(key=lambda x: (x["node_id"], x.get("artifact_path", ""), x["kind"])) return { @@ -470,10 +478,7 @@ def check_v2(root: Path, doc, repository: str): "contract": "docs/plans/2026-10-08-artifact-index-contract.md@e1d1ebc2", "directory_sources": dir_rule, # Q2, L101 + L87 + L94 "url_or_external_sources": ext_rule, # Q3, L102 + L92 - }, - "policy_open": { - "note": "contract silent on hist.* nodes (475/6364544516 Q2); listed only, not pass/fail", - "hist_nodes_without_source": pol_hist, + "hist_nodes": hist_rule, # CoS ruling 475 12:24Z + L90 }, } From d433e93feb02068f2d99af107f2ae9f208a5ce02 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:50:17 -0500 Subject: [PATCH 6/7] GB-6 v2: null last_seen_commit on absent, repository on external_unverified, node_source read at declared repo+pin; #525 composition test --- tests/test_graph_binding_check.py | 88 +++++++++++++++++++++++++++++++ tools/graph_binding_check.py | 81 ++++++++++++++++++++++------ 2 files changed, 152 insertions(+), 17 deletions(-) diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py index 938b431bd..9dd027423 100644 --- a/tests/test_graph_binding_check.py +++ b/tests/test_graph_binding_check.py @@ -356,6 +356,94 @@ def test_v2_rejects_mixed_v1_args(self): rc, _, err = self.run_v2(self.doc([]), None, "--scope", "art/*.json") self.assertEqual(rc, 2) + def test_absent_null_last_seen_commit_kept_honest(self): + # GB-5 L104 "without faking a pin": null = never committed; still + # counted with missing, never bound. + a = {"node_id": "hist.old", "kind": "absent", "repository": REPO, + "artifact_path": "gone.json", "reason": "never committed", + "last_seen_commit": None, "roles": ["source"]} + rc, rep, err = self.run_v2(self.doc([a])) + self.assertEqual(rc, 0, err) + self.assertEqual(rep["absent"][0]["last_seen_commit"], None) + self.assertEqual(rep["bound"], []) + del a["last_seen_commit"] + self.assertEqual(self.run_v2(self.doc([a]))[0], 2) # key still required + a["last_seen_commit"] = "abc" + self.assertEqual(self.run_v2(self.doc([a]))[0], 2) # non-null must be 40-hex + + def test_external_unverified_may_name_repository(self): + e = {"node_id": "TEST.ext", "kind": "external_unverified", "repository": REPO, + "url": "external:Math-PR7", "roles": ["source"]} + rc, rep, err = self.run_v2(self.doc([e])) + self.assertEqual(rc, 0, err) + self.assertEqual(rep["bound"], []) + + def test_foreign_node_source_requires_checkout_and_pin(self): + d = self.doc([]) + d["node_source"]["repository"] = "o/other" + self.assertEqual(self.run_v2(d)[0], 2) # no checkout: fail closed + src = self.root / "src" + src.mkdir() + g = lambda *a: subprocess.run(["git", "-C", str(src), *a], check=True, + capture_output=True, text=True).stdout.strip() + g("init", "-q") + (src / "g.json").write_text(json.dumps(self.graph)) + g("add", "g.json") + g("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "g") + pin = g("rev-parse", "HEAD") + (src / "g.json").write_text(json.dumps({"nodes": {"ONLY.wt": {}}})) # dirty tree ignored + d["node_source"]["commit"] = pin + rc, rep, err = self.run_v2(d, None, "--node-source-checkout", str(src)) + self.assertEqual(rc, 0, err) + self.assertTrue(rep["node_source_read"]["pinned"]) + self.assertEqual(rep["counts"]["graph_nodes"], 7) + d["node_source"]["commit"] = "2" * 40 # pin absent from checkout + self.assertEqual(self.run_v2(d, None, "--node-source-checkout", str(src))[0], 2) + + def test_foreign_repository_records_never_bound(self): + r = self.rec("TEST.n1", "art/a.json", b"A", repository="o/other") + rc, rep, _ = self.run_v2(self.doc([r])) + self.assertEqual(rc, 0) + self.assertEqual(rep["bound"], []) + self.assertEqual(rep["other_repository"][0]["node_id"], "TEST.n1") + + +DATA_525 = "29b7d5969e2cd707debb7449c69a71bbcd39cf63" +MATH_PIN = "dbe6c0a1b1a22c7404b0fd173ee0043b98a08cde" + + +class Composition525Test(unittest.TestCase): + """Pinned composition: main#525 data at 29b7d596 + Math- checkout holding + dbe6c0a1 (set GB_MATH_CHECKOUT). Skips when either commit is unavailable.""" + + def test_525_composition(self): + import os + repo = TOOL.parents[1] + math = os.environ.get("GB_MATH_CHECKOUT") + ok = lambda d, c: subprocess.run(["git", "-C", str(d), "cat-file", "-e", c + "^{commit}"], + capture_output=True).returncode == 0 + if not math or not ok(repo, DATA_525) or not ok(math, MATH_PIN): + self.skipTest("29b7d596 or Math- dbe6c0a1 unavailable") + with tempfile.TemporaryDirectory() as t: + arc = subprocess.run(["git", "-C", str(repo), "archive", DATA_525, "architecture"], + capture_output=True, check=True).stdout + subprocess.run(["tar", "-x", "-C", t], input=arc, check=True) + f = "architecture/graph-artifact-bindings.v2.json" + p = subprocess.run([sys.executable, "-B", str(TOOL), "--root", t, "--v2", + str(Path(t) / f), "--node-source-checkout", math], + capture_output=True, text=True) + self.assertEqual(p.returncode, 0, p.stderr) + rep = json.loads(p.stdout) + self.assertTrue(rep["node_source_read"]["pinned"]) + c = rep["counts"] + self.assertEqual((c["bindings"], c["graph_nodes"], c["bound"], c["absent"], + c["historical"], c["external_unverified"], c["other_repository"], + c["unknown_node"], c["missing"], c["stale"]), + (21, 49, 0, 3, 7, 2, 9, 0, 0, 0)) + self.assertEqual([a["last_seen_commit"] for a in rep["absent"]], [None] * 3) + self.assertIn("hist.lemma_closed", rep["unbound_nodes"]) + self.assertEqual(len(rep["unbound_nodes"]), 28) + if __name__ == "__main__": unittest.main() diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py index eb825a0f5..cba6ef404 100644 --- a/tools/graph_binding_check.py +++ b/tools/graph_binding_check.py @@ -214,7 +214,7 @@ def check(root: Path, nodes, bindings, scopes): "historical": ({"repository", "artifact_path", "commit", "blob_sha", "sha256"}, set()), "tree": ({"repository", "artifact_path", "commit", "blob_sha"}, set()), "absent": ({"repository", "artifact_path", "reason", "last_seen_commit"}, set()), - "external_unverified": ({"url"}, set()), + "external_unverified": ({"url"}, {"repository"}), } V2_OPTIONAL = {"artifact_id"} HEX40 = re.compile(r"[0-9a-f]{40}") @@ -276,8 +276,8 @@ def load_v2(path): kind = r["kind"] if kind not in V2_KINDS: raise InputError(f"v2 binding kind unknown: {kind!r}") - need, _ = V2_KIND_FIELDS[kind] - allowed = V2_REQUIRED | need | V2_OPTIONAL + need, opt = V2_KIND_FIELDS[kind] + allowed = V2_REQUIRED | need | opt | V2_OPTIONAL if not need <= set(r) or not set(r) <= allowed: raise InputError(f"v2 {kind} binding fields must be {sorted(V2_REQUIRED | need)} (+artifact_id)") if not isinstance(r["node_id"], str) or not NODE_ID.fullmatch(r["node_id"]): @@ -295,6 +295,10 @@ def load_v2(path): if "artifact_path" in r: _v2_path(r["artifact_path"]) for f in ("commit", "blob_sha", "last_seen_commit"): + # Contract L104: absent needs last_seen_commit "without faking a + # pin"; null = no commit ever contained the file (honest unknown). + if f == "last_seen_commit" and r.get(f, 0) is None: + continue if f in r: _hex(r[f], HEX40, f) if "sha256" in r: @@ -312,17 +316,50 @@ def load_v2(path): return doc -def _v2_nodes(root: Path, ns): - """Node universe only from the declared node_source file (graph.data.nodes, - or a top-level nodes dict/list). Returns {node_id: node_object}.""" - target = (root / ns["path"]).resolve() +def _git_show(checkout: Path, commit: str, rel: str): + import subprocess try: - target.relative_to(root) - except ValueError: - raise InputError("v2 node_source.path escapes root") - if not target.is_file(): - raise InputError("v2 node_source.path missing at checked ref") - data = _strict_load(target) + out = subprocess.run(["git", "-C", str(checkout), "show", f"{commit}:{rel}"], + capture_output=True, timeout=120) + except (OSError, subprocess.SubprocessError): + return None + return out.stdout if out.returncode == 0 else None + + +def _v2_nodes(root: Path, ns, repository: str, ns_checkout): + """Node universe only from the declared node_source (contract L62): its + repository at its pinned commit. Returns ({node_id: node}, read_info). + + - node_source.repository != --repository: --node-source-checkout (a git + checkout of that repository) is required and the file is read with + `git show :`; no checkout, or the pin is not in it -> exit 2. + - same repository: read at the pin from --root's git if available, else + from the --root working tree, reported as pinned=false.""" + import tempfile + if ns["repository"] != repository and ns_checkout is None: + raise InputError(f"v2 node_source repository {ns['repository']} is not --repository " + f"{repository}; pass --node-source-checkout") + checkout = Path(ns_checkout).resolve() if ns_checkout is not None else root + raw = _git_show(checkout, ns["commit"], ns["path"]) + info = {"repository": ns["repository"], "commit": ns["commit"], "path": ns["path"]} + if raw is not None: + info["pinned"] = True + with tempfile.NamedTemporaryFile(suffix=".json") as fh: + fh.write(raw) + fh.flush() + data = _strict_load(Path(fh.name)) + elif ns_checkout is not None: + raise InputError("v2 node_source pin not readable in --node-source-checkout") + else: + target = (root / ns["path"]).resolve() + try: + target.relative_to(root) + except ValueError: + raise InputError("v2 node_source.path escapes root") + if not target.is_file(): + raise InputError("v2 node_source.path missing at checked ref") + info["pinned"] = False + data = _strict_load(target) nodes = data if isinstance(nodes, dict) and "graph" in nodes: nodes = nodes["graph"] @@ -338,7 +375,7 @@ def _v2_nodes(root: Path, ns): for k in nodes: if not isinstance(k, str) or not NODE_ID.fullmatch(k): raise InputError(f"v2 invalid node id in node_source {k!r}") - return nodes + return nodes, info def _git_tree_sha(root: Path, rel: str): @@ -352,9 +389,12 @@ def _git_tree_sha(root: Path, rel: str): return sha if out.returncode == 0 and HEX40.fullmatch(sha) else None -def check_v2(root: Path, doc, repository: str): +def check_v2(root: Path, doc, repository: str, ns_checkout=None): + """Records are evaluated in one repository only (--repository at --root, + the checked ref, contract L85). Records naming another repository are + reported as other_repository and never bound (fail closed).""" root = root.resolve() - nodes = _v2_nodes(root, doc["node_source"]) + nodes, ns_info = _v2_nodes(root, doc["node_source"], repository, ns_checkout) st = {k: [] for k in ("bound", "stale", "missing", "historical", "absent", "external_unverified", "other_repository", "tree_unverifiable", "unknown_node")} @@ -373,6 +413,8 @@ def check_v2(root: Path, doc, repository: str): if kind in ("historical", "absent", "external_unverified"): if kind == "external_unverified": item["url"] = r["url"] + if kind == "absent": + item["last_seen_commit"] = r["last_seen_commit"] st[kind].append(item) continue if r["repository"] != repository: @@ -467,6 +509,8 @@ def check_v2(root: Path, doc, repository: str): "mode": "report", "scientific_effect": "NONE", "node_source": doc["node_source"], + "node_source_read": ns_info, + "checked_repository": repository, "counts": {"graph_nodes": len(nodes), "bindings": len(doc["bindings"]), "in_scope_artifacts": len(in_scope), **{k: len(v) for k, v in st.items()}}, @@ -488,6 +532,8 @@ def main(argv=None) -> int: ap.add_argument("--root", required=True) ap.add_argument("--bindings") ap.add_argument("--v2", help="v2 bindings file (report mode only)") + ap.add_argument("--node-source-checkout", + help="git checkout of node_source.repository (v2), read at its pinned commit") ap.add_argument("--repository", default="d6g8k5htny-coder/main", help="owner/name of --root, for v2 records") ap.add_argument("--graph-nodes", action="append", default=[]) @@ -499,7 +545,8 @@ def main(argv=None) -> int: try: if args.bindings or args.graph_nodes or args.site_nodes or args.scope: raise InputError("--v2 takes node source and scope from the v2 file only") - report = check_v2(Path(args.root), load_v2(args.v2), args.repository) + report = check_v2(Path(args.root), load_v2(args.v2), args.repository, + args.node_source_checkout) except (InputError, OSError) as exc: print(f"GRAPH_BINDING_INPUT_ERROR: {exc}", file=sys.stderr) return 2 From 5870738f82b99c23950a1e27d0fcbf3b293874ca Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:53:29 -0500 Subject: [PATCH 7/7] GB-6 v2: encode CoS rulings 12:51Z (second repo verified at record pin via --repo; null commit only on absent) --- tests/test_graph_binding_check.py | 70 ++++++++++++++++++++++++++--- tools/graph_binding_check.py | 73 +++++++++++++++++++++++++++---- 2 files changed, 130 insertions(+), 13 deletions(-) diff --git a/tests/test_graph_binding_check.py b/tests/test_graph_binding_check.py index 9dd027423..a75766059 100644 --- a/tests/test_graph_binding_check.py +++ b/tests/test_graph_binding_check.py @@ -265,7 +265,7 @@ def test_missing_unknown_and_other_kinds(self): rc, rep, _ = self.run_v2(self.doc(recs)) self.assertEqual(rc, 0) c = rep["counts"] - self.assertEqual((c["missing"], c["other_repository"], c["external_unverified"], + self.assertEqual((c["missing"], c["unverifiable"], c["external_unverified"], c["unknown_node"], c["absent"], c["historical"], c["bound"]), (1, 1, 1, 1, 1, 1, 0)) @@ -400,12 +400,70 @@ def test_foreign_node_source_requires_checkout_and_pin(self): d["node_source"]["commit"] = "2" * 40 # pin absent from checkout self.assertEqual(self.run_v2(d, None, "--node-source-checkout", str(src))[0], 2) + def _foreign_repo(self): + src = self.root / "o_other" + src.mkdir() + g = lambda *a: subprocess.run(["git", "-C", str(src), *a], check=True, + capture_output=True, text=True).stdout.strip() + g("init", "-q") + (src / "d").mkdir() + (src / "d" / "f.json").write_bytes(b"F") + g("add", ".") + g("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "f") + pin, tree = g("rev-parse", "HEAD"), g("rev-parse", "HEAD:d") + (src / "d" / "f.json").write_bytes(b"TIP") # working tree/tip must be ignored + g("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qam", "tip") + return src, pin, tree + + def test_ruling1_second_repo_verified_at_pinned_commit(self): + src, pin, tree = self._foreign_repo() + f = self.rec("TEST.n1", "d/f.json", b"F", repository="o/other", commit=pin) + t = {"node_id": "TEST.dir", "kind": "tree", "repository": "o/other", + "artifact_path": "d", "commit": pin, "blob_sha": tree, "roles": ["source"]} + rc, rep, err = self.run_v2(self.doc([t, f]), None, "--repo", f"o/other={src}") + self.assertEqual(rc, 0, err) + self.assertEqual(sorted(b["node_id"] for b in rep["bound"]), ["TEST.dir", "TEST.n1"]) + self.assertEqual(rep["unverifiable"], []) + + def test_ruling1_unsupplied_repo_bad_commit_bad_blob_unverifiable(self): + src, pin, tree = self._foreign_repo() + good = self.rec("TEST.n1", "d/f.json", b"F", repository="o/other", commit=pin) + rc, rep, _ = self.run_v2(self.doc([good])) # repo not supplied + self.assertEqual((rc, rep["bound"], rep["unverifiable"][0]["detail"]), + (0, [], "repo_not_supplied")) + cases = {"commit_not_found": dict(commit="2" * 40), + "blob_mismatch": dict(blob_sha=blob(b"TIP"), sha256=sha(b"TIP")), + "sha256_mismatch": dict(sha256=sha(b"X")), + "path_not_at_commit": dict(artifact_path="d/none.json")} + for want, kw in cases.items(): + r = dict(good, **kw) + rc, rep, err = self.run_v2(self.doc([r]), None, "--repo", f"o/other={src}") + self.assertEqual(rc, 0, err) + self.assertEqual(rep["bound"], [], want) + self.assertEqual(rep["unverifiable"][0]["detail"], want) + self.assertIn("TEST.n1", rep["nodes_not_bound"]) + self.assertEqual(self.run_v2(self.doc([good]), None, "--repo", "bad")[0], 2) + + def test_ruling2_null_commit_only_on_absent(self): + for kind in ("file", "historical", "tree"): + r = self.rec("TEST.n1", "art/a.json", b"A", kind=kind, commit=None) + if kind == "tree": + del r["sha256"] + self.assertEqual(self.run_v2(self.doc([r]))[0], 2, kind) + a = {"node_id": "TEST.n1", "kind": "absent", "repository": REPO, + "artifact_path": "gone.json", "reason": "never seen", + "last_seen_commit": None, "roles": ["source"]} + rc, rep, _ = self.run_v2(self.doc([a])) + self.assertEqual(rc, 0) + self.assertIn("TEST.n1", rep["nodes_not_bound"]) + self.assertEqual(rep["bound"], []) + def test_foreign_repository_records_never_bound(self): r = self.rec("TEST.n1", "art/a.json", b"A", repository="o/other") rc, rep, _ = self.run_v2(self.doc([r])) self.assertEqual(rc, 0) self.assertEqual(rep["bound"], []) - self.assertEqual(rep["other_repository"][0]["node_id"], "TEST.n1") + self.assertEqual(rep["unverifiable"][0]["node_id"], "TEST.n1") DATA_525 = "29b7d5969e2cd707debb7449c69a71bbcd39cf63" @@ -430,16 +488,18 @@ def test_525_composition(self): subprocess.run(["tar", "-x", "-C", t], input=arc, check=True) f = "architecture/graph-artifact-bindings.v2.json" p = subprocess.run([sys.executable, "-B", str(TOOL), "--root", t, "--v2", - str(Path(t) / f), "--node-source-checkout", math], + str(Path(t) / f), "--repo", f"d6g8k5htny-coder/Math-={math}"], capture_output=True, text=True) self.assertEqual(p.returncode, 0, p.stderr) rep = json.loads(p.stdout) self.assertTrue(rep["node_source_read"]["pinned"]) c = rep["counts"] self.assertEqual((c["bindings"], c["graph_nodes"], c["bound"], c["absent"], - c["historical"], c["external_unverified"], c["other_repository"], + c["historical"], c["external_unverified"], c["unverifiable"], c["unknown_node"], c["missing"], c["stale"]), - (21, 49, 0, 3, 7, 2, 9, 0, 0, 0)) + (21, 49, 9, 3, 7, 2, 0, 0, 0, 0)) + self.assertEqual({b["verified_at"] for b in rep["bound"]}, {MATH_PIN}) + self.assertEqual(len(rep["nodes_not_bound"]), 40) self.assertEqual([a["last_seen_commit"] for a in rep["absent"]], [None] * 3) self.assertIn("hist.lemma_closed", rep["unbound_nodes"]) self.assertEqual(len(rep["unbound_nodes"]), 28) diff --git a/tools/graph_binding_check.py b/tools/graph_binding_check.py index cba6ef404..9eff9df2b 100644 --- a/tools/graph_binding_check.py +++ b/tools/graph_binding_check.py @@ -389,14 +389,59 @@ def _git_tree_sha(root: Path, rel: str): return sha if out.returncode == 0 and HEX40.fullmatch(sha) else None -def check_v2(root: Path, doc, repository: str, ns_checkout=None): - """Records are evaluated in one repository only (--repository at --root, - the checked ref, contract L85). Records naming another repository are - reported as other_repository and never bound (fail closed).""" +def _verify_pinned(checkout: Path, r): + """CoS ruling (1), 12:51Z: a record in a second repository is checked at + its own pinned commit in an explicitly supplied checkout; never a tip. + Returns None when the commit/blob verifies, else a detail string.""" + import subprocess + def git(*a): + try: + return subprocess.run(["git", "-C", str(checkout), *a], + capture_output=True, timeout=120) + except (OSError, subprocess.SubprocessError): + return None + c = git("cat-file", "-e", r["commit"] + "^{commit}") + if c is None or c.returncode != 0: + return "commit_not_found" + o = git("rev-parse", "--verify", "-q", f"{r['commit']}:{r['artifact_path']}") + if o is None or o.returncode != 0: + return "path_not_at_commit" + actual = o.stdout.decode().strip() + if actual != r["blob_sha"]: + return "blob_mismatch" + if r["kind"] == "tree": + t = git("cat-file", "-t", actual) + return None if t is not None and t.stdout.strip() == b"tree" else "not_a_tree" + data = git("cat-file", "blob", actual) + if data is None or data.returncode != 0: + return "blob_unreadable" + if git_blob_sha(data.stdout) != r["blob_sha"] or hashlib.sha256(data.stdout).hexdigest() != r["sha256"]: + return "sha256_mismatch" + return None + + +def parse_repos(items): + out = {} + for it in items: + name, sep, path = it.partition("=") + if not sep or not REPO.fullmatch(name) or not path or name in out: + raise InputError(f"bad --repo {it!r} (want owner/repo=PATH, once per repo)") + out[name] = path + return out + + +def check_v2(root: Path, doc, repository: str, ns_checkout=None, repos=None): + """Records in --repository are evaluated at --root (the checked ref, L85). + Records in another repository are verified at their own pinned commit in + the --repo checkout (CoS ruling 1); not supplied or not verifiable -> + unverifiable, never bound (fail closed).""" root = root.resolve() + repos = dict(repos or {}) + if ns_checkout is None and doc["node_source"]["repository"] in repos: + ns_checkout = repos[doc["node_source"]["repository"]] nodes, ns_info = _v2_nodes(root, doc["node_source"], repository, ns_checkout) st = {k: [] for k in ("bound", "stale", "missing", "historical", "absent", - "external_unverified", "other_repository", + "external_unverified", "unverifiable", "tree_unverifiable", "unknown_node")} nodes_with_record, file_bound_paths = set(), set() for r in doc["bindings"]: @@ -418,8 +463,15 @@ def check_v2(root: Path, doc, repository: str, ns_checkout=None): st[kind].append(item) continue if r["repository"] != repository: - st["other_repository"].append(item) - continue + if r["repository"] not in repos: + st["unverifiable"].append(dict(item, detail="repo_not_supplied")) + continue + why = _verify_pinned(Path(repos[r["repository"]]), r) + if why is not None: + st["unverifiable"].append(dict(item, detail=why)) + continue + st["bound"].append(dict(item, verified_at=r["commit"])) + continue # scope/unbound_artifacts are --repository paths only target = root / r["artifact_path"] try: target.resolve().relative_to(root) @@ -517,6 +569,9 @@ def check_v2(root: Path, doc, repository: str, ns_checkout=None): **st, "unbound_nodes": sorted(set(nodes) - nodes_with_record), "unbound_artifacts": sorted(in_scope - file_bound_paths), + # CoS ruling: unverifiable / absent (incl. null) records count as + # unbound; L93 unbound_nodes stays "no record at all". + "nodes_not_bound": sorted(set(nodes) - {b["node_id"] for b in st["bound"]}), "empty_scopes": empty_scopes, "source_rules": { "contract": "docs/plans/2026-10-08-artifact-index-contract.md@e1d1ebc2", @@ -532,6 +587,8 @@ def main(argv=None) -> int: ap.add_argument("--root", required=True) ap.add_argument("--bindings") ap.add_argument("--v2", help="v2 bindings file (report mode only)") + ap.add_argument("--repo", action="append", default=[], + help="owner/repo=PATH git checkout for a second repository (v2, repeatable)") ap.add_argument("--node-source-checkout", help="git checkout of node_source.repository (v2), read at its pinned commit") ap.add_argument("--repository", default="d6g8k5htny-coder/main", @@ -546,7 +603,7 @@ def main(argv=None) -> int: if args.bindings or args.graph_nodes or args.site_nodes or args.scope: raise InputError("--v2 takes node source and scope from the v2 file only") report = check_v2(Path(args.root), load_v2(args.v2), args.repository, - args.node_source_checkout) + args.node_source_checkout, parse_repos(args.repo)) except (InputError, OSError) as exc: print(f"GRAPH_BINDING_INPUT_ERROR: {exc}", file=sys.stderr) return 2