diff --git a/docs/contrib/local-check.md b/docs/contrib/local-check.md
index 211fe75ca..0e6b37054 100644
--- a/docs/contrib/local-check.md
+++ b/docs/contrib/local-check.md
@@ -63,6 +63,15 @@ matches, for example a red unit again. The script reads every list from its owne
 `python3 tools/recipes.py pure-crates`, the Python self-tests from the `selftests=(` array of `tools/check-recipes.sh`,
 and the host units from `python3 tools/recipes.py host-nightly`.
 
+`python3 tools/recipes-key-test.py --self-test` checks the public key/ledger CLI in isolated, offline Cargo
+workspaces. Called-source edits must invalidate a recorded green result even when size and mtime stay unchanged.
+Absolute paths and raw identifiers retain conservative owner/dependency bounds when structural resolution is
+uncertain; Cargo-selected roots and declared Rust modules are source regardless of their filename suffix.
+The persistent structural cache must preserve these inputs across fresh processes.
+The selected profiles of every box call remain inputs, including shell defaults and inherited model overrides.
+Changing a selected profile or called source invalidates its consumer; an unrelated private profile or unreached
+source does not. Shared box inputs remain mandatory for every box command.
+
 The mapped checkpoint reader is `bloomery-safetensors`; `gate-safetensors` owns its dtype, span and shard tests.
 `f32_values` also checks directly constructed `Raw` values: shape and dtype must account for the complete
 byte span without overflow. A partial value or a mismatched span is `Error::Format`, not a truncated conversion.
diff --git a/tools/affected-gates.sh b/tools/affected-gates.sh
index 31d844835..bdae0ab5e 100755
--- a/tools/affected-gates.sh
+++ b/tools/affected-gates.sh
@@ -36,5 +36,8 @@
 # the crate graph's true answer. The saving is on leaf changes (serve, tokenizer, sampler, one gate
 # binary, a runner, docs); the larger value is the list itself: no forgotten gate, and a gate left out
 # is a printed record, not a judgment. It cannot see $BLOOMERY_DATA, the ik trees, or the card.
+# Ledger keys deliberately differ: `recipes.py key` may exclude proven-unreached dependency sources
+# while retaining own-target, build/proc-macro, script and opaque bounds. This affected list remains the
+# broad compile/selection contract; narrowing a key does not remove a recipe from the affected list.
 set -euo pipefail
 exec python3 "$(dirname "$0")/recipes.py" affected "$@"
diff --git a/tools/bloomery/fixture_dirs.py b/tools/bloomery/fixture_dirs.py
index 73b3d97f6..95a661253 100644
--- a/tools/bloomery/fixture_dirs.py
+++ b/tools/bloomery/fixture_dirs.py
@@ -11,6 +11,9 @@
 """
 
 import re
+import sys
+
+_REACH_LOADED_CODE = sys._getframe().f_code
 
 FIXTURE_DIRS = "tools/fixture-dirs.tsv"
 REF_PATHS = "tools/ref/ref-paths.sh"
diff --git a/tools/bloomery/reach_cache.py b/tools/bloomery/reach_cache.py
new file mode 100644
index 000000000..aea912a58
--- /dev/null
+++ b/tools/bloomery/reach_cache.py
@@ -0,0 +1,477 @@
+"""Private JSON snapshots of recipes.Reach's structural graph (never per-recipe reach decisions).
+
+The source corpus is content-addressed independently of git, and observed Tree reads/probes are validated
+again before publication and key output. Hydration accepts only the fixed Reach schema below; no code objects,
+pickle, dynamic imports or arbitrary object construction enter the cache protocol.
+"""
+from __future__ import annotations
+
+import dataclasses
+import hashlib
+import functools
+import json
+import marshal
+import os
+import stat
+import sys
+import tempfile
+import types
+import typing
+
+_REACH_LOADED_CODE = sys._getframe().f_code
+
+VERSION = 2
+STATE = ("idents", "lib_root", "lib_roots", "_mods", "_bases", "_macros", "_exported", "_text",
+         "_traits", "ids", "nodes", "adj", "file_root", "builds", "_rules", "_own_rules",
+         "_dispatch", "_rows_nodes", "_enum_nodes", "_family_dirs", "_facts")
+
+
+def _digest(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def _json(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def _bytes(path):
+    with open(path, "rb") as fh:
+        before = os.fstat(fh.fileno())
+        data = fh.read()
+        after = os.fstat(fh.fileno())
+    if (before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (
+            after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns):
+        raise OSError("file changed during reach snapshot: " + path)
+    return data
+
+
+def _file(path):
+    return [os.path.realpath(path), _digest(_bytes(path))]
+
+
+def _corpus(tree):
+    """Enumerate source additions too, including excluded crates and symlinked source referents.
+
+    Cargo's output directory and git's object database are not source owners. Reads from either are still
+    covered by the observed-read provenance. Directory entries (including symlinks) catch new target roots.
+    """
+    rows = []
+    for base, dirs, files in os.walk(tree.root, followlinks=False):
+        dirs[:] = sorted(d for d in dirs if d not in (".git", "target", "__pycache__"))
+        relbase = os.path.relpath(base, tree.root)
+        rows.append([relbase, dirs[:], sorted(files)])
+        for name in sorted(files):
+            if name.endswith((".rs", ".toml", ".py")) or name == "Cargo.lock":
+                path = os.path.join(base, name)
+                rows.append([os.path.relpath(path, tree.root), _file(path)])
+        # Walk symlinked source directories separately, following each real directory only once.
+        for name in dirs[:]:
+            path = os.path.join(base, name)
+            if os.path.islink(path):
+                rows.append([os.path.relpath(path, tree.root), _linked(path)])
+    return _digest(_json(rows))
+
+
+def _linked(root):
+    rows, seen = [], set()
+    def visit(path, rel):
+        real = os.path.realpath(path)
+        if os.path.isdir(path):
+            if real in seen:
+                rows.append([rel, real, "cycle"])
+                return
+            seen.add(real)
+            names = sorted(os.listdir(path))
+            rows.append([rel, real, names])
+            for name in names:
+                visit(os.path.join(path, name), rel + "/" + name)
+        elif os.path.isfile(path):
+            rows.append([rel, _file(path)])
+        else:
+            rows.append([rel, real, "absent"])
+    visit(root, "")
+    return rows
+
+
+@functools.lru_cache(maxsize=16)
+def _disk_code_digest(data, filename, optimize):
+    """Memoize compilation by exact bytes, never file timestamps or a prior disk identity."""
+    code = compile(data, filename, "exec", dont_inherit=True, optimize=optimize)
+    return _digest(marshal.dumps(code, 2))
+
+
+def _code_identity(module, error):
+    """Bind disk bytes to the code actually imported, never reconstruct code from cache data."""
+    loaded = module._REACH_LOADED_CODE
+    path = module.__file__
+    data = _bytes(path)
+    message = "analysis code changed after import: " + path + "; request a fresh process"
+    try:
+        disk = _disk_code_digest(data, loaded.co_filename, sys.flags.optimize)
+    except (SyntaxError, ValueError, OverflowError) as exc:
+        raise error(message) from exc
+    # Version 2 omits marshal's object-sharing flags: runtime references to constants must not
+    # distinguish imported code from the equivalent newly compiled code. Only internal code is dumped.
+    running = _digest(marshal.dumps(loaded, 2))
+    if disk != running:
+        raise error(message)
+    return [os.path.realpath(path), _digest(data), running]
+
+
+def _identity(tree):
+    module = sys.modules[type(tree).__module__]
+    code = [_code_identity(module, module.RecipeError),
+            _code_identity(sys.modules[__name__], module.RecipeError)]
+    fixture = sys.modules.get("bloomery.fixture_dirs")
+    if fixture is not None:
+        code.append(_code_identity(fixture, module.RecipeError))
+    return _digest(_json([VERSION, tree.root, code, tree.metadata,
+                         {n: dataclasses.asdict(p) for n, p in tree.packages.items()}]))
+
+
+def _observed(tree, saw):
+    rows = []
+    for (kind, rel), was in sorted(saw.items()):
+        path = os.path.join(tree.root, rel)
+        if kind == "r":
+            # Tree.read uses text-mode universal newlines; the digest still covers the raw bytes.
+            data = _bytes(path)
+            text = data.decode("utf-8", errors="replace").replace("\r\n", "\n").replace("\r", "\n")
+            if text != was[0]:
+                raise OSError("source changed during reach analysis: " + rel)
+            rows.append([kind, rel, [os.path.realpath(path), _digest(data)]])
+        else:
+            now = os.path.isfile(path) if kind == "f" else os.path.isdir(path)
+            if now != was:
+                raise OSError("source probe changed during reach analysis: " + rel)
+            rows.append([kind, rel, [now, os.path.realpath(path)]])
+    return rows
+
+
+def _holds(tree, rows):
+    if not isinstance(rows, list):
+        return False
+    for row in rows:
+        if not isinstance(row, list) or len(row) != 3:
+            return False
+        kind, rel, was = row
+        if kind not in ("r", "f", "d") or not isinstance(rel, str):
+            return False
+        path = os.path.join(tree.root, rel)
+        if kind == "r":
+            if _file(path) != was:
+                return False
+        elif [(os.path.isfile(path) if kind == "f" else os.path.isdir(path)), os.path.realpath(path)] != was:
+            return False
+    return True
+
+
+def current(reach):
+    try:
+        return (_identity(reach.tree) == reach._key_identity and _corpus(reach.tree) == reach._key_snapshot
+                and _holds(reach.tree, reach._key_observed))
+    except (OSError, ValueError):
+        return False
+
+
+def _encode(value):
+    if value is None or type(value) in (str, int, bool):
+        return value
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [type(value).__name__, [_encode(v) for v in value]]
+    if isinstance(value, dict):
+        return ["dict", [[_encode(k), _encode(v)] for k, v in value.items()]]
+    name = type(value).__name__
+    if name == "RustFacts":
+        return [name, _encode([list(value.units[:value.uses]), list(value.binds), value.exported])]
+    if name == "RuleSet":
+        return [name, _encode(value.rules)]
+    if name in ("Unit", "ImplHead", "Dispatch"):
+        return [name, {f.name: _encode(getattr(value, f.name)) for f in dataclasses.fields(value)}]
+    raise ValueError("unsupported Reach field " + name)
+
+
+def _decode(value, module):
+    if value is None or type(value) in (str, int, bool):
+        return value
+    if not isinstance(value, list) or len(value) != 2 or not isinstance(value[0], str):
+        raise ValueError("invalid Reach value")
+    name, body = value
+    containers = {"list": list, "tuple": tuple, "set": set, "frozenset": frozenset}
+    if name in containers:
+        if not isinstance(body, list):
+            raise ValueError("invalid Reach container")
+        return containers[name](_decode(v, module) for v in body)
+    if name == "dict":
+        if not isinstance(body, list):
+            raise ValueError("invalid Reach map")
+        out = {}
+        for row in body:
+            if not isinstance(row, list) or len(row) != 2:
+                raise ValueError("invalid Reach map entry")
+            key, val = (_decode(v, module) for v in row)
+            if key in out:
+                raise ValueError("duplicate Reach map entry")
+            out[key] = val
+        return out
+    if name == "RustFacts":
+        args = _decode(body, module)
+        if not isinstance(args, list) or len(args) != 3:
+            raise ValueError("invalid Rust facts")
+        return module.RustFacts(*args)
+    if name == "RuleSet":
+        return module.RuleSet(_decode(body, module))
+    if name in ("Unit", "ImplHead", "Dispatch"):
+        cls = getattr(module, name)
+        if not isinstance(body, dict) or set(body) != {f.name for f in dataclasses.fields(cls)}:
+            raise ValueError("invalid Reach record schema")
+        return cls(**{k: _decode(v, module) for k, v in body.items()})
+    raise ValueError("foreign Reach record")
+
+
+@functools.cache
+def _record_types(cls):
+    return typing.get_type_hints(cls)
+
+
+@functools.cache
+def _schema_shape(annotation):
+    return typing.get_origin(annotation), typing.get_args(annotation), dataclasses.is_dataclass(annotation)
+
+
+def _typed(value, annotation):
+    """Strict container/record checks for the fixed structural schema; bool is not a node index."""
+    if annotation is str or annotation is int or annotation is bool or annotation is type(None):
+        return type(value) is annotation
+    origin, args, record = _schema_shape(annotation)
+    if origin in (typing.Union, types.UnionType):
+        return any(_typed(value, a) for a in args)
+    if origin in (list, set, frozenset):
+        return type(value) is origin and all(_typed(v, args[0]) for v in value)
+    if origin is dict:
+        return type(value) is dict and all(_typed(k, args[0]) and _typed(v, args[1]) for k, v in value.items())
+    if origin is tuple:
+        if type(value) is not tuple:
+            return False
+        if len(args) == 2 and args[1] is Ellipsis:
+            return all(_typed(v, args[0]) for v in value)
+        return len(value) == len(args) and all(_typed(v, a) for v, a in zip(value, args))
+    if type(value) is not annotation:
+        return False
+    if record:
+        return all(_typed(getattr(value, k), a) for k, a in _record_types(annotation).items())
+    return True
+
+
+def _validate(state, tree, module):
+    """Validate node/root/unit identity and every edge/rule/dispatch index before touching a live Reach."""
+    if not isinstance(state, dict) or set(state) != set(STATE):
+        raise ValueError("invalid Reach state schema")
+    roots = {(p.name, t.src, test) for p in tree.packages.values() for t in p.targets for test in (False, True)}
+    root = tuple[str, str, bool]
+    hit = tuple[root, str, int]
+    mod = tuple[str, ...]
+    schemas = {
+        "idents": dict[str, set[str]], "lib_root": dict[str, root], "lib_roots": set[root],
+        "_mods": dict[root, dict[mod, str]], "_bases": dict[root, dict[str, list[mod]]],
+        "_macros": dict[root, dict[str, list[tuple[str, int]]]],
+        "_exported": dict[str, dict[str, list[tuple[str, int]]]], "_text": dict[str, str],
+        "_traits": dict[tuple, list[tuple[hit, frozenset[str], str]]],
+        "ids": dict[hit, int], "nodes": list[hit], "adj": list[list[tuple[int, str]]],
+        "file_root": dict[str, root], "builds": dict[str, str],
+        "_rules": module.RuleSet, "_own_rules": dict[root, module.RuleSet],
+        "_dispatch": dict[root, module.Dispatch], "_rows_nodes": dict[int, module.Dispatch],
+        "_enum_nodes": frozenset[int], "_family_dirs": tuple[str, ...],
+        "_facts": dict[tuple[str, bool], module.RustFacts],
+    }
+    for key, annotation in schemas.items():
+        if not _typed(state[key], annotation):
+            raise ValueError("invalid Reach field type: " + key)
+    for (f, test), facts in state["_facts"].items():
+        if f not in state["_text"] or not _typed(facts.units, tuple[module.Unit, ...]):
+            raise ValueError("invalid Reach fact provenance")
+        if not _typed(facts.binds, tuple[tuple[mod, mod, str, bool], ...]) or not _typed(facts.exported, frozenset[str]):
+            raise ValueError("invalid Reach import facts")
+    for f, text in state["_text"].items():
+        if not f.endswith("/<doctests>.rs"):
+            with open(os.path.join(tree.root, f), encoding="utf-8", errors="replace") as fh:
+                if fh.read() != text:
+                    raise ValueError("invalid Reach source provenance")
+    nodes, adj, ids = state["nodes"], state["adj"], state["ids"]
+    if not isinstance(nodes, list) or not isinstance(adj, list) or len(nodes) != len(adj):
+        raise ValueError("invalid Reach adjacency")
+    if ids != {h: i for i, h in enumerate(nodes)} or len(ids) != len(nodes):
+        raise ValueError("invalid Reach node index")
+    def index(i):
+        if type(i) is not int or not 0 <= i < len(nodes):
+            raise ValueError("invalid Reach node reference")
+    for rk, f, k in nodes:
+        if rk not in roots or not isinstance(f, str) or type(k) is not int or k < -1:
+            raise ValueError("invalid Reach node identity")
+        if k >= 0:
+            facts = state["_facts"].get((f, rk[2]))
+            if not isinstance(facts, module.RustFacts) or k >= len(facts.units):
+                raise ValueError("invalid Reach unit index")
+    for edges in adj:
+        if not isinstance(edges, list):
+            raise ValueError("partial Reach graph")
+        for i, why in edges:
+            index(i)
+            if not isinstance(why, str):
+                raise ValueError("invalid Reach edge")
+    for rk, mods in state["_mods"].items():
+        if rk not in roots or not isinstance(mods, dict) or mods.get(()) != rk[1]:
+            raise ValueError("invalid Reach module root")
+    for rs in [state["_rules"], *state["_own_rules"].values()]:
+        if not isinstance(rs, module.RuleSet):
+            raise ValueError("invalid Reach rules")
+        for i, f, why, groups, name in rs.rules:
+            index(i)
+            if nodes[i][1] != f or not isinstance(why, str) or (name is not None and not isinstance(name, str)):
+                raise ValueError("invalid Reach rule identity")
+            for group in groups:
+                for j in group:
+                    index(j)
+    for rk, dispatch in state["_dispatch"].items():
+        if rk not in roots or not isinstance(dispatch, module.Dispatch):
+            raise ValueError("invalid Reach dispatch")
+        index(dispatch.node)
+        for i in dispatch.enumerators:
+            index(i)
+        for i, rows in dispatch.entries.items():
+            index(i)
+            if any(type(r) is not int or not 0 <= r < len(dispatch.rows) for r in rows):
+                raise ValueError("invalid Reach dispatch row")
+    if state["_rows_nodes"] != {d.node: d for d in state["_dispatch"].values()}:
+        raise ValueError("invalid Reach dispatch index")
+    if any(rk not in roots for rk in state["lib_roots"] | set(state["_own_rules"])):
+        raise ValueError("invalid Reach root index")
+    # Reconstructible indexes must agree with their canonical owners, not merely be in range.
+    expected_libs = {p.name: (p.name, p.lib().src, False) for p in tree.packages.values() if p.lib() is not None}
+    if state["lib_root"] != expected_libs or state["lib_roots"] != set(expected_libs.values()):
+        raise ValueError("invalid Reach library root index")
+    expected_files = {}
+    for rk in expected_libs.values():
+        for f in state["_mods"][rk].values():
+            expected_files.setdefault(f, rk)
+    if state["file_root"] != expected_files:
+        raise ValueError("invalid Reach file owner index")
+    for rk, by in state["_bases"].items():
+        expected = {}
+        for path, f in state["_mods"][rk].items():
+            if not path or state["_mods"][rk].get(path[:-1]) != f:
+                expected.setdefault(f, []).append(path)
+        if by != expected:
+            raise ValueError("invalid Reach module base index")
+    for rk, names in state["_macros"].items():
+        expected = {}
+        for f in sorted(set(state["_mods"][rk].values())):
+            for name, indexes in state["_facts"][(f, rk[2])].macro_named.items():
+                expected.setdefault(name, []).extend((f, i) for i in indexes)
+        if names != expected:
+            raise ValueError("invalid Reach macro index")
+    for pkg, names in state["_exported"].items():
+        rk = expected_libs.get(pkg)
+        expected = {} if rk is None else {
+            name: [(f, i) for f, i in hits if name in state["_facts"][(f, rk[2])].exported]
+            for name, hits in state["_macros"][rk].items()}
+        if names != expected:
+            raise ValueError("invalid Reach exported macro index")
+    for values in state["_traits"].values():
+        if any(h not in ids for h, _, _ in values):
+            raise ValueError("invalid Reach trait unit index")
+    if state["_enum_nodes"] != frozenset(i for d in state["_dispatch"].values() for i in d.enumerators):
+        raise ValueError("invalid Reach enumerator index")
+
+
+def prepare(reach):
+    tree = reach.tree
+    module = sys.modules[type(reach).__module__]
+    identity, corpus = _identity(tree), _corpus(tree)
+    # Reach's eager initializer may have walked before this snapshot. Neither a miss nor a hit
+    # may pass those unprotected local walks on to Graph's later key-input discovery.
+    tree._walk_cache.clear()
+    tree.unresolved.clear()
+    namespace = _digest(tree.root.encode())
+    directory = os.path.join(os.path.expanduser("~"), ".cache", "bloomery", "reach", namespace)
+    path = os.path.join(directory, "graph.json")
+    try:
+        st = os.lstat(path)
+        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or st.st_mode & 0o077:
+            raise ValueError("non-private Reach cache")
+        with open(path, encoding="utf-8") as fh:
+            payload = json.load(fh)
+        if (not isinstance(payload, dict) or set(payload) != {"version", "identity", "corpus", "observed", "state", "digest"}
+                or payload["version"] != VERSION or payload["identity"] != identity or payload["corpus"] != corpus
+                or payload["digest"] != _digest(_json(payload["state"])) or not _holds(tree, payload["observed"])):
+            raise ValueError("stale Reach cache")
+        state = _decode(payload["state"], module)
+        _validate(state, tree, module)
+        # Complete validation precedes hydration; partial or foreign data cannot affect a live graph.
+        if _identity(tree) != identity or _corpus(tree) != corpus or not _holds(tree, payload["observed"]):
+            raise module.RecipeError("source tree changed during reach cache validation")
+        reach.__dict__.update(state)
+        # Resolver results are a recomputable per-analysis memo, not structural graph state.
+        # Discard any eager results made before this validated snapshot was installed.
+        reach._resolved.clear()
+        reach._key_identity, reach._key_snapshot, reach._key_observed = identity, corpus, payload["observed"]
+        return
+    except (OSError, ValueError, TypeError, KeyError, IndexError, AttributeError, RecursionError):
+        pass
+    saw = {}
+    tree._saw.append(saw)
+    try:
+        # Reinitialize the same object, not its Tree or Graph owner, under observation. This
+        # discards eager module maps and captures all discovery reads before materialization.
+        type(reach).__init__(reach, tree)
+        for pkg in tree.packages.values():
+            for target in pkg.targets:
+                for test in (False, True):
+                    rk = (pkg.name, target.src, test)
+                    reach.mods(rk)
+                    reach.own_rules(rk)
+                    for f in sorted(tree.tree_of(target.src, test)[0]):
+                        for hit in reach.units(rk, f):
+                            reach.node(hit)
+            if pkg.lib() is not None:
+                rk = (pkg.name, pkg.lib().src, False)
+                for hit in reach.units(rk, reach.doctests(pkg.name)):
+                    reach.node(hit)
+        reach.rules()
+        reach.dispatches()
+        i = 0
+        while i < len(reach.nodes):
+            reach.edges(i)
+            i += 1
+    finally:
+        tree._saw.pop()
+    observed = _observed(tree, saw)
+    if _identity(tree) != identity or _corpus(tree) != corpus or not _holds(tree, observed):
+        raise module.RecipeError("source tree changed during reach analysis")
+    state = {k: getattr(reach, k) for k in STATE}
+    _validate(state, tree, module)
+    encoded = _encode(state)
+    payload = {"version": VERSION, "identity": identity, "corpus": corpus, "observed": observed,
+               "state": encoded, "digest": _digest(_json(encoded))}
+    reach._key_identity, reach._key_snapshot, reach._key_observed = identity, corpus, observed
+    # A read-only HOME is a cache miss, not a key failure. Computation above remains authoritative.
+    temporary = None
+    try:
+        os.makedirs(directory, mode=0o700, exist_ok=True)
+        ds = os.lstat(directory)
+        if not stat.S_ISDIR(ds.st_mode) or ds.st_uid != os.getuid() or ds.st_mode & 0o077:
+            return
+        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=directory, delete=False) as fh:
+            temporary = fh.name
+            # Keep publication in the C encoder, not dump's millions of per-token Python writes.
+            fh.write(json.dumps(payload, separators=(",", ":")))
+            fh.flush()
+            os.fsync(fh.fileno())
+        reach.validate()
+        os.replace(temporary, path)
+        temporary = None
+    except OSError:
+        pass
+    finally:
+        if temporary is not None:
+            os.unlink(temporary)
diff --git a/tools/check-recipes.sh b/tools/check-recipes.sh
index 0ea7a30a8..444e23671 100755
--- a/tools/check-recipes.sh
+++ b/tools/check-recipes.sh
@@ -387,6 +387,7 @@ pytools_list() {
     "tools/model_tax.py --self-test"
     "tools/owners_check.py --self-test"
     "tools/pr-gate-lines.py --self-test"
+    "tools/recipes-key-test.py --self-test"
     "tools/ref/asm-census.py --self-test"
     "tools/ref/cache-options-test.py --self-test"
     "tools/ref/check-int-twins.py --self-test"
diff --git a/tools/gate-batch.sh b/tools/gate-batch.sh
index ee696e59d..8cf55d78f 100755
--- a/tools/gate-batch.sh
+++ b/tools/gate-batch.sh
@@ -359,10 +359,14 @@
 # A skip names its source, `src=lead` or `src=round`, in run.log and the dry run. A rebase moves the
 # key of every item whose closure the landed commits touched, so a landing still reruns those; the
 # saving is in the items the rebase left alone.
-# Each item's input key is `tools/recipes.py key` (its section header lists the parts): the files its
-# targets read (the walk `just affected` uses; the whole tree for a command run on the Mac, a cargo
-# subcommand the parser does not model, or a script that walks the tree), the text of its recipe and
-# of every recipe it depends on, the scripts those name, the cargo globals and the cuda-oxide pin, its
+# Each item's input key is `tools/recipes.py key` (its section header lists the parts): its own target
+# sources, uncut reached dependency sources, build/procedural-macro compilation inputs, named script
+# reads and manifests. Compiled regular-source macro invocations add structural roots even in otherwise
+# unreached modules; implicit impls include non-.rs #[path] modules. Opaque expansion/linkage keeps
+# conservative bounds; a Mac command, unmodelled cargo subcommand or tree-walking script still keys
+# the whole tree. `just affected` keeps its broader
+# compile closure: an affected recipe need not have a changed ledger key. Keys also include the text
+# of the recipe and every dependency recipe, the scripts those name, cargo globals and cuda-oxide pin, its
 # ARGS, its full BLOOMERY_BOX_ENV (the caller's, the item's, its lane's card), the Mac-side variables
 # box.sh carries, and a box manifest read once per batch through one box.sh call (`recipes.py
 # box-manifest` behind box.sh's guard with BLOOMERY_BOX_WAIT=0: refused while the lease or a hold is
@@ -375,6 +379,17 @@
 # is not bounded by its recipe text, so a data change reruns every item — the safe side; a model
 # directory's rows enter only the keys of the items that can open it (a /models literal in the item's
 # files or recipe text, its profile among them, or a /models value in the box env).
+# Structural reach is cached privately in ~/.cache/bloomery/reach/<checkout-id>/graph.json. The cache
+# contains no green verdicts: source/code/metadata/read/probe content must still match, and malformed or
+# stale snapshots rebuild. Code-identity compilation is memoized by exact source bytes, filename and
+# optimization level; cached recipe inputs still recheck current content, and key output is validated again.
+# Immutable file-local syntax facts are memoized by exact text; bindings and macro resolution stay
+# graph-specific. Schema descriptors are reused, never value-validation verdicts. Module ancestors are
+# collected once per root/file through the existing base index, not once per reached item.
+# The snapshot persists structural owners/nodes/edges, not the recomputable resolver lookup memo.
+# Cache hydration discards eager resolver results; lookups are recomputed from the validated structure.
+# Runtime reach is not profile-pruned for keys. The v2 key format deliberately invalidates pre-reach ledger
+# entries once; an old green cannot authorize a narrower new key.
 #   Keys are computed before the lease check. An item whose key is in the ledger does not run: run.log
 # gets `<stem> rc=skip green-at=<commit> <date> lane=<L>`. Every other item runs as without --ledger;
 # one whose final rc is 0 is recorded after its key is computed again (against the same pre-batch box
diff --git a/tools/recipes-key-test.py b/tools/recipes-key-test.py
new file mode 100755
index 000000000..b4d357732
--- /dev/null
+++ b/tools/recipes-key-test.py
@@ -0,0 +1,728 @@
+#!/usr/bin/env python3
+"""Consumer regressions for reached recipe keys, using offline Cargo workspaces.
+
+Run: python3 tools/recipes-key-test.py --self-test
+Against another implementation: add --source-root /path/to/checkout.
+RECIPES_KEY_TEST_FILTER selects unittest names by substring (comma-separated).
+No fixture compiles Rust or runs a box, contacts a registry, or needs a GPU.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import hashlib
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+SOURCE_ROOT = Path(__file__).resolve().parent.parent
+
+
+class Workspace:
+    """A real checkout, metadata, CLI and private cache; every key is a new process."""
+
+    def __init__(self, case: unittest.TestCase, files: dict[str, str], *, git: bool = True):
+        self.case = case
+        self.tmp = tempfile.TemporaryDirectory(prefix="recipes-key-cli-")
+        case.addCleanup(self.tmp.cleanup)
+        self.base = Path(self.tmp.name)
+        self.root = self.base / "checkout"
+        self.home = self.base / "home"
+        self.root.mkdir()
+        self.home.mkdir()
+        old_home = Path.home()
+        self.env = {k: v for k, v in os.environ.items()
+                    if not k.startswith(("BLOOMERY_", "CARGO_", "RUSTFLAGS", "RUSTDOCFLAGS"))}
+        self.env.update(HOME=str(self.home), XDG_CACHE_HOME=str(self.home / ".cache"),
+                        CARGO_HOME=os.environ.get("CARGO_HOME", str(old_home / ".cargo")),
+                        RUSTUP_HOME=os.environ.get("RUSTUP_HOME", str(old_home / ".rustup")),
+                        CARGO_NET_OFFLINE="true", PYTHONDONTWRITEBYTECODE="1",
+                        GIT_CONFIG_GLOBAL=os.devnull, GIT_CONFIG_SYSTEM=os.devnull,
+                        GIT_AUTHOR_NAME="test", GIT_AUTHOR_EMAIL="test@example.invalid",
+                        GIT_COMMITTER_NAME="test", GIT_COMMITTER_EMAIL="test@example.invalid")
+        for rel, text in files.items():
+            self.write(rel, text)
+        # Copy production code, not an echo of the desired key or a patched resolver.
+        self.copy("tools/recipes.py")
+        helpers = SOURCE_ROOT / "tools/bloomery"
+        for src in helpers.rglob("*.py"):
+            self.copy(str(src.relative_to(SOURCE_ROOT)))
+        self.copy("tools/fixture-dirs.tsv")
+        for rel in ("tools/box.sh", "tools/ref/ref-paths.sh",
+                    "tools/ref/models/deepseek41.sh", "tools/ref/models/deepseek2.sh"):
+            self.write(rel, "#!/bin/sh\n")
+        self.manifest = self.base / "manifest.txt"
+        self.manifest.write_text("# box-manifest 1 host=selftest data=/tmp/recipes-key-data\n"
+                                 "env\tbloomery-env.sh\t" + "a" * 64 + "\n"
+                                 "os\tkernel\tselftest\n"
+                                 "tool\trustc\tselftest-offline\n# end\n")
+        self.ledger = self.base / "ledger.tsv"
+        self.parts_dir = self.base / "parts"
+        self.run("cargo", "generate-lockfile", "--offline", "-q")
+        if git:
+            self.run("git", "init", "-q", "-b", "main")
+            self.run("git", "add", "-A")
+            self.run("git", "commit", "-q", "-m", "fixture")
+
+    def copy(self, rel: str) -> None:
+        dst = self.root / rel
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copyfile(SOURCE_ROOT / rel, dst)
+
+    def write(self, rel: str, text: str) -> None:
+        dst = self.root / rel
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        dst.write_text(text)
+
+    def run(self, *args: str) -> subprocess.CompletedProcess:
+        done = subprocess.run(args, cwd=self.root, env=self.env, text=True,
+                              capture_output=True, timeout=120)
+        self.case.assertEqual(done.returncode, 0,
+                              f"{args!r}\nstdout: {done.stdout}\nstderr: {done.stderr}")
+        return done
+
+    def keys(self, *items: str, ledger: bool = False) -> dict[str, tuple[str, str, str, list[str]]]:
+        args = [sys.executable, "tools/recipes.py", "key", "--manifest", str(self.manifest),
+                "--parts-dir", str(self.parts_dir)]
+        if ledger:
+            args += ["--ledger", str(self.ledger)]
+        done = self.run(*args, *items)
+        rows = done.stdout.splitlines()
+        self.case.assertEqual(len(rows), len(items), done.stdout)
+        result = {}
+        for index, (line, item) in enumerate(zip(rows, items)):
+            fields = line.split("\t", 3)
+            self.case.assertEqual(len(fields), 4, line)
+            key, got_item, status, detail = fields
+            self.case.assertEqual(got_item, item)
+            parts = []
+            if status != "error":
+                self.case.assertRegex(key, r"^[0-9a-f]{64}$")
+                parts = (self.parts_dir / f"{index}.parts").read_text().splitlines()
+                self.case.assertEqual(hashlib.sha256(("\n".join(parts) + "\n").encode()).hexdigest(), key)
+                self.case.assertTrue(parts[0].startswith("v\tbloomery-gate-key "))
+            else:
+                self.case.assertEqual(key, "-")
+            result[item] = key, status, detail, parts
+        return result
+
+    def green(self, *items: str) -> dict[str, tuple[str, str, str, list[str]]]:
+        rows = self.keys(*items)
+        lines = []
+        for item, (key, status, _, parts) in rows.items():
+            self.case.assertEqual(status, "key", rows[item])
+            name = re.split(r"[@:]", item, maxsplit=1)[0]
+            lines.append(f"{key}\t{name}\t{item}\tfixture\t2026-10-21T00:00:00Z\t/private/fixture\n")
+            dst = Path(str(self.ledger) + ".parts") / f"{key}.parts"
+            dst.parent.mkdir(exist_ok=True)
+            dst.write_text("\n".join(parts) + "\n")
+        self.ledger.write_text("".join(lines))
+        self.assert_status(self.keys(*items, ledger=True), {item: "skip" for item in items})
+        return rows
+
+    def assert_status(self, rows, expected: dict[str, str]) -> None:
+        self.case.assertEqual({item: row[1] for item, row in rows.items()}, expected)
+
+    @contextlib.contextmanager
+    def changed(self, rel: str, text: str, *, restore_mtime: bool = False):
+        path = self.root / rel
+        before = path.read_bytes()
+        stat = path.stat()
+        path.write_text(text)
+        if restore_mtime:
+            self.case.assertEqual(path.stat().st_size, stat.st_size)
+            os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns))
+        try:
+            yield
+        finally:
+            path.write_bytes(before)
+            os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns))
+
+
+def package(name: str, extra: str = "") -> str:
+    return f'[package]\nname = "{name}"\nversion = "0.1.0"\nedition = "2021"\n' + extra
+
+
+def runtime_files() -> dict[str, str]:
+    return {
+        "Cargo.toml": '[workspace]\nresolver = "2"\nmembers = ["crates/app", "crates/mid", "crates/dep"]\n',
+        "justfile": (
+            "gate-called:\n    ./tools/box.sh 'cargo run -p app --bin called'\n"
+            "gate-linked:\n    ./tools/box.sh 'cargo run -p app --bin linked'\n"
+            "gate-build:\n    ./tools/box.sh 'cargo build -p app --bin called'\n"
+            "gate-wrap: gate-build\n    ./tools/box.sh 'echo wrapped'\n"
+            "gate-only: gate-build\n"
+            "gate-feature:\n    ./tools/box.sh 'cargo run -p app --bin called --features extra'\n"
+            "gate-test:\n    ./tools/box.sh 'cargo test -p app --lib'\n"
+        ),
+        "crates/app/Cargo.toml": package("app", '[features]\nextra = []\n[dependencies]\nmid = { path = "../mid" }\ndep = { path = "../dep" }\n'),
+        "crates/app/src/bin/called.rs": (
+            '#[cfg(feature = "extra")]\nmod extra;\n'
+            'fn main() {\n    let _token = dep::Token;\n    mid::run();\n'
+            '    #[cfg(feature = "extra")]\n    extra::run();\n}\n'),
+        "crates/app/src/bin/extra/mod.rs": "pub fn run() { dep::feature::run(); }\n",
+        "crates/app/src/bin/linked.rs": "fn main() { println!(\"linked\"); }\n",
+        "crates/app/src/lib.rs": '#[cfg(test)]\nmod tests {\n    #[test]\n    fn check() { dep::testonly::run(); }\n}\n',
+        "crates/mid/Cargo.toml": package("mid", '[dependencies]\ndep = { path = "../dep" }\n'),
+        "crates/mid/src/lib.rs": "pub fn run() { dep::used::run(); }\n",
+        "crates/dep/Cargo.toml": package("dep"),
+        "crates/dep/src/lib.rs": "pub mod used;\npub mod unused;\npub mod feature;\npub mod testonly;\npub struct Token;\n",
+        "crates/dep/src/used.rs": 'pub fn run() { println!("used1"); }\n',
+        "crates/dep/src/unused.rs": 'pub fn ignored() { println!("unused1"); }\n',
+        "crates/dep/src/feature.rs": 'pub fn run() { println!("feature1"); }\n',
+        "crates/dep/src/testonly.rs": 'pub fn run() { println!("test1"); }\n',
+    }
+
+
+def macro_files() -> dict[str, str]:
+    files = runtime_files()
+    files["Cargo.toml"] = files["Cargo.toml"].replace('"crates/dep"]', '"crates/dep", "crates/macros"]')
+    files.update({
+        "crates/macros/Cargo.toml": package("macros", '[lib]\nproc-macro = true\n'),
+        "crates/macros/src/lib.rs": r'''extern crate proc_macro;
+mod helper;
+use proc_macro::TokenStream;
+#[proc_macro_attribute]
+pub fn keep(_: TokenStream, item: TokenStream) -> TokenStream { helper::observe(); item }
+#[proc_macro_derive(Generated)]
+pub fn generated(_: TokenStream) -> TokenStream { helper::observe(); TokenStream::new() }
+#[proc_macro_attribute]
+pub fn startup(_: TokenStream, mut item: TokenStream) -> TokenStream {
+    item.extend(r#"#[used] #[link_section = ".init_array"]
+static START: extern "C" fn() = {
+    extern "C" fn launch() { initialize(); }
+    launch
+};"#.parse::<TokenStream>().unwrap());
+    item
+}
+#[proc_macro_attribute]
+pub fn inject(_: TokenStream, _: TokenStream) -> TokenStream {
+    "pub fn run() { crate::hidden::run(); }".parse().unwrap()
+}
+''',
+        "crates/macros/src/helper.rs": 'pub fn observe() { let _ = include_str!("../macro-input.txt"); }\n',
+        "crates/macros/macro-input.txt": "macro1\n",
+    })
+    return files
+
+
+def source_macro_files(kind: str) -> dict[str, str]:
+    files = runtime_files()
+    files["crates/dep/src/lib.rs"] += "mod hidden;\nmod startup;\n"
+    files["crates/dep/src/hidden.rs"] = 'pub fn run() { println!("source-macro1"); }\n'
+    if kind in ("exported-startup", "exported-expression", "exported-import", "exported-proc-startup"):
+        files["Cargo.toml"] = files["Cargo.toml"].replace('"crates/dep"]', '"crates/dep", "crates/macros"]')
+        files["crates/dep/Cargo.toml"] += '[dependencies]\nmacros = { path = "../macros" }\n'
+        files["crates/macros/Cargo.toml"] = package("macros")
+        files["crates/macros/src/lib.rs"] = r'''#[macro_export]
+macro_rules! boot {
+    () => {
+        #[used]
+        #[link_section = ".init_array"]
+        static START: extern "C" fn() = {
+            extern "C" fn launch() { crate::hidden::run(); }
+            launch
+        };
+    };
+}
+'''
+        files["crates/dep/src/startup.rs"] = "macros::boot!();\n"
+        if kind in ("exported-expression", "exported-import"):
+            files["crates/macros/src/lib.rs"] = '#[macro_export]\nmacro_rules! boot { () => { crate::hidden::run() }; }\n'
+            if kind == "exported-import":
+                files["crates/macros/src/lib.rs"] = '#[macro_export]\nmacro_rules! boot { () => {{ use crate::hidden::run; run(); }}; }\n'
+            files["crates/dep/src/startup.rs"] = "pub fn run() { macros::boot!(); }\n"
+            files["crates/dep/src/used.rs"] = "pub fn run() { crate::startup::run(); }\n"
+        elif kind == "exported-proc-startup":
+            files["Cargo.toml"] = files["Cargo.toml"].replace('"crates/macros"]', '"crates/macros", "crates/procgen"]')
+            files["crates/dep/Cargo.toml"] += 'procgen = { path = "../procgen" }\n'
+            files["crates/macros/src/lib.rs"] = '#[macro_export]\nmacro_rules! boot { () => { procgen::boot!(); }; }\n'
+            files["crates/procgen/Cargo.toml"] = package("procgen") + "[lib]\nproc-macro = true\n"
+            files["crates/procgen/src/lib.rs"] = r'''extern crate proc_macro;
+#[proc_macro]
+pub fn boot(_: proc_macro::TokenStream) -> proc_macro::TokenStream {
+    r#"#[used] #[link_section = ".init_array"]
+       static START: extern "C" fn() = {
+           extern "C" fn launch() { crate::hidden::run(); }
+           launch
+       };"#.parse().unwrap()
+}
+'''
+    elif kind == "local-relative-import":
+        files["crates/dep/src/lib.rs"] = ('macro_rules! boot { () => {{ use super::hidden::run; run(); }}; }\n'
+                                         + files["crates/dep/src/lib.rs"])
+        files["crates/dep/src/startup.rs"] = "pub fn run() { boot!(); }\n"
+        files["crates/dep/src/used.rs"] = "pub fn run() { crate::startup::run(); }\n"
+    else:
+        source = ('macro_rules! boot { () => { impl Drop for crate::Token {\n'
+                  '    fn drop(&mut self) { crate::hidden::run(); }\n'
+                  '} }; }\nboot!();\n')
+        files["crates/dep/src/startup.rs"] = source.replace("boot", "println") if kind == "shadowed-builtin" else source
+    return files
+
+
+class RecipeKeyCLI(unittest.TestCase):
+    def assert_file(self, row, rel: str, present: bool = True) -> None:
+        paths = {p.split("\t")[1] for p in row[3] if p.startswith(("file\t", "link\t", "manifest\t"))}
+        self.assertEqual(rel in paths, present, f"{rel}: {row}")
+
+    def test_called_versus_linked_transitive_and_ledger(self):
+        ws = Workspace(self, runtime_files())
+        items = ("gate-called", "gate-linked")
+        baseline = ws.green(*items)
+        self.assert_file(baseline[items[0]], "crates/dep/src/used.rs")
+        self.assert_file(baseline[items[1]], "crates/dep/src/used.rs", False)
+        self.assert_file(baseline[items[0]], "crates/dep/src/unused.rs", False)
+        for rel, expected in (("crates/dep/src/unused.rs", {i: "skip" for i in items}),
+                              ("crates/dep/src/used.rs", {items[0]: "run", items[1]: "skip"})):
+            with self.subTest(rel=rel), ws.changed(rel, (ws.root / rel).read_text().replace("1", "2"), restore_mtime=True):
+                rows = ws.keys(*items, ledger=True)
+                ws.assert_status(rows, expected)
+                for item in items:
+                    self.assertEqual(rows[item][0] == baseline[item][0], expected[item] == "skip")
+
+    def test_profiles_and_reached_sources_invalidate_their_consumers(self):
+        files = runtime_files()
+        files["justfile"] += (
+            "gate-profiled:\n"
+            "    BLOOMERY_MODEL=${BLOOMERY_MODEL:-qwen35moe} ./tools/box.sh 'cargo run -p app --bin called'\n"
+            "    BLOOMERY_MODEL=deepseek2 ./tools/box.sh 'cargo run -p app --bin linked'\n"
+            "gate-other-profile:\n"
+            "    BLOOMERY_MODEL=qwen3moe ./tools/box.sh 'cargo run -p app --bin linked'\n"
+        )
+        for override, selected in ((None, "qwen35moe"), ("glm5next", "glm5next")):
+            with self.subTest(inherited_profile=override):
+                ws = Workspace(self, files)
+                for profile in ("deepseek2", "qwen3moe", "qwen35moe", "glm5next"):
+                    ws.write(f"tools/ref/models/{profile}.sh", f"PROFILE_VALUE={profile}-1\n")
+                if override is not None:
+                    ws.env["BLOOMERY_MODEL"] = override
+                items = ("gate-profiled", "gate-other-profile")
+                ws.green(*items)
+                mutations = (
+                    (f"tools/ref/models/{selected}.sh", ("run", "skip")),
+                    ("tools/ref/models/deepseek2.sh", ("run", "skip")),
+                    ("tools/ref/models/qwen3moe.sh", ("skip", "run")),
+                    ("crates/dep/src/used.rs", ("run", "skip")),
+                    ("crates/dep/src/unused.rs", ("skip", "skip")),
+                )
+                for rel, verdicts in mutations:
+                    changed = (ws.root / rel).read_text().replace("1", "2")
+                    with self.subTest(input=rel), ws.changed(rel, changed, restore_mtime=True):
+                        ws.assert_status(ws.keys(*items, ledger=True), dict(zip(items, verdicts)))
+
+    def test_recipe_dependencies_and_dependency_only_wrapper(self):
+        ws = Workspace(self, runtime_files())
+        items = ("gate-build", "gate-wrap", "gate-only")
+        rows = ws.green(*items)
+        for row in rows.values():
+            self.assert_file(row, "crates/app/src/bin/called.rs")
+            self.assert_file(row, "crates/dep/src/used.rs")
+            self.assert_file(row, "crates/dep/src/unused.rs", False)
+        for rel, status in (("crates/dep/src/used.rs", "run"), ("crates/dep/src/unused.rs", "skip"),
+                            ("crates/app/src/bin/called.rs", "run")):
+            with self.subTest(rel=rel), ws.changed(rel, (ws.root / rel).read_text() + "// changed\n"):
+                ws.assert_status(ws.keys(*items, ledger=True), {i: status for i in items})
+
+    def test_new_edge_in_previously_excluded_source(self):
+        ws = Workspace(self, runtime_files())
+        baseline = ws.green("gate-called")
+        rel = "crates/dep/src/unused.rs"
+        self.assert_file(baseline["gate-called"], rel, False)
+        with ws.changed(rel, 'impl Drop for crate::Token {\n    fn drop(&mut self) { crate::used::run(); }\n}\n'):
+            rows = ws.keys("gate-called", ledger=True)
+            ws.assert_status(rows, {"gate-called": "run"})
+            self.assert_file(rows["gate-called"], rel)
+
+    def test_nonstandard_drop_impl_keeps_its_transitive_helper(self):
+        files = runtime_files()
+        files["crates/dep/src/lib.rs"] += '#[path = "drop.rust"] mod destructor;\nmod hidden;\n'
+        files["crates/dep/src/drop.rust"] = 'impl Drop for crate::Token { fn drop(&mut self) { crate::hidden::run(); } }\n'
+        files["crates/dep/src/hidden.rs"] = 'pub fn run() { println!("drop1"); }\npub fn alt() { println!("alt1"); }\n'
+        ws = Workspace(self, files)
+        items = ("gate-called", "gate-linked")
+        baseline = ws.green(*items)
+        for rel in ("crates/dep/src/drop.rust", "crates/dep/src/hidden.rs"):
+            self.assert_file(baseline["gate-called"], rel)
+            self.assert_file(baseline["gate-linked"], rel, False)
+            changed = files[rel].replace("hidden::run()", "hidden::alt()") if rel.endswith(".rust") else files[rel].replace("drop1", "drop2")
+            with self.subTest(rel=rel), ws.changed(rel, changed, restore_mtime=True):
+                ws.assert_status(ws.keys(*items, ledger=True), {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_absolute_and_raw_paths_invalidate_called_inputs(self):
+        for call, module in (("::dep::used::run()", "used"), ("dep::r#type::run()", "type")):
+            with self.subTest(call=call):
+                files = runtime_files()
+                files["crates/mid/src/lib.rs"] = f"pub fn run() {{ {call}; }}\n"
+                if module == "type":
+                    files["crates/dep/src/lib.rs"] += "pub mod r#type;\n"
+                    files["crates/dep/src/type.rs"] = 'pub fn run() { println!("raw1"); }\n'
+                ws = Workspace(self, files)
+                ws.green("gate-called", "gate-linked")
+                rel = f"crates/dep/src/{module}.rs"
+                with ws.changed(rel, files[rel].replace("1", "2"), restore_mtime=True):
+                    ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                                     {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_nonstandard_target_root_keeps_called_inputs(self):
+        files = runtime_files()
+        del files["crates/app/src/bin/called.rs"]
+        files["crates/app/Cargo.toml"] += '\n[[bin]]\nname = "called"\npath = "src/main.rust"\n'
+        files["crates/app/src/main.rust"] = "fn main() { dep::used::run(); }\n"
+        ws = Workspace(self, files)
+        ws.green("gate-called", "gate-linked")
+        rel = "crates/dep/src/used.rs"
+        with ws.changed(rel, files[rel].replace("used1", "used2"), restore_mtime=True):
+            ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                             {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_known_source_macro_startup_and_implicit_impl_inputs(self):
+        for kind in ("exported-startup", "exported-expression", "exported-import", "exported-proc-startup",
+                     "local-relative-import", "local-drop", "shadowed-builtin"):
+            with self.subTest(expansion=kind):
+                files = source_macro_files(kind)
+                ws = Workspace(self, files)
+                baseline = ws.green("gate-called", "gate-linked")
+                rel = "crates/dep/src/hidden.rs"
+                self.assert_file(baseline["gate-called"], rel)
+                self.assert_file(baseline["gate-called"], "crates/dep/src/startup.rs")
+                if kind.startswith("exported-"):
+                    self.assert_file(baseline["gate-called"], "crates/macros/src/lib.rs")
+                if kind in ("local-drop", "exported-expression"):
+                    self.assert_file(baseline["gate-called"], "crates/dep/src/unused.rs", False)
+                with ws.changed(rel, files[rel].replace("source-macro1", "source-macro2"), restore_mtime=True):
+                    ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                                     {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_features_and_test_roots(self):
+        ws = Workspace(self, runtime_files())
+        items = ("gate-called", "gate-feature", "gate-test")
+        baseline = ws.green(*items)
+        self.assertEqual(len({row[0] for row in baseline.values()}), len(items))
+        for rel, affected in (("crates/app/src/bin/extra/mod.rs", {"gate-feature"}),
+                              ("crates/dep/src/testonly.rs", {"gate-test"})):
+            with self.subTest(rel=rel), ws.changed(rel, (ws.root / rel).read_text() + "// changed\n"):
+                ws.assert_status(ws.keys(*items, ledger=True),
+                                 {i: "run" if i in affected else "skip" for i in items})
+
+    def test_includes_and_literal_directory_children(self):
+        files = runtime_files()
+        files.update({
+            "crates/dep/src/used.rs": (
+                'pub fn run() {\n    println!("{}", include_str!("../data/message.txt"));\n'
+                '    for p in std::fs::read_dir("tools/key-data").unwrap() { println!("{:?}", p); }\n}\n'),
+            "crates/dep/data/message.txt": "included\n",
+            "tools/key-data/first.txt": "child\n",
+        })
+        ws = Workspace(self, files)
+        baseline = ws.green("gate-called", "gate-linked")
+        for rel in ("crates/dep/data/message.txt", "tools/key-data/first.txt"):
+            self.assert_file(baseline["gate-called"], rel)
+            self.assert_file(baseline["gate-linked"], rel, False)
+            with self.subTest(rel=rel), ws.changed(rel, "changed\n"):
+                ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                                 {"gate-called": "run", "gate-linked": "skip"})
+        ws.write("tools/key-data/new.txt", "new child\n")
+        ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                         {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_missing_linked_source_and_include_are_named_errors(self):
+        ws = Workspace(self, runtime_files())
+        ws.green("gate-called", "gate-linked")
+        rel = "crates/dep/src/unused.rs"
+        (ws.root / rel).unlink()
+        rows = ws.keys("gate-called", "gate-linked", ledger=True)
+        ws.assert_status(rows, {"gate-called": "error", "gate-linked": "error"})
+        for row in rows.values():
+            self.assertIn("unused", row[2])
+        ws.write(rel, "pub fn ignored() {}\n")
+        with ws.changed("crates/dep/src/used.rs", 'pub fn run() { let _ = include_str!("../data/missing.txt"); }\n'):
+            row = ws.keys("gate-called", ledger=True)["gate-called"]
+            self.assertEqual(row[1], "error", row)
+            self.assertIn("missing.txt", row[2])
+
+    def test_non_git_checkout_content_invalidation(self):
+        ws = Workspace(self, runtime_files(), git=False)
+        ws.green("gate-called", "gate-linked")
+        self.assertFalse((ws.root / ".git").exists())
+        rel = "crates/dep/src/used.rs"
+        with ws.changed(rel, (ws.root / rel).read_text().replace("used1", "used2"), restore_mtime=True):
+            ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                             {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_malformed_and_foreign_cache_recompute_safely(self):
+        ws = Workspace(self, runtime_files())
+        baseline = ws.green("gate-called", "gate-linked")
+        cache = ws.home / ".cache"
+        entries = [p for p in cache.rglob("*") if p.is_file()]
+        self.assertTrue(entries, "actual key CLI must persist structural Reach state in private HOME")
+        foreign_files = runtime_files()
+        foreign_files["crates/mid/src/lib.rs"] = "pub fn run() { dep::unused::ignored(); }\n"
+        foreign = Workspace(self, foreign_files)
+        foreign.green("gate-called")
+        foreign_entries = [p for p in (foreign.home / ".cache").rglob("*") if p.is_file()]
+        self.assertTrue(foreign_entries, "second checkout must produce its own cache")
+        foreign_payload = max((p.read_bytes() for p in foreign_entries), key=len)
+        for label, payload in (("truncated", b'{"truncated":'),
+                               ("foreign-version", b'{"version":"foreign","graph":{},"identity":"foreign"}\n'),
+                               ("foreign-checkout", foreign_payload)):
+            with self.subTest(cache=label):
+                for path in entries:
+                    path.write_bytes(payload)
+                rows = ws.keys("gate-called", "gate-linked", ledger=True)
+                ws.assert_status(rows, {"gate-called": "skip", "gate-linked": "skip"})
+                for item in rows:
+                    self.assertEqual((rows[item][0], rows[item][3]), (baseline[item][0], baseline[item][3]))
+                rel = "crates/dep/src/used.rs"
+                with ws.changed(rel, (ws.root / rel).read_text().replace("used1", "used2"), restore_mtime=True):
+                    ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                                     {"gate-called": "run", "gate-linked": "skip"})
+                entries = [p for p in cache.rglob("*") if p.is_file()]
+
+    def test_real_build_script_helper_input_and_workspace_build_dependency(self):
+        files = runtime_files()
+        files["Cargo.toml"] = files["Cargo.toml"].replace('"crates/dep"]', '"crates/dep", "crates/builddep"]')
+        files["crates/dep/Cargo.toml"] += '[build-dependencies]\nbuilddep = { path = "../builddep" }\n'
+        del files["crates/app/src/lib.rs"]  # exercise an actual bin-only package's build script
+        files.update({
+            "crates/dep/build.rs": 'mod build_helper;\nfn main() { build_helper::run(); builddep::run(); }\n',
+            "crates/dep/build_helper.rs": 'pub fn run() { println!("cargo:rustc-env=KEY_INPUT={}", include_str!("build-input.txt").trim()); }\n',
+            "crates/dep/build-input.txt": "value1\n",
+            "crates/builddep/Cargo.toml": package("builddep"),
+            "crates/builddep/src/lib.rs": 'mod helper;\npub fn run() { helper::run(); }\n',
+            "crates/builddep/src/helper.rs": 'pub fn run() { println!("cargo:rustc-cfg=key_fixture"); }\n',
+            # A bin-only package executes its build script too.
+            "crates/app/build.rs": 'fn main() { println!("cargo:rustc-cfg=app_fixture"); }\n',
+        })
+        ws = Workspace(self, files)
+        items = ("gate-called", "gate-linked")
+        baseline = ws.green(*items)
+        for rel in ("crates/dep/build.rs", "crates/dep/build_helper.rs", "crates/dep/build-input.txt",
+                    "crates/builddep/src/lib.rs", "crates/builddep/src/helper.rs", "crates/app/build.rs"):
+            for row in baseline.values():
+                self.assert_file(row, rel)
+            suffix = "// changed\n" if rel.endswith(".rs") else "changed\n"
+            with self.subTest(rel=rel), ws.changed(rel, (ws.root / rel).read_text() + suffix):
+                ws.assert_status(ws.keys(*items, ledger=True), {i: "run" for i in items})
+
+    def test_real_attribute_derive_and_proc_macro_helper(self):
+        files = macro_files()
+        files["crates/app/Cargo.toml"] += 'macros = { path = "../macros" }\n'
+        files["crates/app/src/bin/called.rs"] = (
+            '#[macros::keep]\n#[derive(macros::Generated)]\nstruct Probe;\n'
+            'fn main() { let _ = Probe; mid::run(); }\n')
+        ws = Workspace(self, files)
+        ws.green("gate-linked")
+        with ws.changed("crates/macros/src/helper.rs", "pub fn observe() { panic!(); }\n"):
+            ws.assert_status(ws.keys("gate-linked", ledger=True), {"gate-linked": "run"})
+        baseline = ws.green("gate-called")
+        for rel in ("crates/macros/src/lib.rs", "crates/macros/src/helper.rs", "crates/macros/macro-input.txt"):
+            self.assert_file(baseline["gate-called"], rel)
+            suffix = "// changed\n" if rel.endswith(".rs") else "changed\n"
+            with self.subTest(rel=rel), ws.changed(rel, (ws.root / rel).read_text() + suffix):
+                ws.assert_status(ws.keys("gate-called", ledger=True), {"gate-called": "run"})
+
+    def test_script_read_keeps_data_also_named_by_unreached_rust(self):
+        files = runtime_files()
+        files["crates/dep/src/unused.rs"] = 'pub fn ignored() { let _ = "crates/dep/data/shared.txt"; }\n'
+        files["crates/dep/data/shared.txt"] = "shared1\n"
+        files["tools/read-shared.sh"] = "#!/bin/sh\ncat crates/dep/data/shared.txt\n"
+        files["justfile"] = files["justfile"].replace(
+            "gate-called:\n", "gate-called:\n    ./tools/box.sh 'bash tools/read-shared.sh'\n")
+        ws = Workspace(self, files)
+        baseline = ws.green("gate-called", "gate-linked")
+        self.assert_file(baseline["gate-called"], "crates/dep/data/shared.txt")
+        with ws.changed("crates/dep/data/shared.txt", "shared2\n", restore_mtime=True):
+            ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                             {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_procedural_startup_in_unreached_module_and_conditional_attribute(self):
+        for kind in ("startup", "nonstandard-module", "conditional"):
+            with self.subTest(expansion=kind):
+                files = macro_files()
+                files["crates/dep/Cargo.toml"] += '[dependencies]\nmacros = { path = "../macros" }\n'
+                files["crates/dep/src/lib.rs"] += "pub mod hidden;\npub mod startup;\n"
+                files["crates/dep/src/hidden.rs"] = 'pub fn run() { println!("hidden1"); }\n'
+                if kind in ("startup", "nonstandard-module"):
+                    source = "startup.rust" if kind == "nonstandard-module" else "startup.rs"
+                    if kind == "nonstandard-module":
+                        files["crates/dep/src/lib.rs"] = files["crates/dep/src/lib.rs"].replace(
+                            "pub mod startup;", '#[path = "startup.rust"] pub mod startup;')
+                    files["crates/dep/src/" + source] = '#[macros::startup]\nfn initialize() { crate::hidden::run(); }\n'
+                else:
+                    files["crates/dep/Cargo.toml"] += '[features]\ndefault = ["instrumented"]\ninstrumented = []\n'
+                    files["crates/dep/src/startup.rs"] = ""
+                    files["crates/dep/src/used.rs"] = '#[cfg_attr(feature = "instrumented", macros::inject)]\npub fn run() {}\n'
+                ws = Workspace(self, files)
+                baseline = ws.green("gate-called")
+                self.assert_file(baseline["gate-called"], "crates/dep/src/hidden.rs")
+                with ws.changed("crates/dep/src/hidden.rs", 'pub fn run() { println!("hidden2"); }\n', restore_mtime=True):
+                    ws.assert_status(ws.keys("gate-called", ledger=True), {"gate-called": "run"})
+
+    def test_negated_conditions_enum_globs_and_c_abi_callbacks_do_not_activate_unreached_code(self):
+        files = runtime_files()
+        files["crates/dep/src/used.rs"] = (
+            'enum Choice { A }\npub fn run() {\n    use Choice::*;\n    let _ = A;\n    if !(false) { println!("used1"); }\n'
+            '    while !(true) {}\n    let _callback: extern "C" fn() = callback;\n}\n'
+            'extern "C" fn callback() {}\n')
+        ws = Workspace(self, files)
+        baseline = ws.green("gate-called")
+        rel = "crates/dep/src/unused.rs"
+        self.assert_file(baseline["gate-called"], rel, False)
+        with ws.changed(rel, files[rel].replace("unused1", "unused2")):
+            ws.assert_status(ws.keys("gate-called", ledger=True), {"gate-called": "skip"})
+
+    def test_function_macro_startup_keeps_owner_and_forward_not_sibling_code(self):
+        files = macro_files()
+        files["Cargo.toml"] = files["Cargo.toml"].replace('"crates/macros"]', '"crates/macros", "crates/sibling"]')
+        files["crates/app/Cargo.toml"] += 'sibling = { path = "../sibling" }\n'
+        files["crates/mid/Cargo.toml"] += 'macros = { path = "../macros" }\n'
+        files["crates/mid/src/lib.rs"] += "mod hidden;\nmod startup;\n"
+        files["crates/mid/src/hidden.rs"] = 'pub fn run() { println!("hidden1"); }\n'
+        files["crates/mid/src/startup.rs"] = "macros::boot!();\n"
+        files["crates/macros/src/lib.rs"] += (
+            '#[proc_macro]\npub fn boot(_: TokenStream) -> TokenStream {\n'
+            '    startup(TokenStream::new(), "fn initialize() { crate::hidden::run(); }".parse().unwrap())\n}\n')
+        files["crates/sibling/Cargo.toml"] = package("sibling")
+        files["crates/sibling/src/lib.rs"] = 'pub fn unused() { println!("sibling1"); }\n'
+        ws = Workspace(self, files)
+        items = ("gate-called", "gate-linked")
+        baseline = ws.green(*items)
+        for rel in ("crates/mid/src/hidden.rs", "crates/dep/src/unused.rs"):
+            self.assert_file(baseline["gate-called"], rel)
+            with self.subTest(rel=rel), ws.changed(rel, files[rel] + "// edited\n"):
+                ws.assert_status(ws.keys(*items, ledger=True), {"gate-called": "run", "gate-linked": "skip"})
+        rel = "crates/sibling/src/lib.rs"
+        self.assert_file(baseline["gate-called"], rel, False)
+        with ws.changed(rel, files[rel].replace("sibling1", "sibling2")):
+            ws.assert_status(ws.keys(*items, ledger=True), {item: "skip" for item in items})
+
+    def test_unused_import_keeps_linked_crate_startup_inputs(self):
+        for imported in ("extern crate sibling as _;", "use sibling as _;", "use sibling::Marker as _;",
+                         "use sibling::Marker;", "use sibling::{Marker as _};", "use sibling::*;"):
+            with self.subTest(imported=imported):
+                files = macro_files()
+                files["Cargo.toml"] = files["Cargo.toml"].replace('"crates/macros"]', '"crates/macros", "crates/sibling"]')
+                files["crates/app/Cargo.toml"] += 'sibling = { path = "../sibling" }\n'
+                files["crates/app/src/bin/called.rs"] = imported + "\n" + files["crates/app/src/bin/called.rs"]
+                files["crates/sibling/Cargo.toml"] = package("sibling", '[dependencies]\nmacros = { path = "../macros" }\n')
+                files["crates/sibling/src/lib.rs"] = "pub struct Marker;\nmod hidden;\nmod startup;\n"
+                files["crates/sibling/src/hidden.rs"] = 'pub fn run() { println!("side1"); }\n'
+                files["crates/sibling/src/startup.rs"] = '#[macros::startup]\nfn initialize() { crate::hidden::run(); }\n'
+                ws = Workspace(self, files)
+                baseline = ws.green("gate-called", "gate-linked")
+                rel = "crates/sibling/src/hidden.rs"
+                self.assert_file(baseline["gate-called"], rel)
+                self.assert_file(baseline["gate-linked"], rel, False)
+                with ws.changed(rel, files[rel].replace("side1", "side2"), restore_mtime=True):
+                    ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                                     {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_crlf_and_bare_cr_sources_remain_valid_and_content_keyed(self):
+        for newline in ("\r\n", "\r"):
+            with self.subTest(newline=repr(newline)):
+                files = {path: text.replace("\n", newline) if path.endswith(".rs") else text
+                         for path, text in runtime_files().items()}
+                ws = Workspace(self, files)
+                ws.green("gate-called", "gate-linked")
+                rel = "crates/dep/src/used.rs"
+                with ws.changed(rel, files[rel].replace("used1", "used2"), restore_mtime=True):
+                    ws.assert_status(ws.keys("gate-called", "gate-linked", ledger=True),
+                                     {"gate-called": "run", "gate-linked": "skip"})
+
+    def test_analysis_code_edited_after_import_never_emits_a_skippable_key(self):
+        for relative in ("tools/recipes.py", "tools/bloomery/reach_cache.py", "tools/bloomery/fixture_dirs.py"):
+            with self.subTest(analysis=relative):
+                ws = Workspace(self, runtime_files())
+                ws.green("gate-called")
+                driver = r'''
+import contextlib
+import io
+import os
+import importlib.util
+from pathlib import Path
+import sys
+sys.path.insert(0, str(Path.cwd() / "tools"))
+spec = importlib.util.spec_from_file_location("loaded_recipes", "tools/recipes.py")
+module = importlib.util.module_from_spec(spec)
+sys.modules[spec.name] = module
+spec.loader.exec_module(module)
+from bloomery import reach_cache
+path = Path(sys.argv[1])
+before = path.read_bytes()
+state = path.stat()
+with contextlib.redirect_stdout(io.StringIO()) as baseline:
+    result = module.main(["key", "--manifest", sys.argv[2], "--ledger", sys.argv[3], "gate-called"])
+assert result == 0 and baseline.getvalue().strip().split("\t")[2] == "skip", baseline.getvalue()
+try:
+    changed = before.replace(b"_REACH_LOADED_CODE", b"_REACH_EDITED_CODE", 1)
+    assert changed != before and len(changed) == len(before)
+    path.write_bytes(changed)
+    os.utime(path, ns=(state.st_atime_ns, state.st_mtime_ns))
+    result = module.main(["key", "--manifest", sys.argv[2], "--ledger", sys.argv[3], "gate-called"])
+finally:
+    path.write_bytes(before)
+    os.utime(path, ns=(state.st_atime_ns, state.st_mtime_ns))
+raise SystemExit(result)
+'''
+                observed = ws.run(sys.executable, "-c", driver, relative, str(ws.manifest), str(ws.ledger))
+                row = observed.stdout.strip().split("\t", 3)
+                self.assertEqual(row[:3], ["-", "gate-called", "error"], observed.stdout + observed.stderr)
+                self.assertIn(relative.rsplit("/", 1)[1], row[3])
+
+    def test_module_added_between_eager_discovery_and_key_is_included_or_refused(self):
+        ws = Workspace(self, runtime_files())
+        driver = r'''
+import importlib.util
+import json
+from pathlib import Path
+import sys
+sys.path.insert(0, str(Path.cwd() / "tools"))
+spec = importlib.util.spec_from_file_location("loaded_recipes", "tools/recipes.py")
+module = importlib.util.module_from_spec(spec)
+sys.modules[spec.name] = module
+spec.loader.exec_module(module)
+side = module.make_side(str(Path.cwd()))
+module.reach_of(side.tree)
+Path("crates/dep/src/lib.rs").write_text(Path("crates/dep/src/lib.rs").read_text() + "pub mod fresh;\n")
+Path("crates/dep/src/used.rs").write_text("pub fn run() { crate::fresh::run(); }\n")
+Path("crates/dep/src/fresh.rs").write_text("pub fn run() {}\n")
+context = module.KeyContext(side, sys.argv[1])
+parts, error, never, summary = context.parts("gate-called")
+print(json.dumps({"parts": parts, "error": error, "never": never}))
+'''
+        observed = ws.run(sys.executable, "-c", driver, str(ws.manifest))
+        import json
+        result = json.loads(observed.stdout)
+        if result["error"] is not None:
+            self.assertFalse(result["parts"], result)
+        else:
+            self.assertTrue(any(part.startswith("file\tcrates/dep/src/fresh.rs\t") for part in result["parts"]), result)
+
+
+def main() -> int:
+    global SOURCE_ROOT
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--self-test", action="store_true", required=True)
+    parser.add_argument("--source-root", type=Path, default=SOURCE_ROOT,
+                        help="checkout whose production CLI/helpers are copied into each fixture")
+    args = parser.parse_args()
+    SOURCE_ROOT = args.source_root.resolve()
+    suite = unittest.defaultTestLoader.loadTestsFromTestCase(RecipeKeyCLI)
+    filters = [s for s in os.environ.get("RECIPES_KEY_TEST_FILTER", "").split(",") if s]
+    if filters:
+        suite = unittest.TestSuite(test for test in suite if any(s in test.id() for s in filters))
+        if not suite.countTestCases():
+            parser.error("RECIPES_KEY_TEST_FILTER matched no test")
+    return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tools/recipes.py b/tools/recipes.py
index 8588c78cc..589bc8521 100755
--- a/tools/recipes.py
+++ b/tools/recipes.py
@@ -85,6 +85,19 @@
   - its own text in the justfile (a recipe whose text changed is selected), and its dependencies'
     inputs.
 
+The ledger's `key` uses a separate `Graph.key_inputs` contract: the uncut structural item graph, the selected
+targets' own module/root files, mandatory build-script/build-dependency/procedural-macro compilation inputs,
+manifest views, dep-info, explicit reads and dependency recipes. Unused linked runtime modules are not ledger
+inputs. Procedural/startup attributes are inspected throughout every runtime-active owner's compiled files
+(an uncut reached item or explicit extern-crate link activates an owner). Opaque attributes/derives retain
+their owner's compiled source/literals and linked dependency closure, not unrelated reverse/sibling packages.
+This is compilation-code ownership, not arbitrary procedural-macro filesystem completeness:
+explicit/script reads remain independently keyed.
+Opaque macros retain their owner's compilation/dependency closure. Unknown generated includes and foreign
+linkage retain the broad recipe inputs conservatively.
+Reach's complete structural facts/edges are privately cached as validated JSON, content-checked against source
+bytes, enumeration, metadata and analysis code; that cache identity never enters individual ledger keys.
+
 Honest expectation: a change in crates/gpu or crates/model selects every GPU gate — that is the
 crate graph's true answer (a one-line enum change in the fault word selects them all, and it should).
 The saving is on leaf changes (serve, tokenizer, sampler, one gate binary, a runner, docs — which no gate
@@ -231,6 +244,8 @@
 from collections.abc import Callable, Iterable, Iterator
 from dataclasses import dataclass, field
 
+_REACH_LOADED_CODE = sys._getframe().f_code
+
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))  # `python3 -I` drops the script's directory
 from bloomery.fixture_dirs import FAMILY_NAME, FIXTURE_DIRS, REF_PATHS, RecipeError, fixture_table  # noqa: E402,F401
 
@@ -744,6 +759,7 @@ class TargetMeta:
     name: str
     src: str  # relative to the tree root
     required: list[str]
+    proc_macro: bool = False  # Cargo's target kind, not inferred from source spelling
 
 
 @dataclass
@@ -785,23 +801,23 @@ def find(self, kind: str, name: str) -> TargetMeta | None:
 # a file for that recipe, never a view (`ManifestUse.whole`).
 TARGET_TABLES = ("bin", "test", "bench", "example")
 Entry = tuple[str, str, str]  # (kind, name, src) of a target whose entry a view keeps
-_TOML_MEMO: dict[tuple[str, int, int], dict] = {}
+_TOML_MEMO: dict[tuple[str, str], dict] = {}
 
 
 def read_manifest(root: str, rel: str) -> dict:
-    """`rel` parsed with tomllib, memoized on its stat. A manifest that does not parse is a named error,
+    """`rel` parsed with tomllib, memoized on its bytes. A manifest that does not parse is a named error,
     never a fallback to the whole file."""
     import tomllib
 
     p = os.path.join(root, rel)
-    st = os.stat(p)
-    k = (p, st.st_size, st.st_mtime_ns)
+    with open(p, "rb") as fh:
+        data = fh.read()
+    k = (p, hashlib.sha256(data).hexdigest())
     if k not in _TOML_MEMO:
-        with open(p, "rb") as fh:
-            try:
-                _TOML_MEMO[k] = tomllib.load(fh)
-            except tomllib.TOMLDecodeError as err:
-                raise RecipeError(f"{rel} does not parse as TOML ({err}): a package manifest is read by table") from err
+        try:
+            _TOML_MEMO[k] = tomllib.loads(data.decode("utf-8"))
+        except (tomllib.TOMLDecodeError, UnicodeDecodeError) as err:
+            raise RecipeError(f"{rel} does not parse as TOML ({err}): a package manifest is read by table") from err
     return _TOML_MEMO[k]
 
 
@@ -1033,6 +1049,7 @@ class Tree:
     def __init__(self, root: str, meta: dict | None = None):
         self.root = os.path.realpath(root)
         meta = meta if meta is not None else cargo_metadata(self.root)
+        self.metadata = meta
         wroot = os.path.realpath(meta["workspace_root"])
         members = set(meta["workspace_members"])
         self.packages: dict[str, Package] = {}
@@ -1058,6 +1075,7 @@ def __init__(self, root: str, meta: dict | None = None):
                     name=t["name"],
                     src=os.path.relpath(os.path.realpath(t["src_path"]), wroot),
                     required=list(t.get("required-features") or []),
+                    proc_macro="proc-macro" in t["kind"] or "proc-macro" in t.get("crate_types", []),
                 )
                 for t in p["targets"]
             ]
@@ -1092,6 +1110,22 @@ def isdir(self, rel: str) -> bool:
             self._saw[-1].setdefault(("d", rel), ok)
         return ok
 
+    def directory_files(self, rel: str) -> set[str]:
+        """Every regular file read under an explicit directory, including ignored and symlinked children."""
+        out: set[str] = set()
+        seen: set[str] = set()
+        for base, dirs, files in os.walk(os.path.join(self.root, rel), followlinks=True):
+            real = os.path.realpath(base)
+            if real in seen:
+                dirs.clear()
+                continue
+            seen.add(real)
+            for name in files:
+                path = os.path.join(base, name)
+                if os.path.isfile(path):
+                    out.add(os.path.relpath(path, self.root))
+        return out
+
     def read(self, rel: str) -> str:
         path = os.path.join(self.root, rel)
         if not self._saw or ("r", rel) in self._saw[-1]:
@@ -1686,6 +1720,7 @@ def __init__(self, tree: Tree, recipes: dict[str, Recipe]):
         self.tree = tree
         self.recipes = recipes
         self._inputs: dict[str, RecipeInputs] = {}
+        self._key_inputs: dict[str, RecipeInputs] = {}
         self._script_cache: dict[str, dict[str, str]] = {}
 
     def script_closure(self, rel: str) -> dict[str, str]:
@@ -1723,6 +1758,20 @@ def _script_closure(self, rel: str) -> dict[str, str]:
             out.setdefault(s, f"script {rel}")
         return out
 
+    def named_reads(self, rc: RecipeCommands) -> tuple[dict[str, str], list[str]]:
+        """Script/explicit-path reads independently of Cargo's first-writer origin explanations."""
+        reads: dict[str, str] = {}
+        errors: list[str] = []
+        for s in rc.scripts + rc.paths:
+            if self.tree.exists(s):
+                for f, why in self.script_closure(s).items():
+                    reads.setdefault(f, why)
+            elif self.tree.isdir(s):
+                reads.setdefault(s.rstrip("/") + "/", "named directory")
+            elif s in rc.scripts or s.endswith(_SCRIPT_EXT):
+                errors.append(f"names {s}, which is not in the tree")
+        return reads, errors
+
     def inputs(self, name: str, _stack: tuple[str, ...] = ()) -> RecipeInputs:
         if name in self._inputs:
             return self._inputs[name]
@@ -1768,19 +1817,11 @@ def inputs(self, name: str, _stack: tuple[str, ...] = ()) -> RecipeInputs:
                 errors.append(str(err))
             if any(m is not None and "{{" in m for m in rc.models):
                 prefixes.setdefault("tools/ref/models/", "recipe parameter selects the profile")
-        for s in rc.scripts + rc.paths:
-            if self.tree.exists(s):
-                for f, why in self.script_closure(s).items():
-                    if f.endswith("/"):
-                        prefixes.setdefault(f, why)
-                    else:
-                        files.setdefault(f, why)
-                        if f in self.tree.manifest_paths:
-                            use.whole.add(f)
-            elif self.tree.isdir(s):
-                prefixes.setdefault(s.rstrip("/") + "/", "named directory")
-            elif s in rc.scripts or s.endswith(_SCRIPT_EXT):
-                errors.append(f"names {s}, which is not in the tree")
+        reads, read_errors = self.named_reads(rc)
+        errors.extend(read_errors)
+        for f, why in reads.items():
+            (prefixes if f.endswith("/") else files).setdefault(f, why)
+        use.whole |= reads.keys() & self.tree.manifest_paths
         built = {t for (_, k, t, _) in targets if k == "bin"}
         for dep in recipe.deps:
             if dep not in self.recipes:
@@ -1805,6 +1846,70 @@ def inputs(self, name: str, _stack: tuple[str, ...] = ()) -> RecipeInputs:
         self._inputs[name] = ri
         return ri
 
+    def key_inputs(self, name: str, _stack: tuple[str, ...] = ()) -> RecipeInputs:
+        """Ledger inputs, not compilation/selection inputs: uncut item reach plus mandatory reads.
+
+        Dependency recipes are closed separately: their targets are deliberately not in `inputs(name).targets`.
+        Scripts, explicit paths, manifests, build-time code and dep-info retain their conservative contracts.
+        `inputs` never calls this method; affected/narrow/release continue to use compilation ownership.
+        """
+        if name in _stack:
+            raise RecipeError(f"recipe dependency cycle: {' -> '.join(_stack + (name,))}")
+        reach = reach_of(self.tree)
+        if reach._key_snapshot is not None:
+            from bloomery import reach_cache
+
+            if not reach_cache.current(reach):
+                if _stack:
+                    raise RecipeError("source tree changed during recipe dependency analysis")
+                # A new request may follow an edit in a long-lived process. Forget derived state before
+                # beginning that request; changes during the request itself are rejected by validate().
+                self.tree._walk_cache.clear()
+                self.tree.unresolved.clear()
+                self._inputs.clear()
+                self._key_inputs.clear()
+                self._script_cache.clear()
+                reach = self.tree._reach = Reach(self.tree)
+            elif name in self._key_inputs:
+                return self._key_inputs[name]
+        reach.prepare()
+        broad = self.inputs(name)
+        reach.validate()
+        keep = reach.key_files(self, name)
+        files = {f: why for f, why in broad.files.items() if f in keep or not _linked_origin(why)}
+        prefixes = {f: why for f, why in broad.prefixes.items() if f in keep or not _linked_origin(why)}
+        if name in reach._key_wide:
+            files = {f: f"{why} (conservative: opaque generated code, macro expansion or foreign linkage)" if _linked_origin(why) else why
+                     for f, why in files.items()}
+            prefixes = {f: f"{why} (conservative: opaque generated code, macro expansion or foreign linkage)" if _linked_origin(why) else why
+                        for f, why in prefixes.items()}
+        # Origins explain the first read only: Cargo may have claimed a file also read by cat/a script.
+        # Re-union the independent read closure rather than treating that explanation as ownership.
+        reads, _ = self.named_reads(broad.commands)
+        for f, why in reads.items():
+            (prefixes if f.endswith("/") else files).setdefault(f, why)
+        for f in keep:
+            if f.endswith("/"):
+                prefixes.setdefault(f, "reached directory / mandatory compilation input")
+            elif self.tree.exists(f):
+                files.setdefault(f, "reached code / mandatory compilation input")
+        use = ManifestUse()
+        use.merge(broad.manifests)
+        errors = list(broad.errors)
+        for dep in self.recipes[name].deps:
+            if dep not in self.recipes:
+                continue
+            di = self.key_inputs(dep, _stack + (name,))
+            for f, why in di.files.items():
+                files.setdefault(f, f"{why} (via {dep})")
+            for p, why in di.prefixes.items():
+                prefixes.setdefault(p, f"{why} (via {dep})")
+            use.merge(di.manifests)
+            errors.extend(f"{dep}: {error}" for error in di.errors)
+        ri = RecipeInputs(files, prefixes, list(broad.targets), errors, broad.commands, use, broad.bare_cargo)
+        self._key_inputs[name] = ri
+        return ri
+
 
 # ----------------------------------------------------------------------------------------------
 # check
@@ -2416,7 +2521,8 @@ def cmd_targets(args: argparse.Namespace) -> int:
             raise RecipeError(f"no recipe {n}")
         ri = side.graph.inputs(n)
         tl = ", ".join(f"{p}:{k}{(' ' + t) if t else ''}{('[' + ','.join(f) + ']') if f else ''}" for p, k, t, f in ri.targets)
-        print(f"{n}\t{len(ri.files)} files\t{tl}\tscripts: {' '.join(sorted(set(ri.commands.scripts))) or '-'}\truns: {' '.join(ri.commands.runs) or '-'}")
+        ki = side.graph.key_inputs(n)
+        print(f"{n}\t{len(ri.files)} compilation files, {len(ki.files)} key files\t{tl}\tscripts: {' '.join(sorted(set(ri.commands.scripts))) or '-'}\truns: {' '.join(ri.commands.runs) or '-'}")
         for e in ri.errors:
             print(f"  error: {e}")
     return 0
@@ -2460,7 +2566,7 @@ def cmd_why(args: argparse.Namespace) -> int:
         names = list(args.recipe)
     for f in args.files:
         f = _norm(f)
-        hits = [(n, side.graph.inputs(n).match(f)) for n in names]
+        hits = [(n, side.graph.key_inputs(n).match(f)) for n in names]
         if args.recipe:
             for n, w in hits:
                 print(f"{f} in the key of {n}: " + (f"yes [{w}]" if w else "no"))
@@ -2482,10 +2588,9 @@ def cmd_why(args: argparse.Namespace) -> int:
 # key against reach: a crate a gate's targets never run code of must not move that gate's key
 # ----------------------------------------------------------------------------------------------
 
-# The crates held to it, each with why: the gates whose key closure (Graph.inputs) holds a file of the crate are
-# the gates whose targets reach a unit of it (Reach). A crate edge that no target of a gate uses (an optional
-# dependency a feature turns on only for the binaries that need it, a crate the common code does not name) is the
-# gap; a commit that touches the crate then re-runs every gate that merely links it.
+# The pinned runtime crates must not enter keys merely through Cargo linkage. Manifests remain mandatory
+# compilation inputs and are not counted as runtime code by key_holders. Build scripts/proc macros and opaque
+# generated code can legitimately remain wider than runtime reach; gaps reports that distinction honestly.
 KEY_REACH_PINS = {
     "bloomery-hf": "a Hugging Face fetch or checkpoint configuration reader: its commits move only the gates whose targets run its code "
     "(model-hf, gpu-gates under the `hf` feature, and gpu-glm5next under its `hf` feature for gate-gpu-glm5next-tp's folder test)",
@@ -2511,8 +2616,8 @@ def key_holders(side: Side, crate: str, names: list[str]) -> list[str]:
     pre = side.tree.packages[crate].dir.rstrip("/") + "/"
     out = []
     for n in names:
-        ri = side.graph.inputs(n)
-        if any(f.startswith(pre) and _linked_origin(w) for f, w in ri.files.items()) or any(
+        ri = side.graph.key_inputs(n)
+        if any(f.startswith(pre) and f not in side.tree.manifest_paths and _linked_origin(w) for f, w in ri.files.items()) or any(
             (p.startswith(pre) or pre.startswith(p)) and _linked_origin(w) for p, w in ri.prefixes.items()
         ):
             out.append(n)
@@ -2520,10 +2625,12 @@ def key_holders(side: Side, crate: str, names: list[str]) -> list[str]:
 
 
 def reach_holders(side: Side, crate: str, names: list[str]) -> list[str]:
-    """The recipes of `names` with a cargo target whose reach (Reach.gate_files) holds a file of package `crate`."""
+    """Recipes whose uncut target closure reaches package code (independent of model/profile cuts)."""
     pre = side.tree.packages[crate].dir.rstrip("/") + "/"
     r = reach_of(side.tree)
-    return [n for n in names if side.graph.inputs(n).targets and any(f.startswith(pre) for f in r.gate_files(side.graph, n))]
+    return [n for n in names if any(any(f.startswith(pre) for f in
+            (r.nodes[m][1] for m in r._close(side.graph, dep, None)))
+            for dep in recipe_closure(side.recipes, n))]
 
 
 def key_reach_gap(side: Side, crate: str, names: list[str] | None = None) -> list[str]:
@@ -3187,6 +3294,8 @@ def crate_idents(tree: Tree, pkg: str) -> set[str]:
 _FN_NAME = re.compile(r"\bfn\s+(?:r#)?([A-Za-z_][A-Za-z0-9_]*)")
 _MACRO_EXPORT = re.compile(r"#\s*\[\s*macro_export\b[^\]]*\]\s*macro_rules\s*!\s*(?:r#)?([A-Za-z_][A-Za-z0-9_]*)")
 _MACRO_CALL = re.compile(r"(?<![A-Za-z0-9_:])([A-Za-z_][A-Za-z0-9_]*)\s*!\s*[(\[{]")
+_MACRO_PATH_CALL = re.compile(r"((?:(?:r#)?[A-Za-z_]\w*\s*::\s*)*(?:r#)?[A-Za-z_]\w*)\s*!\s*[(\[{]")
+_MACRO_IMPORT = re.compile(r"\b(?:use|extern\s+crate)\b")
 _WORD = re.compile(r"(?<![A-Za-z0-9_])[A-Za-z_][A-Za-z0-9_]*")
 _PUB_BEFORE = re.compile(r"\bpub\b(?:\s*\([^)]*\))?\s*$")
 _TYPE_PATH = re.compile(r"(?<![A-Za-z0-9_'])((?:[A-Za-z_][A-Za-z0-9_]*\s*::\s*)*[A-Za-z_][A-Za-z0-9_]*)")
@@ -3287,6 +3396,76 @@ def _split_angles(text: str) -> list[str]:
     return [x.strip() for x in out if x.strip()]
 
 
+def _key_attributes(shape: str) -> tuple[bool, bool]:
+    """(opaque expansion, foreign linkage) in all attributes of lexed Rust, including cfg_attr payloads.
+
+    cfg_attr predicates are deliberately not evaluated: any applicable startup/procedural attribute can
+    execute code without an ordinary item edge. Builtin derives are the only derives resolved here.
+    """
+    standard = {"cfg", "allow", "warn", "deny", "forbid", "expect", "doc", "test", "ignore", "inline",
+                "cold", "repr", "path", "macro_export", "must_use", "non_exhaustive",
+                "track_caller", "automatically_derived", "deprecated", "should_panic", "used"}
+    builtin_derives = {"Clone", "Copy", "Debug", "Default", "Eq", "PartialEq", "Ord", "PartialOrd", "Hash"}
+    foreign = {"no_mangle", "export_name", "link", "link_name", "link_section"}
+
+    def attribute(payload: str) -> tuple[bool, bool]:
+        match = re.match(r"\s*((?:[A-Za-z_]\w*\s*::\s*)*[A-Za-z_]\w*)", payload)
+        if match is None:
+            return True, False
+        name = re.sub(r"\s+", "", match.group(1))
+        rest = payload[match.end():].strip()
+        if name in ("cfg_attr", "derive", "unsafe"):
+            if not rest.startswith("(") or _close(rest, 0) != len(rest) - 1:
+                return True, False
+            args = _split_top(rest[1:-1])
+            if name == "derive":
+                return any(arg not in builtin_derives for arg in args), False
+            children = args[1:] if name == "cfg_attr" else args
+            flags = [attribute(child) for child in children]
+            return any(a for a, _ in flags), any(b for _, b in flags)
+        return name not in standard and name not in foreign, name in foreign
+
+    opaque = linkage = False
+    for match in re.finditer(r"#\s*!?\s*\[", shape):
+        end = _close(shape, match.end() - 1)
+        if end is None:
+            opaque = True
+            continue
+        a, b = attribute(shape[match.end():end])
+        opaque |= a
+        linkage |= b
+    return opaque, linkage
+
+
+@dataclass(frozen=True)
+class KeySyntax:
+    extern_crates: frozenset[str]
+    opaque: bool
+    foreign: bool
+    macro_calls: frozenset[str]
+    qualified: frozenset[tuple[str, ...]]
+    conditional: bool
+
+
+@functools.cache
+def _key_syntax(text: str) -> KeySyntax:
+    """File-local syntax only; bindings, macro resolution and active owners remain graph-specific."""
+    code, shape = rust_lex(text)
+    opaque, linkage = _key_attributes(shape)
+    calls = frozenset(match.group(1) for match in _MACRO_PATH_CALL.finditer(shape))
+    return KeySyntax(
+        frozenset(re.findall(r"\bextern\s+crate\s+([A-Za-z_]\w*)", shape)),
+        opaque or bool(re.search(r"(?<![\w:>\]])::|r#[A-Za-z_]", shape)),
+        linkage or bool(re.search(r'\bextern\s*(?:"[^"]*"\s*)?\{', code))
+        or bool(re.search(r'\binclude!\s*\(\s*(?!")\S', code)
+                or re.search(r'\binclude!\s*\(\s*"[^"]*(?<!\.rs)"', code)),
+        calls,
+        frozenset(tuple(part.removeprefix("r#") for part in re.split(r"\s*::\s*", call))
+                  for call in calls if "::" in call),
+        bool(re.search(r"#\s*!?\s*\[\s*cfg(?:_attr)?\b", shape)),
+    )
+
+
 def _type_names(text: str, params: frozenset[str]) -> tuple[tuple[str, ...] | None, tuple[tuple[str, ...], ...]]:
     """(head path, the other paths) of a type or trait as an impl header writes it; a generic parameter, a lifetime
     or a keyword is no path, and a head that is one leaves the head None."""
@@ -3481,6 +3660,15 @@ def _assoc_units(shape: str, a: int, b: int, ctx: tuple[str, ...], hdr: tuple[in
         i = end
 
 
+@functools.cache
+def _source_macro_bodies(text: str, test: bool) -> dict[int, str]:
+    """Lexed definition bodies, indexed exactly like rust_facts, for expansion safety checks."""
+    _, shape = build_view(text, test)
+    raw: list = []
+    _module_units(shape, 0, len(shape), (), raw, [])
+    return {k: shape[a:b] for k, (kind, _, _, a, b, _, _) in enumerate(raw) if kind == "macro_rules!"}
+
+
 @functools.cache
 def rust_facts(text: str, test: bool) -> RustFacts:
     """The facts the reach reads from one file, as a build with cfg(test) = `test` compiles it (build_view)."""
@@ -3720,6 +3908,10 @@ def __init__(self, tree: Tree):
         self._archs: dict[tuple[str, str], tuple[frozenset[str] | None, dict[str, str]]] = {}
         self._dropped: dict[tuple[str, str], frozenset[int]] = {}
         self._dir_archs: dict[str, set[str]] | None = None
+        self._facts: dict[tuple[str, bool], RustFacts] = {}
+        self._key_snapshot = None
+        self._key_observed = {}
+        self._key_wide: set[str] = set()
 
     def text(self, f: str) -> str:
         if f not in self._text:
@@ -3727,7 +3919,10 @@ def text(self, f: str) -> str:
         return self._text[f]
 
     def facts(self, rk: RootKey, f: str) -> RustFacts:
-        return rust_facts(self.text(f), rk[2])
+        key = (f, rk[2])
+        if key not in self._facts:
+            self._facts[key] = rust_facts(self.text(f), rk[2])
+        return self._facts[key]
 
     def mods(self, rk: RootKey) -> dict[tuple[str, ...], str]:
         if rk not in self._mods:
@@ -3785,8 +3980,8 @@ def doctests(self, pkg: str) -> str:
         return path
 
     def units(self, rk: RootKey, f: str) -> list[Hit]:
-        """Every unit of file `f` in root `rk`; a file with no Rust items (a data file, an empty module) as -1."""
-        if not f.endswith(".rs"):
+        """Every unit of a Rust module; a non-source data file uses index -1."""
+        if not f.endswith(".rs") and f not in self.mods(rk).values():
             return [(rk, f, -1)]
         n = len(self.facts(rk, f).units)
         return [(rk, f, i) for i in range(n)] or [(rk, f, -1)]
@@ -4072,8 +4267,6 @@ def root_rules(self, rk: RootKey, files: list[str]) -> list[tuple[int, str, str,
         name some reached unit must say, None for an item std calls unnamed; rule 1)."""
         out = []
         for f in files:
-            if not f.endswith(".rs"):
-                continue
             facts = self.facts(rk, f)
             bases = self.bases(rk, f) or [()]
             sides: dict[ImplHead, tuple[list[list[int]], bool]] = {}
@@ -4320,7 +4513,8 @@ def gate(self, graph: Graph, n: str) -> dict[int, tuple[int, str]]:
         self._gates[key] = {**full, **cut}
         return self._gates[key]
 
-    def _close(self, graph: Graph, n: str, open_rows: frozenset[int] | None) -> dict[int, tuple[int, str]]:
+    def _close(self, graph: Graph, n: str, open_rows: frozenset[int] | None,
+               extra: set[Hit] | None = None) -> dict[int, tuple[int, str]]:
         """The closure of gate `n`'s targets; `open_rows`, when given, are the only rows of the registry's table whose
         entries its dispatch edges lead to, until a unit that walks every row is reached."""
         self.dispatches()
@@ -4368,11 +4562,14 @@ def fire(rs: RuleSet, i: int) -> None:
                 own_set = self.own_rules(rk)
                 if own_set.rules and own_set not in sets:
                     sets.append(own_set)
-            own = [h for f in sorted(self.tree.tree_of(t.src, test)[0]) for h in self.units(rk, f)]
+            own = [h for f in sorted(self.tree.tree_of(t.src, test)[0]) if f in linked for h in self.units(rk, f)]
             # a test harness runs its #[test] fns, and they reach the rest; a target with none runs its main
             entries = [h for h in own if kind in ("libtest", "test") and h[2] >= 0 and self.facts(h[0], h[1]).units[h[2]].test]
             for h in entries or own:
                 seed(h, label)
+        for h in extra or ():
+            if h[1] in linked:
+                seed(h, "source macro compilation effect")
         builds = {self.tree.packages[bpkg].name: src for src, bpkg in self.builds.items() if src in linked}
         names = frozenset().union(*(rs.names for rs in sets))
         for rs in sets:
@@ -4432,6 +4629,222 @@ def gate_files(self, graph: Graph, n: str) -> dict[str, int]:
             self._gate_files[key] = out
         return self._gate_files[key]
 
+    def prepare(self) -> None:
+        """Materialize and persist the structural graph, never recipe/profile-specific cuts.
+
+        All target roots (including excluded-by-feature modules and test views) are discovered before publication.
+        Closure, environment and architecture decisions remain lazy and are never serialized as structural facts.
+        """
+        if self._key_snapshot is not None:
+            return
+        from bloomery import reach_cache
+
+        reach_cache.prepare(self)
+
+    def validate(self) -> None:
+        """Reject source movement during analysis/key assembly rather than publish a mixed-state green key."""
+        if self._key_snapshot is not None:
+            from bloomery import reach_cache
+
+            if not reach_cache.current(self):
+                raise RecipeError("source tree changed during reach analysis; request a fresh key")
+
+    def key_files(self, graph: Graph, n: str) -> set[str]:
+        """Uncut structural reach and mandatory compilation reads for the recipe's own targets.
+
+        Unlike gate_files this is not a runtime/profile allowlist. A selected target owns its complete module
+        tree; every linked package still owns its manifest/build script, and build dependencies and proc macros
+        retain their complete compile-time closures even if no runtime call reaches that package.
+        """
+        ri, tree = graph.inputs(n), self.tree
+        par = self._close(graph, n, None)
+        keep: set[str] = set()
+        keep |= {f for f in ri.files if f in tree.manifest_paths or f in CARGO_GLOBALS or f in OXIDE_GLOBALS
+                 or ri.files[f].startswith("dep-info")}
+        linked_packages = {p.name for p in tree.packages.values() if p.manifest in ri.files}
+        boundary = set(ri.files) | set(ri.prefixes)
+        compiled: dict[str, set[str]] = {p: set() for p in linked_packages}
+        reads: dict[str, set[str]] = {p: set() for p in linked_packages}
+        views: dict[tuple[str, str], set[RootKey]] = {}
+
+        def compile_root(pname: str, src: str, test: bool = False) -> set[str]:
+            files, lits = tree.tree_of(src, test)
+            files = files & boundary
+            for f in files:
+                compiled[pname].add(f)
+                views.setdefault((pname, f), set()).add((pname, src, test))
+            found = files | set(tree._resolve_literals(tree.packages[pname], [(f, lit) for f, lit in lits if f in files]))
+            reads[pname].update(found)
+            return found
+
+        def dependency_closure(owners: set[str]) -> set[str]:
+            # Follow every edge within the already linked boundary. all_features on just the first owner
+            # does not enable a transitive package's optional edges; Cargo may have unified those elsewhere.
+            closed = set(owners)
+            work = list(owners)
+            while work:
+                for dep in tree.packages[work.pop()].deps:
+                    if dep.pkg in linked_packages and dep.pkg not in closed:
+                        closed.add(dep.pkg)
+                        work.append(dep.pkg)
+            return closed
+
+        mandatory: set[str] = set()
+        for pname in linked_packages:
+            pkg = tree.packages[pname]
+            lib = pkg.lib()
+            if lib is not None:
+                compile_root(pname, lib.src)
+            if any(t.proc_macro for t in pkg.targets):
+                mandatory.add(pname)
+            mandatory.update(dep.pkg for dep in pkg.deps if dep.kind == "build" and dep.pkg in linked_packages)
+            for target in pkg.targets:
+                if target.kind == "build":
+                    keep |= compile_root(pname, target.src)
+        for pname, kind, tname, _ in ri.targets:
+            pkg = tree.packages[pname]
+            target = pkg.lib() if kind in ("lib", "libtest", "doctest") else pkg.find(kind, tname or "")
+            if target is not None:
+                keep |= compile_root(pname, target.src, kind not in ("lib", "doctest"))
+        for pname in dependency_closure(mandatory):
+            keep |= reads[pname]
+        safe_macros = {"assert", "assert_eq", "assert_ne", "debug_assert", "debug_assert_eq", "debug_assert_ne",
+                       "println", "print", "eprintln", "eprint", "format", "format_args", "panic", "unreachable",
+                       "unimplemented", "todo", "vec", "matches", "write", "writeln", "dbg", "concat", "stringify",
+                       "env", "option_env", "file", "line", "column", "module_path", "cfg", "include", "include_str",
+                       "include_bytes", "compile_error"}
+        # `if !(...)` and `while !(...)` are negated expressions, not calls to macros named `if`/`while`.
+        syntax_words = {"if", "while", "return", "break", "match", "let", "else", "for", "loop",
+                        "yield", "continue", "move", "async", "await", "unsafe", "const"}
+        opaque_owners: set[str] = set()
+        outside_ownership = False
+        # An unused --extern rlib is a compilation input, not necessarily part of the executable. Start
+        # with uncut reached owners, then honor explicit side-effect links even without a called item.
+        # Within each runtime-active owner scan *all* compilation files: an unreachable ctor can run.
+        active_owners = {self.nodes[m][0][0] for m in par if self.nodes[m][2] >= 0}
+        work = list(active_owners)
+        source_seeds: set[Hit] = set()
+        closed_seeds: set[Hit] = set()
+        exported_names = {name for p in linked_packages for name, hits in self.exported(p).items() if hits}
+        while True:
+            while work:
+                pname = work.pop()
+                pkg = tree.packages[pname]
+                for f in compiled[pname]:
+                    syntax = _key_syntax(self.text(f))
+                    facts_by_root = [(rk, self.facts(rk, f)) for rk in views[pname, f]]
+                    links = set(syntax.extern_crates)
+                    # Even an unused import can make rustc link a crate and run its constructors.
+                    links.update(path[0].removeprefix("r#")
+                                 for _, facts in facts_by_root for _, path, _, _ in facts.binds
+                                 if path and path[0] not in ("crate", "self", "super"))
+                    for dep in pkg.deps:
+                        if dep.pkg not in linked_packages or dep.pkg in active_owners:
+                            continue
+                        lib = tree.packages[dep.pkg].lib()
+                        if dep.key.replace("-", "_") in links or (lib is not None and lib.name.replace("-", "_") in links):
+                            active_owners.add(dep.pkg)
+                            work.append(dep.pkg)
+                    opaque = syntax.opaque
+                    outside_ownership |= syntax.foreign
+                    calls, qualified_paths = syntax.macro_calls, syntax.qualified
+                    qualified_known: set[tuple[RootKey, tuple[str, ...]]] = set()
+                    # Source macros use the existing unhygienic expansion edges, even from otherwise
+                    # unreached items: an expansion may emit a constructor or an implicit trait impl.
+                    for rk, facts in facts_by_root:
+                        known = set(self.macros(rk)) | exported_names
+                        unresolved: set[tuple[str, ...]] = set()
+                        for k, unit in enumerate(facts.units):
+                            if any(name in known for name in unit.macro_calls):
+                                source_seeds.add((rk, f, k))
+                            # Definition-site bindings are not copied into the invocation context.
+                            for name in unit.macro_calls:
+                                for g, i in self.macros(rk).get(name, ()):
+                                    body = _source_macro_bodies(self.text(g), rk[2]).get(i)
+                                    if body is None or _MACRO_IMPORT.search(body):
+                                        opaque = True
+                            for path in set(unit.sites) & qualified_paths:
+                                hits = [h for base in self.bases(rk, f) or [()]
+                                        for h in self.resolve(rk, base + unit.ctx, path) if h[2] >= 0]
+                                if hits and all(h[0][0] in linked_packages and
+                                                self.facts(h[0], h[1]).units[h[2]].kind == "macro_rules!" for h in hits):
+                                    source_seeds.add((rk, f, k))
+                                    qualified_known.add((rk, path))
+                                    for h in hits:
+                                        body = _source_macro_bodies(self.text(h[1]), h[0][2]).get(h[2])
+                                        # Imports and nested expansions may emit names absent from source edges.
+                                        # Bound the invoker, not an unrelated item elsewhere in the macro crate.
+                                        if body is None or _MACRO_IMPORT.search(body) or any(_key_attributes(body)) or any(
+                                                match.group(1).removeprefix("r#") not in syntax_words
+                                                for match in _MACRO_PATH_CALL.finditer(body)):
+                                            opaque = True
+                                else:
+                                    unresolved.add(path)
+                        qualified_known.difference_update((rk, path) for path in unresolved)
+                    # Only unconditionally named local enums are proven to import variants, never macros.
+                    # Imported/conditional paths remain opaque: a different binding may export a macro.
+                    glob_shadow = False
+                    conditional_bindings = syntax.conditional
+                    for rk, facts in facts_by_root:
+                        for ctx, path in facts.globs:
+                            hits = [h for base in self.bases(rk, f) or [()]
+                                    for h in self.walk(rk, base + ctx, path[:-1], True)
+                                    if h[2] >= 0] if len(path) == 2 and path[0] not in facts.bound and not conditional_bindings else []
+                            if not hits or any(self.facts(h[0], h[1]).units[h[2]].kind != "enum" for h in hits):
+                                glob_shadow = True
+                    for call in calls:
+                        spelling = re.split(r"\s*::\s*", call)[-1]
+                        if spelling in syntax_words:
+                            continue
+                        macro = spelling.removeprefix("r#")
+                        shadowed = glob_shadow or any(macro in facts.bound for _, facts in facts_by_root)
+                        local_source = all(macro in self.macros(rk) for rk, _ in facts_by_root)
+                        path = tuple(part.removeprefix("r#") for part in re.split(r"\s*::\s*", call))
+                        if "::" in call and all((rk, path) in qualified_known for rk, _ in facts_by_root):
+                            continue
+                        if "::" not in call and not shadowed and (macro in safe_macros or local_source):
+                            continue
+                        opaque = True
+                    if opaque:
+                        opaque_owners.add(pname)
+                        # Opaque expansion can name its package's forward dependencies, not unrelated
+                        # reverse/sibling packages. Inspect activated owners for linkage obligations too.
+                        added = dependency_closure({pname}) - active_owners
+                        active_owners.update(added)
+                        work.extend(added)
+            if source_seeds == closed_seeds:
+                break
+            par = self._close(graph, n, None, source_seeds)
+            closed_seeds = set(source_seeds)
+            added = {self.nodes[m][0][0] for m in par if self.nodes[m][2] >= 0} - active_owners
+            active_owners.update(added)
+            work.extend(added)
+        keep |= {self.nodes[m][1] for m in par}
+        # Module declarations/re-exports on the path to reached code are behavior, not merely linkage.
+        for rk, f in {(self.nodes[m][0], self.nodes[m][1]) for m in par}:
+            mods = self.mods(rk)
+            for path in self.bases(rk, f):
+                for depth in range(len(path) + 1):
+                    ancestor = mods.get(path[:depth])
+                    if ancestor in ri.files:
+                        keep.add(ancestor)
+        for m in par:
+            rk, f, k = self.nodes[m]
+            if k < 0:
+                continue
+            unit = self.facts(rk, f).units[k]
+            pkg = tree.packages[rk[0]]
+            keep.update(tree._resolve_literals(pkg, [(f, lit) for lit in unit.literals]))
+        # Macro expansion can name code in its own package and linked dependencies, but not an
+        # unrelated reverse/sibling package through ordinary Cargo paths. Named data/script reads are
+        # closed independently by Graph.key_inputs; arbitrary proc-macro I/O is outside this guarantee.
+        for pname in dependency_closure(opaque_owners):
+            keep |= reads[pname]
+        if outside_ownership:
+            self._key_wide.add(n)
+            keep |= set(ri.files) | set(ri.prefixes)
+        return keep
+
     def reason(self, graph: Graph, n: str, m: int) -> str:
         """Why gate `n` keeps unit `m`: a static chain, a registry row its model files name (with the signal that named it),
         or a gate that cannot tell its files / walks the whole table, which keeps every row."""
@@ -7382,11 +7795,11 @@ def cmd_orphan_tests(args: argparse.Namespace) -> int:
 #   v         the key format
 #   recipe    the text of the recipe and of every recipe it depends on (just's own parse: a comment
 #             edit in the justfile does not move it), and the justfile's settings and assignments
-#   scope     `closure` — the files below come from the walk `affected` uses (crate graph, module
-#             trees, the scripts the recipe names and what they name; cfg-ignoring, so a superset) —
+#   scope     `closure` — uncut target-code reach plus mandatory compilation inputs (own module trees,
+#             build/proc-macro closures, manifests, dep-info), scripts/explicit reads and recipe dependencies —
 #             or `tree`, every file tools/box.sh ships, with the reason: a command run on the Mac, a
 #             cargo subcommand the parser does not model (`cargo fmt`), a script that walks the tree
-#   file      path relative to the tree, content sha256 (a link: its target)
+#   file      path relative to the tree, content sha256 (a link: its spelling and referent content)
 #   manifest  a package manifest the closure reads by table: path, sha256 of the item's view of it
 #             (manifest_view: the shared tables and its own targets' entries, canonical JSON — a comment
 #             or another target's entry does not move it); one read whole is a `file` part
@@ -7405,7 +7818,7 @@ def cmd_orphan_tests(args: argparse.Namespace) -> int:
 #             fixture directory (KeyContext.fixture_dirs; every file under it)
 # Not in it: the binary (its paths are per track) and the commit.
 
-KEY_VERSION = "bloomery-gate-key 1"
+KEY_VERSION = "bloomery-gate-key 2"
 MANIFEST_VERSION = "box-manifest 1"
 MAC_ENV = ("BLOOMERY_BOX", "BLOOMERY_CARD", "BLOOMERY_DATA", "BLOOMERY_FIXTURE_VARIANT", "BLOOMERY_MODEL", "BLOOMERY_REF_MODEL", "BLOOMERY_TIER", "BLOOMERY_V41_MODEL")
 # The fixture tier's files (tools/ref/ref-paths.sh: one directory a family under it). Its rows belong to the keys of fixture-tier
@@ -8233,6 +8646,7 @@ def __init__(self, side: Side, manifest_path: str | None, box_env: str = "", env
             self.manifest_error = f"the box manifest failed: {manifest_error}"
         self.settings = settings if settings is not None else justfile_settings(os.path.join(side.tree.root, "justfile"))
         self._sha: dict[str, str] = {}
+        self._links: dict[str, str | None] = {}
         self._lit: dict[str, frozenset[str]] = {}
         self._tree: list[str] | None = None
         self._fixture_table: tuple[dict[str, str], str] | None = None
@@ -8244,11 +8658,13 @@ def tree_files(self) -> list[str]:
 
     def file_part(self, rel: str) -> str:
         p = os.path.join(self.side.tree.root, rel)
-        if os.path.islink(p):
-            return f"link\t{rel}\t{os.readlink(p)}"
-        if rel not in self._sha:
-            self._sha[rel] = sha256_file(p)
-        return f"file\t{rel}\t{self._sha[rel]}"
+        digest = sha256_file(p)
+        self._sha[rel] = digest
+        spelling = os.readlink(p) if os.path.islink(p) else None
+        self._links[rel] = spelling
+        if spelling is not None:
+            return f"link\t{rel}\t{spelling}\t{digest}"
+        return f"file\t{rel}\t{digest}"
 
     def view_part(self, rel: str, wants: frozenset[Entry]) -> str:
         """The part of a manifest read by table: the sha256 of the item's view of it."""
@@ -8347,7 +8763,10 @@ def globals(self) -> list[str]:
         out = []
         for g in names:
             p = os.path.join(root, g)
-            out.append(f"global\t{g}\t{sha256_file(p) if os.path.isfile(p) else 'absent'}")
+            digest = sha256_file(p) if os.path.isfile(p) else "absent"
+            if digest != "absent":
+                self._sha[g] = digest
+            out.append(f"global\t{g}\t{digest}")
         toml = os.path.join(root, "Cargo.toml")
         out.append(f"global\toxide-rev\t{oxide_rev(self.side.tree.read('Cargo.toml')) if os.path.isfile(toml) else 'absent'}")
         return out
@@ -8403,7 +8822,9 @@ def never(self, names: list[str], ri: RecipeInputs, envs: list[str]) -> str | No
             hit = scan_file(self.side.tree.root, f, IK_READ)
             if hit:
                 return f"{hit} reads a reference tree, which is outside the key"
-        for f in sorted(ri.files):
+        # Safety detectors deliberately inspect the broad compilation closure, not just ledger reach.
+        detector_files = {f for n in names for f in self.side.graph.inputs(n).files}
+        for f in sorted(detector_files):
             if f.endswith(".rs") and self.side.tree.exists(f):
                 hit = scan_file(self.side.tree.root, f, HOME_LITERAL)
                 if hit:
@@ -8455,14 +8876,29 @@ def parts(self, item: str) -> tuple[list[str], str | None, str | None, str]:
         if self.manifest_error:
             return [], self.manifest_error, None, ""
         try:
-            ri = graph.inputs(name)
+            ri = graph.key_inputs(name)
         except (OSError, RecipeError) as err:
             return [], f"{name}: its inputs cannot be read: {err}", None, ""
         if ri.errors:
             return [], f"{name}: {ri.errors[0]}", None, ""
         names = recipe_closure(recipes, name)
         try:
-            return self._parts(name, names, ri, envs, argv)
+            self._sha.clear()
+            self._links.clear()
+            self._directories = {}
+            result = self._parts(name, names, ri, envs, argv)
+            for rel, digest in self._sha.items():
+                if sha256_file(os.path.join(tree.root, rel)) != digest:
+                    raise RecipeError(f"input changed during key assembly: {rel}")
+            for rel, spelling in self._links.items():
+                path = os.path.join(tree.root, rel)
+                if (os.readlink(path) if os.path.islink(path) else None) != spelling:
+                    raise RecipeError(f"link changed during key assembly: {rel}")
+            for prefix, files in self._directories.items():
+                if tree.directory_files(prefix) != files:
+                    raise RecipeError(f"directory changed during key assembly: {prefix}")
+            reach_of(tree).validate()
+            return result
         except (OSError, RecipeError) as err:
             return [], f"{name}: an input cannot be read: {err}", None, ""
 
@@ -8484,7 +8920,8 @@ def _parts(self, name: str, names: list[str], ri: RecipeInputs, envs: list[str],
                     if tree.exists(rel):
                         named.add(rel)
                     elif tree.isdir(rel):
-                        named |= {f for f in self.tree_files() if f.startswith(rel.rstrip("/") + "/")}
+                        found = self._directories[rel] = tree.directory_files(rel)
+                        named |= found
         parts = [f"v\t{KEY_VERSION}"]
         parts += [f"recipe\t{n}\t{sha256_text(recipes[n].text)}" for n in names]
         parts.append(f"recipe\t<justfile>\t{sha256_text(json.dumps(self.settings, sort_keys=True))}")
@@ -8494,7 +8931,8 @@ def _parts(self, name: str, names: list[str], ri: RecipeInputs, envs: list[str],
         else:
             files = set(ri.files)
             for p in ri.prefixes:
-                files |= {f for f in self.tree_files() if f.startswith(p)}
+                found = self._directories[p] = tree.directory_files(p)
+                files |= found
         for n in names:
             rc = self.side.graph.inputs(n).commands
             files |= {f"tools/ref/models/{profile}.sh" for profile in rc.profiles(self.env)}
@@ -8502,7 +8940,7 @@ def _parts(self, name: str, names: list[str], ri: RecipeInputs, envs: list[str],
         files -= set(KEY_GLOBALS)
         if scope == "closure":
             for u in tree.unresolved:
-                if u.split(": ", 1)[0] in files:
+                if u.split(": ", 1)[0] in self.side.graph.inputs(name).files:
                     return [], f"source walk: {u}", None, ""
         for rel in sorted(files):
             # a package manifest the closure reads by table is keyed on that view (manifest_view, the reader
@@ -10763,26 +11201,11 @@ def oxide_rev_self_test(expect) -> None:
 
 
 def key_reach_self_test(expect, side: Side) -> None:
-    """The pins of KEY_REACH_PINS on the real tree, FAIL-first on a copy of it whose gpu-gates links hf for every binary,
-    and the chains `why --recipe` prints."""
-    import copy
-
+    """The runtime key/reach pins and genuine Cargo-linkage explanation chains on the real tree."""
+    # PIN(2026-10-22): Drop the plain-link hf mutation: v2 keys exclude uncalled runtime code.
     for crate, reason in KEY_REACH_PINS.items():
         gap = key_reach_gap(side, crate)
         expect(not gap, f"{crate} ({reason}): {len(gap)} recipes key it without reaching it: {' '.join(gap[:8])}")
-    # FAIL-first: gpu-gates' optional hf dependency made plain is the edge the `hf` feature cuts. Every recipe that builds a
-    # gate binary then keys hf (gate-gpu-p1 and gate-gpu-ds41-attn call neither the fetch nor a checkpoint reader), and the
-    # measure names them.
-    tree2 = copy.copy(side.tree)
-    packages = dict(side.tree.packages)
-    gates = copy.copy(packages["bloomery-gpu-gates"])
-    gates.deps = [dataclasses.replace(d, optional=False) if d.pkg == "bloomery-hf" else d for d in gates.deps]
-    packages["bloomery-gpu-gates"] = gates
-    tree2.packages = packages
-    side2 = Side(tree2, side.recipes, Graph(tree2, side.recipes))
-    probe = ["gate-gpu-p1", "gate-gpu-ds41-attn", "gate-hf"]
-    expect(sorted(key_holders(side2, "bloomery-hf", probe)) == sorted(probe), f"FAIL-first: hf is not in every probe's key with a plain gpu-gates edge: {key_holders(side2, 'bloomery-hf', probe)}")
-    expect(key_reach_gap(side2, "bloomery-hf", probe) == ["gate-gpu-p1", "gate-gpu-ds41-attn"], f"FAIL-first: the gap of a plain gpu-gates edge is {key_reach_gap(side2, 'bloomery-hf', probe)}")
     # why --recipe: a recipe that links the crate names the chain from its target, one that does not names none
     expect(why_chains(side, "gate-gpu-p1", "crates/hf/src/lib.rs") == [], f"why: gate-gpu-p1 links hf: {why_chains(side, 'gate-gpu-p1', 'crates/hf/src/lib.rs')}")
     chain = why_chains(side, "gate-gpu-glm5next-twocard", "crates/hf/src/lib.rs")
@@ -10799,6 +11222,16 @@ def key_reach_self_test(expect, side: Side) -> None:
 def reach_self_test(expect, side: Side, row, stats: dict[str, int]) -> None:
     """The derivation (Reach) on a synthetic workspace, one file per rule it decides, and the table's check of it on
     the real tree: a draft row that leaves out a gate reaching its files through another lib, and the exceptions."""
+    for source, want in (
+        ("#[derive(Clone, Debug, PartialEq)] struct S;", (False, False)),
+        ('// #[ctor::ctor]\nconst S: &str = "#[ctor::ctor]";', (False, False)),
+        ("#[ctor::ctor] fn startup() {}", (True, False)),
+        ('#[cfg_attr(feature = "boot", ctor::ctor)] fn startup() {}', (True, False)),
+        ('#[cfg_attr(unix, cfg_attr(feature = "boot", ctor::ctor))] fn startup() {}', (True, False)),
+        ('#[cfg_attr(unix, derive(Debug, custom::Error))] enum E {}', (True, False)),
+        ('#[unsafe(link_section = ".init_array")] static INIT: fn() = boot;', (False, True)),
+    ):
+        expect(_key_attributes(rust_lex(source)[1]) == want, f"key attributes misclassify {source!r}")
     with tempfile.TemporaryDirectory(prefix="recipes-reach-") as tmp:
         files = {
             "base/Cargo.toml": "",
@@ -10821,13 +11254,19 @@ def reach_self_test(expect, side: Side, row, stats: dict[str, int]) -> None:
             "base/src/rows.rs": (
                 "pub const LIST: &str = include_str!(\"../data/list.txt\");\npub const IDLE: &str = include_str!(\"../data/idle.txt\");\n"
                 "pub const TABLE: &str = \"../tools/table.tsv\";\npub const HELD: &str = \"../tools/lib-held\";\n"
+                "pub const UNUSED: &str = \"../tools/idle.tsv\";\npub const UNUSED_DIR: &str = \"../tools/idle-held\";\n"
                 + "".join(f"pub fn {f}() -> &'static str {{\n    {c}\n}}\n" for f, c in (("list", "LIST"), ("idle", "IDLE"), ("table", "TABLE"), ("held", "HELD")))
             ),
             "mid/Cargo.toml": "",
             "mid/src/lib.rs": (
+                "mod idle;\n"
                 "pub fn call() -> u32 {\n    base::b::run();\n    base::mk!();\n    base::rows::list();\n    base::rows::table();\n"
                 "    base::rows::held();\n    base::g::K\n}\n"
             ),
+            "mid/src/idle.rs": "pub fn idle() {}\n",
+            "tools/shared-read.sh": (
+                'cat tools/idle.tsv\nfind "$ROOT/tools/idle-held" -type f\nfind "$ROOT/base" -name Cargo.toml\n'
+            ),
             "app/Cargo.toml": "",
             "app/src/lib.rs": "",
             "app/tests/it.rs": (
@@ -10839,7 +11278,8 @@ def reach_self_test(expect, side: Side, row, stats: dict[str, int]) -> None:
         }
         for x in "abcdefh":
             files[f"base/src/{x}.rs"] = "pub fn run() {}\n"
-        for rel in ("base/data/list.txt", "base/data/idle.txt", "tools/table.tsv", "tools/lib-held/b.ids", "tools/own-held/a.ids"):
+        for rel in ("base/data/list.txt", "base/data/idle.txt", "tools/table.tsv", "tools/lib-held/b.ids",
+                    "tools/own-held/a.ids", "tools/idle.tsv", "tools/idle-held/extra.txt"):
             files[rel] = "x\n"
         for rel, text in files.items():
             os.makedirs(os.path.join(tmp, os.path.dirname(rel)), exist_ok=True)
@@ -10890,6 +11330,26 @@ def pkg(name: str, targets: list[tuple[str, str, str]], deps: list[str]) -> dict
         # reached, and what only an unreached unit holds is not
         for f, reached in (("base/data/list.txt", True), ("tools/table.tsv", True), ("base/data/idle.txt", False)):
             expect(bool(dep_reach(st, sg, ["gate-app"], f)) == reached, f"reach: {f} {'not reached' if reached else 'reached'} through the unit that holds it")
+        ki = sg.key_inputs("gate-app")
+        for f in want:
+            expect(ki.match(f) is not None, f"key reach omits reached dependency code {f}")
+        for f in ("base/src/a.rs", "base/src/d.rs", "base/src/f.rs", "base/data/idle.txt"):
+            expect(ki.match(f) is None, f"key reach retains unused linked input {f}")
+        for f in ("app/tests/it.rs", "app/tests/common/mod.rs", "base/src/lib.rs",
+                  "tools/lib-held/b.ids", "tools/own-held/a.ids"):
+            expect(ki.match(f) is not None, f"key reach omits a root/module/directory input {f}")
+        expect(ki.match("mid/src/idle.rs") is None, "key reach retains an unused reverse-dependent module")
+        for command in ("cat tools/idle.tsv", "bash tools/shared-read.sh"):
+            scripted = Recipe("gate-read", [*rs["gate-app"].lines, command], [], "")
+            reads_graph = Graph(st, {"gate-read": scripted})
+            expect(_linked_origin(reads_graph.inputs("gate-read").files["tools/idle.tsv"]),
+                   "shared-read regression needs Cargo to have supplied the first explanation")
+            read_inputs = reads_graph.key_inputs("gate-read")
+            expect(read_inputs.match("tools/idle.tsv") is not None, f"{command}: a Cargo-first explicit/script read was erased")
+            if command.startswith("bash"):
+                expect(read_inputs.match("tools/idle-held/extra.txt") is not None,
+                       "a Cargo-first literal prefix erased a script's independent find read")
+                expect("base/Cargo.toml" in read_inputs.manifests.whole, "script directory read lost the manifest whole override")
         # the row check reads the same links: a row for a path-literal file naming the gates that reach it is clean, and
         # one that leaves gate-app out is refused with the chain
         for f in ("base/data/list.txt", "tools/table.tsv"):
@@ -10947,6 +11407,50 @@ def pkg(name: str, targets: list[tuple[str, str, str]], deps: list[str]) -> dict
                f"gate-paths: a row for a test's own directory literal not refused: {got}")
         got = gate_paths_problems(st, rs, [row("tools/lib-held/b.ids", ["gate-app", "gate-mid"])])
         expect(not got, f"gate-paths: a row for a lib's directory literal refused: {got[:2]}")
+        # Opaque expansions are bounded by their runtime-active package and forward dependencies, not
+        # reverse/sibling packages. A transitive optional edge can be linked by a different Cargo root.
+        extra = {
+            "leaf/Cargo.toml": "", "leaf/src/lib.rs": "mod idle;\npub fn leaf() {}\n",
+            "leaf/src/idle.rs": "pub fn idle() {}\n",
+            "opt/Cargo.toml": "", "opt/src/lib.rs": "mod idle;\npub fn optional() {}\n",
+            "opt/src/idle.rs": "pub fn idle() {}\n",
+            "side/Cargo.toml": "", "side/src/lib.rs": "#[derive(custom::Hidden)] pub struct Side;\n",
+        }
+        for rel, text in extra.items():
+            os.makedirs(os.path.join(tmp, os.path.dirname(rel)), exist_ok=True)
+            with open(os.path.join(tmp, rel), "w", encoding="utf-8") as fh:
+                fh.write(text)
+        leaf = pkg("leaf", [("lib", "leaf", "src/lib.rs")], ["opt"])
+        leaf["dependencies"][0]["optional"] = True
+        opaque_meta = {
+            "workspace_root": tmp, "workspace_members": ["base", "mid", "app", "leaf", "opt", "side"],
+            "packages": [
+                pkg("base", [("lib", "base", "src/lib.rs")], ["leaf"]),
+                pkg("mid", [("lib", "mid", "src/lib.rs")], ["base"]),
+                pkg("app", [("lib", "app", "src/lib.rs"), ("test", "it", "tests/it.rs")], ["mid", "base", "opt", "side"]),
+                leaf, pkg("opt", [("lib", "opt", "src/lib.rs")], []), pkg("side", [("lib", "side", "src/lib.rs")], []),
+            ],
+        }
+        for source, widened in (
+            ("#[derive(Debug)] pub struct Hidden;\n", False),
+            ("#[derive(Debug, thiserror::Error)] pub enum Hidden {}\n", True),
+            ("#[ctor::ctor] fn startup() { crate::d::run(); }\n", True),
+            ('#[cfg_attr(unix, cfg_attr(feature = "boot", ctor::ctor))] fn startup() { crate::d::run(); }\n', True),
+        ):
+            with open(os.path.join(tmp, "base/src/a.rs"), "w", encoding="utf-8") as fh:
+                fh.write(source)
+            opaque_graph = Graph(Tree(tmp, opaque_meta), rs)
+            inputs = opaque_graph.key_inputs("gate-app")
+            for rel in ("base/src/d.rs", "base/data/idle.txt", "leaf/src/idle.rs", "opt/src/idle.rs"):
+                expect((inputs.match(rel) is not None) == widened, f"opaque owner closure miskeys {rel}: {source!r}")
+            for rel in ("mid/src/idle.rs", "side/src/lib.rs"):
+                expect(inputs.match(rel) is None, f"opaque owner restored an unrelated reverse/sibling input {rel}: {source!r}")
+        with open(os.path.join(tmp, "base/src/a.rs"), "w", encoding="utf-8") as fh:
+            fh.write("extern crate side as _;\n")
+        # An extern-crate link must be declared by its importer, not only by the final application's sibling edge.
+        opaque_meta["packages"][0] = pkg("base", [("lib", "base", "src/lib.rs")], ["leaf", "side"])
+        inputs = Graph(Tree(tmp, opaque_meta), rs).key_inputs("gate-app")
+        expect(inputs.match("side/src/lib.rs") is not None, "explicit side-effect extern-crate link lost its opaque owner")
 
     # the table's check on the real tree: the held rows, and the derivation's own lines
     tree, recipes, graph = side.tree, side.recipes, side.graph