Phase 2: static pass (bin/plugin-inventory), signals catalog, fixture tests

This commit is contained in:
Oleks
2026-07-29 15:44:10 +03:00
parent 36b2d63af2
commit bf88742bf4
10 changed files with 655 additions and 3 deletions
+6 -3
View File
@@ -20,6 +20,9 @@ steps:
- name: shellcheck
image: koalaman/shellcheck-alpine:stable
commands:
# bin/ is empty until Phase 2 — degrade gracefully rather than failing
# on "no files specified".
- sh -c 'files=$(find bin tests -type f \( -name "*.sh" -o -perm -u+x \) 2>/dev/null); if [ -n "$files" ]; then shellcheck $files; else echo "no scripts yet"; fi'
# bin/ mixes python and bash (plugin-inventory is python) — filter by
# actual shebang, not just the executable bit or a *.sh glob, or
# shellcheck chokes on the python entries (SC1071).
- >-
sh -c 'files=$(grep -lE "^#!.*(bash|/bin/sh)" bin/* tests/*.sh 2>/dev/null);
if [ -n "$files" ]; then shellcheck $files; else echo "no shell scripts yet"; fi'
+428
View File
@@ -0,0 +1,428 @@
#!/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()
+107
View File
@@ -0,0 +1,107 @@
# signals-catalog.md — static-pass lexicons and thresholds
Loaded at runtime by `bin/plugin-inventory`. Kept out of the script body so a
lexicon or threshold change is a reviewable diff, not a code change — per
`design/plan.md` §3.
Format: one `key: value` per line, grouped under `[section]` headers. Lines
starting with `#` are comments. Whitespace-insensitive; commas separate list
values. `plugin-inventory` parses this file itself — a genuine `pure-script`
read (T1T5 all pass: closed format, one right parse, deterministic, no
missing context, bounded grammar), so it is not a candidate for anything more
than what it already is.
## [script-verbs]
Occurrences count toward the script-verb signal. Matched case-insensitively,
whole-word, against skill/agent body prose (frontmatter excluded).
count, sum, rank, sort, diff, parse, validate, enumerate, dedupe, dedup,
deduplicate, format, compare, aggregate, glob, grep, filter, normalize,
normalise, tabulate, checksum, hash, lint, tally, group, join, merge, split,
truncate, paginate, timestamp, serialize, serialise, deserialize, encode,
decode
## [judgment-verbs]
Occurrences count toward the judgment-verb signal. Same matching rule.
decide, judge, explain, prioritize, prioritise, name, write, weigh,
interpret, recommend, assess, evaluate, justify, persuade, negotiate,
empathize, empathise, phrase, rephrase, summarize, summarise, synthesize,
synthesise, disambiguate, reconcile, adjudicate, arbitrate, narrate
# Note: summarize/synthesize sit in judgment, not script — compressing text
# into a good summary is not a total function of its inputs (T4), even though
# "summarize" sounds mechanical. If a step turns out to be a fixed-format
# extraction (e.g. "summarize" really means "pull these five fields"), that is
# a digest problem — flag it as low/medium, not high, until a human confirms
# the extraction is exhaustive
## [mechanical-tools]
Tools whose presence in an agent's `tools:` list leans that agent toward
"built to fetch." Used for `mechanical_tool_share`.
Bash, Read, Grep, Glob, WebFetch, WebSearch, LS, NotebookRead,
ListMcpResourcesTool, ReadMcpResourceTool, ReadMcpResourceDirTool, TaskGet,
TaskList, TaskOutput, Monitor
# Deliberately excluded (judgment-implying, even when mechanical-sounding)
# Edit, Write, NotebookEdit (produce durable state — a mistake here is
# consequential, T5/P4), AskUserQuestion (is itself an escalation), Agent
# (spawns judgment, doesn't replace it)
## [thresholds]
verb_ratio_high: 0.6
# verb_ratio (script / (script+judgment)) above this with zero bin/ script
# coverage is the loudest static smell per spec.md §5.1
script_coverage_low: 0.15
# script LOC / (skill+agent body words) below this, combined with
# verb_ratio_high, is the target profile for a real candidate
procedural_density_high: 0.35
# fraction of a skill body that is numbered/imperative steps containing a
# literal command, above which the skill reads as a recipe being re-derived
# every invocation rather than prose reasoning
mechanical_tool_share_high: 0.75
# fraction of an agent's declared tools that are in [mechanical-tools]
# above which the agent's own tool allowlist argues it was built to fetch
duplicate_block_min_skills: 2
# a fenced command block (after normalization — see FR-3.5 masking rules
# applied identically here) appearing in at least this many distinct skills
# counts as "a shared script that was never written."
hook_prose_drift_min_shared_tokens: 2
# a hook script and a skill body sharing at least this many identifier-shaped
# tokens (snake_case, SCREAMING_CASE, or a literal filename — NOT plain
# English words, which overlap constantly between topically-related prose)
# counts as the hook's logic being restated in prose — two sources of truth
@@ -0,0 +1,5 @@
{
"name": "sample-plugin",
"version": "0.1.0",
"description": "Fixture plugin for tests/inventory.test.sh — not a real plugin."
}
+8
View File
@@ -0,0 +1,8 @@
---
name: fetcher
description: Fixture agent — mostly mechanical tools.
model: sonnet
tools: Bash, Read, Grep, Glob, Edit
---
Fetches and greps things.
+3
View File
@@ -0,0 +1,3 @@
#!/usr/bin/env bash
# Fixture script — three lines of LOC for tests/inventory.test.sh.
echo "noop"
+7
View File
@@ -0,0 +1,7 @@
#!/usr/bin/env bash
# Fixture hook for tests/inventory.test.sh — checks FIXTURE_ENV_VAR via fixture_check.
set -euo pipefail
fixture_check() {
[ -n "${FIXTURE_ENV_VAR:-}" ]
}
fixture_check
+21
View File
@@ -0,0 +1,21 @@
---
name: dupe-a
description: |
Fixture skill for tests/inventory.test.sh. Trigger on "fixture-a".
allowed-tools: Bash, Read
---
# dupe-a
1. Count and sort the input.
2. Then judge which result matters and explain why.
3. Set `FIXTURE_ENV_VAR` before running — `fixture_check` restates this hook's
guard, which is exactly the drift this fixture exists to catch.
```bash
git status --short
```
| Finding | Action |
|---|---|
| foo | bar |
+14
View File
@@ -0,0 +1,14 @@
---
name: dupe-b
description: |
Fixture skill for tests/inventory.test.sh. Trigger on "fixture-b".
allowed-tools: Bash
---
# dupe-b
Parse and validate the output, then decide what to recommend.
```bash
git status --short
```
+56
View File
@@ -0,0 +1,56 @@
#!/usr/bin/env bash
# Asserts bin/plugin-inventory against tests/fixtures/sample-plugin, whose
# every signal value is known by construction (see the fixture files
# themselves for what each one is designed to trigger).
set -euo pipefail
cd "$(dirname "${BASH_SOURCE[0]}")/.."
OUT=$(python3 bin/plugin-inventory tests/fixtures/sample-plugin --json)
fail=0
check() {
local desc="$1" expr="$2"
if ! python3 -c "
import json, sys
d = json.loads('''$OUT''')
assert ($expr), 'FAILED: $desc'
"; then
echo "FAIL: $desc" >&2
fail=1
else
echo "ok: $desc"
fi
}
check "plugin name resolved from manifest" "d['plugin']['name'] == 'sample-plugin'"
check "two skills found" "len(d['skills']) == 2"
check "dupe-a verb_ratio is 0.5 (2 script, 2 judgment)" "d['skills'][0]['verb_ratio'] == 0.5"
check "dupe-a rule_tables counted" "d['skills'][0]['rule_tables'] == 1"
check "dupe-b has no rule table" "d['skills'][1]['rule_tables'] == 0"
check "duplicated command block found across dupe-a and dupe-b" "
any(b['sig'] == 'git status --short' and sorted(b['skills']) == ['dupe-a', 'dupe-b']
for b in d['aggregate']['duplicated_command_blocks'])
"
check "one agent found" "len(d['agents']) == 1"
check "fetcher mechanical_tool_share is 0.8 (4 of 5 tools)" "d['agents'][0]['mechanical_tool_share'] == 0.8"
check "one hook found" "len(d['hooks']) == 1 and d['hooks'][0]['file'] == 'guard.sh'"
check "hook LOC matches file" "d['hooks'][0]['loc'] == 8"
check "one bin script found, bash detected" "d['scripts'][0]['lang'] == 'bash' and d['scripts'][0]['loc'] == 3"
check "hook_prose_drift catches FIXTURE_ENV_VAR/fixture_check restated in dupe-a" "
any(p['hook'] == 'guard.sh' and p['skill'] == 'dupe-a'
and {'fixture_env_var', 'fixture_check'} <= set(p['shared_tokens'])
for p in d['aggregate']['hook_prose_drift'])
"
check "hook_prose_drift does NOT fire for dupe-b (no shared identifiers)" "
not any(p['skill'] == 'dupe-b' for p in d['aggregate']['hook_prose_drift'])
"
# Target resolution: bare name against a nonexistent cache entry fails loudly.
if python3 bin/plugin-inventory this-plugin-does-not-exist >/dev/null 2>&1; then
echo "FAIL: unresolvable target should exit non-zero" >&2
fail=1
else
echo "ok: unresolvable target exits non-zero"
fi
exit "$fail"