g4borg
f1de89e869
Standalone plugin with one skill: when to use orbit and when grep, a mandatory preflight for stale indexes and blind regions (cfg(test), features, macros, dyn dispatch), a scripted review pass over isolated worktree DBs, and a JSONL feedback log for mining real usage later. Registered in the marketplace as a prototype.
150 lines
8.3 KiB
Python
Executable File
150 lines
8.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Recipe 4: module/crate boundary guard for a diff.
|
|
|
|
Usage: boundary.py <env-file> <defdiff.json> <rules-file> [--audit] [--measure]
|
|
- NEW crate->crate import edges (tip minus base, whole repo) with an example file
|
|
- NEW import rows in changed files, checked against allow/deny rules
|
|
- NEW intra-crate module edges (crate::a -> crate::b) introduced by changed files
|
|
--audit also check every tip import against the rules (validates the rules themselves)
|
|
--measure compare files with new imports (orbit) against files whose diff adds a `use` line (text)
|
|
Imports are statement-level and kept even in cfg-gated/test files (unlike definitions).
|
|
Raw paths: `super::`/relative paths are attributed to the file's own crate; re-exports are not followed.
|
|
"""
|
|
import collections, fnmatch, json, os, re, sys
|
|
sys.path.insert(0, __file__.rsplit('/', 1)[0])
|
|
from ob import load_env, q, git
|
|
|
|
env = load_env(sys.argv[1]); dd = json.load(open(sys.argv[2])); RULES = sys.argv[3]
|
|
DB, BW, TW, REPO, B, T = env['DB'], env['BASE_WT'], env['TIP_WT'], env['REPO'], env['BASE_SHA'], env['TIP_SHA']
|
|
|
|
def root_pkg():
|
|
m = re.search(r'\[package\][^\[]*?name\s*=\s*"([^"]+)"', open(f"{TW}/Cargo.toml").read(), re.S)
|
|
return m.group(1).replace('-', '_') if m else '_root'
|
|
ROOT = root_pkg()
|
|
CRATES = {d.replace('-', '_') for d in os.listdir(f"{TW}/crates") if os.path.isdir(f"{TW}/crates/{d}")} | {ROOT}
|
|
|
|
def crate_of(f):
|
|
p = f.split('/')
|
|
return p[1].replace('-', '_') if p[0] == 'crates' and len(p) > 2 else ROOT
|
|
|
|
def dst_of(f, path):
|
|
head = path.split('::')[0]
|
|
if head in ('crate', 'self', 'super', ''): return crate_of(f)
|
|
return head.replace('-', '_')
|
|
|
|
allow, deny = {}, []
|
|
for line in open(RULES):
|
|
line = line.split('#', 1)[0].strip() if not line.lstrip().startswith('deny') else line.strip()
|
|
if line.startswith('allow'):
|
|
a, b = line[5:].split(':', 1); allow[a.strip()] = set(b.split())
|
|
elif line.startswith('deny'):
|
|
body, _, note = line[4:].partition('#')
|
|
src, _, rx = body.partition('->'); deny.append((src.strip(), re.compile(rx.strip()), note.strip()))
|
|
|
|
def rows(wt):
|
|
return q("SELECT file_path f, import_path p, identifier_name n, import_type it, start_line s FROM gl_imported_symbol "
|
|
"WHERE file_path LIKE '%.rs' AND import_type <> 'ModDeclaration'", repo=wt, db=DB)
|
|
R = {'base': rows(BW), 'tip': rows(TW)}
|
|
renames = dd['renames']
|
|
def key(r, side):
|
|
f = renames.get(r['f'], r['f']) if side == 'base' else r['f']
|
|
return (f, r['p'], r['n'])
|
|
|
|
def violations(r):
|
|
src, dst = crate_of(r['f']), dst_of(r['f'], r['p'])
|
|
out = []
|
|
if src in allow and dst in CRATES and dst != src and dst not in allow[src]:
|
|
out.append(f"allow-list: {src} may not import {dst}")
|
|
for g, rx, note in deny:
|
|
if fnmatch.fnmatch(r['f'], g) and rx.search(r['p']) and dst != src: out.append(f"deny: {note or rx.pattern}")
|
|
return out
|
|
|
|
# crate edges
|
|
def crate_edges(side):
|
|
e = collections.defaultdict(list)
|
|
for r in R[side]:
|
|
s, d = crate_of(r['f']), dst_of(r['f'], r['p'])
|
|
if d in CRATES and d != s: e[(s, d)].append(r['f'])
|
|
return e
|
|
be, te = crate_edges('base'), crate_edges('tip')
|
|
print(f"# boundary {B[:8]}..{T[:8]} root_pkg={ROOT} workspace_crates={len(CRATES)} rules: allow={len(allow)} deny={len(deny)}")
|
|
new_edges = sorted(set(te) - set(be)); gone = sorted(set(be) - set(te))
|
|
print(f"new crate edges: {len(new_edges)}" + ''.join(f"\n + {s} -> {d} ({len(te[(s,d)])} imports, e.g. {te[(s,d)][0]})" for s, d in new_edges))
|
|
if gone: print(f"removed crate edges: " + ', '.join(f"{s}->{d}" for s, d in gone))
|
|
|
|
changed = set(dd['changed_files'])
|
|
# grep fallback 1: inline qualified paths (`other_crate::x::Y` without a `use`) are NOT import rows
|
|
cre = re.compile(r'\b(' + '|'.join(sorted(CRATES, key=len, reverse=True)) + r')::')
|
|
def inline_edges(wt, files):
|
|
te_ = collections.defaultdict(set)
|
|
for f in files:
|
|
try: src = open(f"{wt}/{f}", errors='replace').read().split('\n')
|
|
except FileNotFoundError: continue
|
|
for i, l in enumerate(src):
|
|
t = l.strip()
|
|
if t.startswith(('//', 'use ', 'pub use ')): continue
|
|
for m in cre.finditer(l):
|
|
if m.group(1) != crate_of(f): te_[(crate_of(f), m.group(1))].add(f"{f}:{i+1}")
|
|
return te_
|
|
inv_ren = {v: k for k, v in renames.items()}
|
|
text_edges = inline_edges(TW, changed)
|
|
base_text = inline_edges(BW, {inv_ren.get(f, f) for f in changed})
|
|
imp_pairs = set(te)
|
|
extra = {k: v for k, v in text_edges.items() if k not in imp_pairs}
|
|
if extra: print("crate edges seen only as inline paths in changed files (no import row):" + ''.join(
|
|
f"\n ~ {s} -> {d} e.g. {sorted(v)[0]}{' [NEW: not in base imports or base inline paths]' if (s, d) not in be and (s, d) not in base_text else ''}" for (s, d), v in sorted(extra.items())))
|
|
# grep fallback 2: Cargo.toml dependency changes (the compiler-enforced graph)
|
|
cd = git(REPO, 'diff', '-U0', B, T, '--', '*Cargo.toml'); cf = None; deps = []
|
|
for l in cd.split('\n'):
|
|
if l.startswith('+++ '): cf = l[6:]
|
|
m = re.match(r'^([-+])\s*([\w-]+)\s*=', l)
|
|
if m and cf and not l.startswith(('+++', '---')) and m.group(2).replace('-', '_') in CRATES | {'dioxus', 'opencode_client'}:
|
|
deps.append(f"{m.group(1)}{m.group(2)} in {cf}")
|
|
print(f"Cargo.toml workspace-dep changes: {deps if deps else 'none'}")
|
|
for dline in deps:
|
|
sign, rest = dline[0], dline[1:]; dname, _, cf2 = rest.partition(' in ')
|
|
sc = cf2.split('/')[1].replace('-', '_') if cf2.startswith('crates/') else ROOT
|
|
if sign == '+' and sc in allow and dname.replace('-', '_') not in allow[sc]:
|
|
print(f" !! {sc} gains dependency {dname}, not in documented allow-list")
|
|
bkeys = {key(r, 'base') for r in R['base']}
|
|
new_rows = [r for r in R['tip'] if r['f'] in changed and key(r, 'tip') not in bkeys]
|
|
tkeys = {key(r, 'tip') for r in R['tip']}
|
|
gone_rows = [r for r in R['base'] if renames.get(r['f'], r['f']) in changed and key(r, 'base') not in tkeys]
|
|
print(f"new import rows in changed files: {len(new_rows)} (in {len({r['f'] for r in new_rows})} files); removed rows: {len(gone_rows)}")
|
|
viol = [(r, v) for r in new_rows for v in violations(r)]
|
|
print(f"rule violations among NEW imports: {len(viol)}" + ''.join(f"\n !! {r['f']}:{r['s']} use {r['p']} [{v}]" for r, v in viol[:20]))
|
|
|
|
# intra-crate module edges introduced by changed files
|
|
def mod_edges(rs):
|
|
e = set()
|
|
for r in rs:
|
|
p = r['p'].split('::')
|
|
if p[0] == 'crate' and len(p) > 2:
|
|
f = r['f']; parts = f.split('/')
|
|
src_mod = (parts[3] if parts[0] == 'crates' and len(parts) > 4 else parts[1] if parts[0] == 'src' and len(parts) > 2 else '(root)').replace('.rs', '')
|
|
if src_mod != p[1]: e.add((crate_of(f), src_mod, p[1]))
|
|
return e
|
|
allbase = mod_edges(R['base'])
|
|
nm = sorted(mod_edges(new_rows) - allbase)
|
|
print(f"new intra-crate module edges: {len(nm)}" + ''.join(f"\n + {c}: {a} -> {b}" for c, a, b in nm[:15]) + (f"\n … {len(nm)-15} more" if len(nm) > 15 else ''))
|
|
|
|
if '--audit' in sys.argv:
|
|
allv = collections.Counter(); ex = {}
|
|
for r in R['tip']:
|
|
for v in violations(r):
|
|
k = (crate_of(r['f']), dst_of(r['f'], r['p']), v); allv[k] += 1; ex.setdefault(k, f"{r['f']}:{r['s']} {r['p']}")
|
|
print(f"# audit: existing violations at tip: {sum(allv.values())}")
|
|
for k, n in allv.most_common(12): print(f" {n:4d} {k[0]} -> {k[1]} [{k[2]}] e.g. {ex[k]}")
|
|
|
|
if '--measure' in sys.argv:
|
|
diff = git(REPO, 'diff', '-U0', B, T, '--', '*.rs'); f = None; text_files = set(); text_del = set()
|
|
for l in diff.split('\n'):
|
|
if l.startswith('+++ '): f = l[6:]
|
|
elif l.startswith('--- '): of = l[6:]
|
|
elif re.match(r'^\+\s*(pub(\([^)]*\))?\s+)?use\s', l): text_files.add(f)
|
|
elif re.match(r'^-\s*(pub(\([^)]*\))?\s+)?use\s', l): text_del.add(renames.get(of, of))
|
|
ofiles = {r['f'] for r in new_rows}; ogone = {renames.get(r['f'], r['f']) for r in gone_rows}
|
|
print(f"# measure: files with added `use` line (text)={len(text_files)} orbit-new-import files={len(ofiles)} both={len(text_files & ofiles)} "
|
|
f"recall={len(text_files & ofiles)/max(1,len(text_files)):.0%}; text-missed={sorted(text_files - ofiles)[:5]}; orbit-only={len(ofiles - text_files)}")
|
|
print(f"# measure: files with removed `use` line (text)={len(text_del)} orbit-removed-import files={len(ogone)} both={len(text_del & ogone)}")
|