Oleks
429 lines
16 KiB
Python
Executable File
429 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Static pass over a Claude Code plugin's definitions -> one JSON document.
|
|
|
|
Usage: plugin-inventory <path|name> [--json] [--catalog PATH]
|
|
|
|
Reads a plugin's skills/, agents/, hooks/, and bin/, and reports the static
|
|
signals defined in design/spec.md §5.1 (verb ratios, procedural density,
|
|
duplicated command blocks, rule tables, script coverage, tool-allowlist
|
|
shape). Stdlib only, offline, no PyYAML — this file itself is the plugin's
|
|
own thesis (FR-8): reading a plugin's structure is a pure-script step by
|
|
every one of the five determinism tests in references/boundary-rubric.md.
|
|
|
|
No --json: pretty-printed (indent=2), for humans.
|
|
--json: compact single line, for piping into bin/offload-scan or tests.
|
|
"""
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
PLUGIN_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
DEFAULT_CATALOG = os.path.join(PLUGIN_ROOT, "references", "signals-catalog.md")
|
|
CACHE_GLOB = os.path.expanduser("~/.claude/plugins/cache/*/{name}/*/")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# signals-catalog.md loader
|
|
# --------------------------------------------------------------------------
|
|
|
|
def load_catalog(path):
|
|
sections = {}
|
|
current = None
|
|
with open(path, encoding="utf-8") as f:
|
|
for raw in f:
|
|
stripped = raw.strip()
|
|
if not stripped:
|
|
continue
|
|
# Section headers look like comments (start with #) but must be
|
|
# checked before generic comment-stripping discards them.
|
|
m = re.match(r"^#{0,2}\s*\[([a-z-]+)\]$", stripped)
|
|
if m:
|
|
current = m.group(1)
|
|
sections[current] = {} if current == "thresholds" else set()
|
|
continue
|
|
if stripped.startswith("#"):
|
|
continue
|
|
line = raw.split("#", 1)[0].strip()
|
|
if not line or current is None:
|
|
continue
|
|
if current == "thresholds":
|
|
if ":" in line:
|
|
k, v = line.split(":", 1)
|
|
sections[current][k.strip()] = float(v.strip())
|
|
else:
|
|
for item in line.split(","):
|
|
item = item.strip()
|
|
if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", item):
|
|
sections[current].add(item)
|
|
return sections
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# target resolution
|
|
# --------------------------------------------------------------------------
|
|
|
|
def semver_key(dirname):
|
|
m = re.match(r"^(\d+)\.(\d+)\.(\d+)", dirname)
|
|
return tuple(int(g) for g in m.groups()) if m else (0, 0, 0)
|
|
|
|
|
|
def resolve_target(target):
|
|
if os.sep in target or os.path.isdir(target):
|
|
path = os.path.abspath(target)
|
|
if not os.path.isdir(path):
|
|
sys.exit(f"error: no such directory: {path}")
|
|
return path, "path"
|
|
|
|
candidates = []
|
|
for entry in glob.glob(CACHE_GLOB.format(name=re.escape(target)).replace(r"\*", "*")):
|
|
if os.path.exists(os.path.join(entry, ".orphaned_at")):
|
|
continue
|
|
candidates.append(entry.rstrip("/"))
|
|
if not candidates:
|
|
sys.exit(f"error: no cached plugin found for name {target!r} "
|
|
f"(and not a directory)")
|
|
candidates.sort(key=lambda p: semver_key(os.path.basename(p)))
|
|
return candidates[-1], "oleks-local"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# frontmatter (no PyYAML — this environment's SKILL.md/agent.md frontmatter
|
|
# is a narrow subset: scalar `key: value` lines and `key: |` block scalars)
|
|
# --------------------------------------------------------------------------
|
|
|
|
def parse_frontmatter(text):
|
|
if not text.startswith("---"):
|
|
return {}, text
|
|
lines = text.split("\n")
|
|
if lines[0].strip() != "---":
|
|
return {}, text
|
|
end = None
|
|
for i in range(1, len(lines)):
|
|
if lines[i].strip() == "---":
|
|
end = i
|
|
break
|
|
if end is None:
|
|
return {}, text
|
|
meta = {}
|
|
i = 1
|
|
while i < end:
|
|
line = lines[i]
|
|
if not line.strip() or line.lstrip().startswith("#"):
|
|
i += 1
|
|
continue
|
|
m = re.match(r"^([A-Za-z0-9_-]+):\s*(.*)$", line)
|
|
if not m:
|
|
i += 1
|
|
continue
|
|
key, val = m.group(1), m.group(2).strip()
|
|
if val in ("|", "|-", ">", ">-"):
|
|
block = []
|
|
i += 1
|
|
while i < end and (lines[i].startswith(" ") or lines[i].startswith("\t") or not lines[i].strip()):
|
|
block.append(lines[i].strip())
|
|
i += 1
|
|
meta[key] = " ".join(b for b in block if b)
|
|
continue
|
|
meta[key] = val.strip('"\'')
|
|
i += 1
|
|
body = "\n".join(lines[end + 1:])
|
|
return meta, body
|
|
|
|
|
|
def split_list(value):
|
|
if not value:
|
|
return []
|
|
return [v.strip() for v in value.split(",") if v.strip()]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# signal computation
|
|
# --------------------------------------------------------------------------
|
|
|
|
def verb_counts(body, verbs):
|
|
total = 0
|
|
for verb in verbs:
|
|
total += len(re.findall(r"\b" + re.escape(verb) + r"\b", body, re.IGNORECASE))
|
|
return total
|
|
|
|
|
|
def procedural_density(body):
|
|
list_item_re = re.compile(r"^\s*(?:\d+[.)]|[-*])\s+.+$", re.MULTILINE)
|
|
items = list_item_re.findall(body)
|
|
if not items:
|
|
return 0.0
|
|
with_command = sum(1 for item in items if "`" in item)
|
|
return round(with_command / len(items), 3)
|
|
|
|
|
|
def extract_command_blocks(body):
|
|
blocks = re.findall(r"```(?:bash|sh|shell)?\n(.*?)```", body, re.DOTALL)
|
|
sigs = {}
|
|
for block in blocks:
|
|
first_line = next((ln.strip() for ln in block.splitlines() if ln.strip()), None)
|
|
if not first_line:
|
|
continue
|
|
sig = re.sub(r"\s+", " ", first_line)
|
|
sigs[sig] = sigs.get(sig, 0) + 1
|
|
return [{"sig": sig, "count": count} for sig, count in sorted(sigs.items())]
|
|
|
|
|
|
def count_rule_tables(body):
|
|
return len(re.findall(r"^\|.+\|\s*\n\|[\s:|-]+\|\s*$", body, re.MULTILINE))
|
|
|
|
|
|
def find_references(body, references_dir):
|
|
if not os.path.isdir(references_dir):
|
|
return []
|
|
known = set(os.listdir(references_dir))
|
|
found = set()
|
|
for m in re.finditer(r"([A-Za-z0-9_-]+\.md)", body):
|
|
if m.group(1) in known:
|
|
found.add(m.group(1))
|
|
return sorted(found)
|
|
|
|
|
|
def wc(text):
|
|
return len(text.split())
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# per-plugin walk
|
|
# --------------------------------------------------------------------------
|
|
|
|
def walk_skills(plugin_path, catalog):
|
|
"""Returns (skills, bodies) — bodies maps skill name -> body text, kept
|
|
out of the JSON output but needed by build_aggregate for drift detection."""
|
|
skills = []
|
|
bodies = {}
|
|
skills_dir = os.path.join(plugin_path, "skills")
|
|
if not os.path.isdir(skills_dir):
|
|
return skills, bodies
|
|
for name in sorted(os.listdir(skills_dir)):
|
|
skill_md = os.path.join(skills_dir, name, "SKILL.md")
|
|
if not os.path.isfile(skill_md):
|
|
continue
|
|
with open(skill_md, encoding="utf-8") as f:
|
|
text = f.read()
|
|
meta, body = parse_frontmatter(text)
|
|
script_v = verb_counts(body, catalog["script-verbs"])
|
|
judgment_v = verb_counts(body, catalog["judgment-verbs"])
|
|
ratio = round(script_v / (script_v + judgment_v), 3) if (script_v + judgment_v) else 0.0
|
|
skill_name = meta.get("name", name)
|
|
bodies[skill_name] = body
|
|
skills.append({
|
|
"name": skill_name,
|
|
"description_words": wc(meta.get("description", "")),
|
|
"body_words": wc(body),
|
|
"allowed_tools": split_list(meta.get("allowed-tools", meta.get("tools", ""))),
|
|
"procedural_density": procedural_density(body),
|
|
"script_verbs": script_v,
|
|
"judgment_verbs": judgment_v,
|
|
"verb_ratio": ratio,
|
|
"command_blocks": extract_command_blocks(body),
|
|
"rule_tables": count_rule_tables(body),
|
|
"references": find_references(body, os.path.join(skills_dir, name, "references")),
|
|
})
|
|
return skills, bodies
|
|
|
|
|
|
def walk_agents(plugin_path, catalog):
|
|
agents = []
|
|
agents_dir = os.path.join(plugin_path, "agents")
|
|
if not os.path.isdir(agents_dir):
|
|
return agents
|
|
mechanical = catalog["mechanical-tools"]
|
|
for fname in sorted(os.listdir(agents_dir)):
|
|
if not fname.endswith(".md"):
|
|
continue
|
|
with open(os.path.join(agents_dir, fname), encoding="utf-8") as f:
|
|
text = f.read()
|
|
meta, body = parse_frontmatter(text)
|
|
tools = split_list(meta.get("tools", ""))
|
|
mech = sum(1 for t in tools if t in mechanical)
|
|
agents.append({
|
|
"name": meta.get("name", fname[:-3]),
|
|
"model": meta.get("model", ""),
|
|
"tools": tools,
|
|
"body_words": wc(body),
|
|
"mechanical_tool_share": round(mech / len(tools), 3) if tools else 0.0,
|
|
})
|
|
return agents
|
|
|
|
|
|
def walk_hooks(plugin_path):
|
|
"""Returns (hooks, texts) — texts maps hook filename -> file content, kept
|
|
out of the JSON output but needed by build_aggregate for drift detection."""
|
|
hooks = []
|
|
texts = {}
|
|
hooks_dir = os.path.join(plugin_path, "hooks")
|
|
if not os.path.isdir(hooks_dir):
|
|
return hooks, texts
|
|
hooks_json = os.path.join(plugin_path, ".claude-plugin", "hooks.json")
|
|
events_by_file = {}
|
|
if os.path.isfile(hooks_json):
|
|
try:
|
|
with open(hooks_json, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
for event, entries in (data.get("hooks") or {}).items():
|
|
for entry in entries:
|
|
for hook in entry.get("hooks", []):
|
|
cmd = hook.get("command", "")
|
|
m = re.search(r"([\w.-]+\.sh)\b", cmd)
|
|
if m:
|
|
events_by_file[m.group(1)] = event
|
|
except (json.JSONDecodeError, OSError):
|
|
pass
|
|
for fname in sorted(os.listdir(hooks_dir)):
|
|
fpath = os.path.join(hooks_dir, fname)
|
|
if not os.path.isfile(fpath):
|
|
continue
|
|
with open(fpath, encoding="utf-8", errors="ignore") as f:
|
|
content = f.read()
|
|
texts[fname] = content
|
|
hooks.append({"file": fname, "event": events_by_file.get(fname, "unknown"),
|
|
"loc": content.count("\n") + 1})
|
|
return hooks, texts
|
|
|
|
|
|
def detect_lang(fpath):
|
|
try:
|
|
with open(fpath, encoding="utf-8", errors="ignore") as f:
|
|
first = f.readline()
|
|
except OSError:
|
|
return "unknown"
|
|
if first.startswith("#!"):
|
|
if "python" in first:
|
|
return "python"
|
|
if "bash" in first:
|
|
return "bash"
|
|
if "sh" in first:
|
|
return "sh"
|
|
if fpath.endswith(".py"):
|
|
return "python"
|
|
return "unknown"
|
|
|
|
|
|
def walk_scripts(plugin_path):
|
|
scripts = []
|
|
bin_dir = os.path.join(plugin_path, "bin")
|
|
if not os.path.isdir(bin_dir):
|
|
return scripts
|
|
for fname in sorted(os.listdir(bin_dir)):
|
|
fpath = os.path.join(bin_dir, fname)
|
|
if not os.path.isfile(fpath):
|
|
continue
|
|
with open(fpath, encoding="utf-8", errors="ignore") as f:
|
|
loc = sum(1 for _ in f)
|
|
scripts.append({"file": f"bin/{fname}", "loc": loc, "lang": detect_lang(fpath)})
|
|
return scripts
|
|
|
|
|
|
def distinctive_tokens(text):
|
|
"""Identifier-shaped tokens only (snake_case, SCREAMING_CASE, or
|
|
file/tool names with an underscore or a dot) — deliberately NOT plain
|
|
English words, which overlap constantly between topically-related prose
|
|
and produce false positives (e.g. every worktree-discipline skill shares
|
|
"already"/"session"/"branch" with every worktree-discipline hook,
|
|
regardless of whether any actual logic is restated). Real drift shows up
|
|
as the same *specific* identifier — an env var, a function name, a
|
|
literal filename — named in both places."""
|
|
snake = re.findall(r"\b[a-z][a-z0-9]*(?:_[a-z0-9]+){1,}\b", text)
|
|
screaming = re.findall(r"\b[A-Z][A-Z0-9]*(?:_[A-Z0-9]+){1,}\b", text)
|
|
dotted = re.findall(r"\b[a-zA-Z][a-zA-Z0-9_-]{3,}\.(?:sh|py|json|md)\b", text)
|
|
return {t.lower() for t in snake + screaming + dotted}
|
|
|
|
|
|
def build_aggregate(skills, agents, hooks, scripts, catalog, bodies=None, hook_texts=None):
|
|
skill_body_words = sum(s["body_words"] for s in skills)
|
|
agent_body_words = sum(a["body_words"] for a in agents)
|
|
script_loc = sum(s["loc"] for s in scripts) + sum(h["loc"] for h in hooks)
|
|
total_words = skill_body_words + agent_body_words
|
|
coverage = round(script_loc / total_words, 3) if total_words else 0.0
|
|
|
|
sig_skills = {}
|
|
for skill in skills:
|
|
for block in skill["command_blocks"]:
|
|
sig_skills.setdefault(block["sig"], set()).add(skill["name"])
|
|
min_skills = int(catalog["thresholds"].get("duplicate_block_min_skills", 2))
|
|
duplicated = [
|
|
{"sig": sig, "skills": sorted(names)}
|
|
for sig, names in sorted(sig_skills.items())
|
|
if len(names) >= min_skills
|
|
]
|
|
|
|
drift = []
|
|
bodies = bodies or {}
|
|
hook_texts = hook_texts or {}
|
|
min_shared = int(catalog["thresholds"].get("hook_prose_drift_min_shared_tokens", 3))
|
|
for hook in hooks:
|
|
hook_tokens = distinctive_tokens(hook_texts.get(hook["file"], ""))
|
|
if not hook_tokens:
|
|
continue
|
|
for skill_name, body in bodies.items():
|
|
shared = hook_tokens & distinctive_tokens(body)
|
|
if len(shared) >= min_shared:
|
|
drift.append({
|
|
"hook": hook["file"],
|
|
"skill": skill_name,
|
|
"shared_tokens": sorted(shared)[:10],
|
|
})
|
|
|
|
return {
|
|
"skill_body_words": skill_body_words,
|
|
"agent_body_words": agent_body_words,
|
|
"script_loc": script_loc,
|
|
"script_coverage": coverage,
|
|
"duplicated_command_blocks": duplicated,
|
|
"hook_prose_drift": drift,
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(prog="plugin-inventory")
|
|
parser.add_argument("target", help="plugin directory path, or a bare name to resolve from the plugin cache")
|
|
parser.add_argument("--json", action="store_true", help="compact single-line JSON instead of pretty-printed")
|
|
parser.add_argument("--catalog", default=DEFAULT_CATALOG, help="path to signals-catalog.md")
|
|
args = parser.parse_args()
|
|
|
|
catalog = load_catalog(args.catalog)
|
|
plugin_path, source = resolve_target(args.target)
|
|
|
|
manifest_path = os.path.join(plugin_path, ".claude-plugin", "plugin.json")
|
|
manifest = {}
|
|
if os.path.isfile(manifest_path):
|
|
with open(manifest_path, encoding="utf-8") as f:
|
|
manifest = json.load(f)
|
|
|
|
skills, bodies = walk_skills(plugin_path, catalog)
|
|
agents = walk_agents(plugin_path, catalog)
|
|
hooks, hook_texts = walk_hooks(plugin_path)
|
|
scripts = walk_scripts(plugin_path)
|
|
aggregate = build_aggregate(skills, agents, hooks, scripts, catalog, bodies, hook_texts)
|
|
|
|
doc = {
|
|
"plugin": {
|
|
"name": manifest.get("name", os.path.basename(plugin_path)),
|
|
"version": manifest.get("version", "unknown"),
|
|
"source": source,
|
|
"path": plugin_path,
|
|
},
|
|
"skills": skills,
|
|
"agents": agents,
|
|
"hooks": hooks,
|
|
"scripts": scripts,
|
|
"aggregate": aggregate,
|
|
}
|
|
|
|
if args.json:
|
|
print(json.dumps(doc))
|
|
else:
|
|
print(json.dumps(doc, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|