#!/usr/bin/env bash # Pre-publish acceptance gate. # # Statically intersects what each shipped skill *prescribes* (every # backtick-quoted `mcp____` token in SKILL.md and references/*.md) # against the dispatched specialist's frontmatter `tools:` list. Catches the # class of bug where a skill prescribes a tool the specialist does not have, # and where a skill prescribes a forbidden direct-execution path # (`cypher-shell`, `neo4j-admin` invocations, raw-Cypher DML in prose). # # Wired into the root `packages/create-maxy-code/package.json` `prepublishOnly` # script so a regression cannot reach npm publish without firing. # # One stdout line per (skill, specialist) pair: # [verify] skill=/ specialist= resolved=/ forbidden= # # Exit 0: every prescribed token resolves AND no forbidden tokens. # Exit 1: any unresolved or any forbidden — stderr names the offending token. # # Skill→specialist mapping comes from PLUGIN.md frontmatter `specialist:` field. # Plugins without that field are admin-owned (loaded by the admin agent # directly via plugin-read); for those the gate only enforces the forbidden- # token rule, since admin's tool surface is the union of all enabled plugins. set -euo pipefail REPO_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd) cd "$REPO_ROOT" # Script uses PEP 604 union syntax (`X | None`); requires Python ≥3.10. # macOS system /usr/bin/python3 is 3.9, so pin to the highest 3.10+ available. PY="" for cand in python3.13 python3.12 python3.11 python3.10; do if command -v "$cand" >/dev/null 2>&1; then PY="$cand"; break; fi done if [ -z "$PY" ]; then v=$(python3 -c 'import sys; print("%d.%d" % sys.version_info[:2])') case "$v" in 3.1[0-9]|3.[2-9][0-9]|[4-9].*) PY=python3 ;; esac fi if [ -z "$PY" ]; then echo "FATAL: verify-skill-tool-surface needs Python >= 3.10 (PEP 604 syntax); none found" >&2 exit 1 fi "$PY" - <<'PYEOF' import os, re, sys REPO_ROOT = os.getcwd() PLUGINS_DIR = os.path.join(REPO_ROOT, "platform", "plugins") SPECIALISTS_DIR = os.path.join(REPO_ROOT, "platform", "templates", "specialists", "agents") # Per-skill specialist ownership for plugins where the plugin itself is # multi-purpose. The PLUGIN.md `specialist:` field handles single-owner # plugins (linkedin-import → librarian). Mixed-use plugins like `memory` # declare per-skill ownership here. EXPLICIT_OWNERSHIP = { # plugin/skill -> specialist "memory/document-ingest": "librarian", "memory/conversation-archive": "librarian", "memory/conversation-archive-enrich": "librarian", } # Skills that are explicitly admin-owned (loaded via plugin-read by the admin # agent itself, not delegated to a specialist). These get only the forbidden- # token check since admin's effective tool set is the union of all enabled # plugins. ADMIN_OWNED_SKILLS = { "memory/conversational-memory", } # Regex for skill-name tokens in the prompt body that require `Skill` in tools: # 1. `skill-load skillName=` pattern (how templates reference Skill calls) # 2. `Skill ` pattern (explicit Skill tool invocation in enforcement directives) # Only hyphenated lowercase names are skill names. SKILL_LOAD_RE = re.compile(r"skillName=([a-z][a-z0-9-]+)") SKILL_INVOKE_RE = re.compile(r"`Skill\s+([a-z][a-z0-9-]+)`") TOKEN_RE = re.compile(r"`(mcp__[a-z][a-z0-9_-]*__[a-z][a-z0-9_-]*)`") FENCED_BLOCK_RE = re.compile(r"```(?P[a-zA-Z]*)\n(?P.*?)\n```", re.S) PROSE_CYPHER_RE = re.compile( r"`(?:MERGE|CREATE|DETACH\s+DELETE)\s+\(", re.IGNORECASE, ) def parse_frontmatter(path: str) -> dict | None: """Parse YAML-ish frontmatter without PyYAML — handles `key: value` and `key:\n - item\n - item` shapes used in PLUGIN.md and specialist files.""" try: text = open(path, encoding="utf-8").read() except FileNotFoundError: return None m = re.match(r"^---\n(.*?)\n---", text, re.S) if not m: return None block = m.group(1) out: dict = {} cur_key: str | None = None cur_list: list[str] | None = None for line in block.split("\n"): if not line.strip(): continue # List item under cur_key if line.startswith(" - ") or line.startswith("- "): if cur_list is None: continue cur_list.append(line.split("- ", 1)[1].strip()) continue # Top-level key m2 = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\s*:\s*(.*)$", line) if not m2: continue cur_key = m2.group(1) rhs = m2.group(2).strip() if rhs: # inline value — strip surrounding quotes if (rhs.startswith('"') and rhs.endswith('"')) or ( rhs.startswith("'") and rhs.endswith("'") ): rhs = rhs[1:-1] out[cur_key] = rhs cur_list = None else: cur_list = [] out[cur_key] = cur_list return out def specialist_tools(specialist: str) -> set[str] | None: fm = parse_frontmatter(os.path.join(SPECIALISTS_DIR, f"{specialist}.md")) if fm is None: return None raw = fm.get("tools", "") if isinstance(raw, list): items = raw else: items = [t.strip() for t in str(raw).split(",")] return {t for t in items if t} def extract_prescribed_tokens(text: str) -> set[str]: return set(TOKEN_RE.findall(text)) def extract_forbidden(text: str) -> list[tuple[str, str]]: forbidden: list[tuple[str, str]] = [] # Forbidden invocations inside fenced shell blocks for m in FENCED_BLOCK_RE.finditer(text): lang = m.group("lang").lower() body = m.group("body") if lang in {"bash", "sh", "shell", "zsh"}: if re.search(r"\bcypher-shell\b", body): forbidden.append(("cypher-shell", "in fenced shell block")) if re.search(r"\bneo4j-admin\s+dbms\b", body): forbidden.append(("neo4j-admin", "in fenced shell block")) # Strip fenced blocks for the prose-Cypher heuristic prose = FENCED_BLOCK_RE.sub("", text) if PROSE_CYPHER_RE.search(prose): forbidden.append(("raw-cypher-dml", "in backtick-quoted prose")) return forbidden def aggregate_skill_text(skill_dir: str) -> str: out: list[str] = [] skill_md = os.path.join(skill_dir, "SKILL.md") if not os.path.exists(skill_md): return "" out.append(open(skill_md, encoding="utf-8").read()) refs = os.path.join(skill_dir, "references") if os.path.isdir(refs): for name in sorted(os.listdir(refs)): if name.endswith(".md"): out.append(open(os.path.join(refs, name), encoding="utf-8").read()) return "\n".join(out) def collect_skill_names() -> set[str]: """Enumerate all installed skill directory names from platform/plugins/*/skills/.""" names: set[str] = set() if not os.path.isdir(PLUGINS_DIR): return names for plugin in os.listdir(PLUGINS_DIR): skills_dir = os.path.join(PLUGINS_DIR, plugin, "skills") if not os.path.isdir(skills_dir): continue for entry in os.listdir(skills_dir): if os.path.isdir(os.path.join(skills_dir, entry)): names.add(entry) return names def specialist_body(specialist_path: str) -> str: """Return the prompt body of a specialist template (content below frontmatter).""" try: text = open(specialist_path, encoding="utf-8").read() except FileNotFoundError: return "" m = re.match(r"^---\n.*?\n---\n(.*)", text, re.S) return m.group(1) if m else text def check_builtin_tool_rule( specialist: str, body: str, tools: set[str], known_skills: set[str], ) -> list[str]: """Built-in-tool rule: if the specialist prompt references a skill load that matches an installed skill directory, `Skill` must appear in tools:.""" referenced: set[str] = set() # Pattern 1: skill-load skillName= (template invocation form) referenced |= set(SKILL_LOAD_RE.findall(body)) & known_skills # Pattern 2: `Skill ` (enforcement directive form) referenced |= set(SKILL_INVOKE_RE.findall(body)) & known_skills if not referenced or "Skill" in tools: return [] return [ f"[verify] specialist={specialist} claims-loads={name} " f"but Skill missing from tools" for name in sorted(referenced) ] def check_webfetch_rule( specialist: str, body: str, tools: set[str], ) -> list[str]: """WebFetch rule: if the specialist prompt names WebFetch explicitly or directs dereferencing an operator-provided URL, WebFetch must be in tools:.""" if "WebFetch" in tools: return [] has_webfetch = bool(re.search(r"\bWebFetch\b", body)) has_fetch_url = bool(re.search(r"\bfetch\b.*\bURL\b", body, re.IGNORECASE)) if not (has_webfetch or has_fetch_url): return [] reason = "WebFetch" if has_webfetch else r"\bfetch\b.*\bURL\b" return [ f"[verify] specialist={specialist} directs WebFetch but tool missing " f"(matched: {reason})" ] def main() -> int: if not os.path.isdir(PLUGINS_DIR): print(f"[verify] PLUGINS_DIR not found: {PLUGINS_DIR}", file=sys.stderr) return 1 if not os.path.isdir(SPECIALISTS_DIR): print(f"[verify] SPECIALISTS_DIR not found: {SPECIALISTS_DIR}", file=sys.stderr) return 1 summary: list[str] = [] errors: list[str] = [] pairs_checked = 0 for plugin in sorted(os.listdir(PLUGINS_DIR)): pdir = os.path.join(PLUGINS_DIR, plugin) if not os.path.isdir(pdir): continue plugin_fm = parse_frontmatter(os.path.join(pdir, "PLUGIN.md")) or {} plugin_specialist = plugin_fm.get("specialist") skills_dir = os.path.join(pdir, "skills") if not os.path.isdir(skills_dir): continue for skill_name in sorted(os.listdir(skills_dir)): sdir = os.path.join(skills_dir, skill_name) if not os.path.isdir(sdir): continue text = aggregate_skill_text(sdir) if not text: continue prescribed = extract_prescribed_tokens(text) forbidden = extract_forbidden(text) ownership_key = f"{plugin}/{skill_name}" if ownership_key in ADMIN_OWNED_SKILLS: specialist = None else: specialist = ( EXPLICIT_OWNERSHIP.get(ownership_key) or plugin_specialist ) if specialist is None: # Admin-owned: only enforce forbidden-token rule. for tok, ctx in forbidden: errors.append( f"[verify] skill={ownership_key} specialist=admin " f"FORBIDDEN token={tok} context=\"{ctx}\"" ) summary.append( f"[verify] skill={ownership_key} specialist=admin (admin-owned) " f"tokens={len(prescribed)} forbidden={len(forbidden)}" ) continue tools = specialist_tools(specialist) if tools is None: errors.append( f"[verify] skill={ownership_key} specialist={specialist} " f"ERROR specialist frontmatter not parseable" ) continue unresolved = sorted(prescribed - tools) for tok in unresolved: errors.append( f"[verify] skill={ownership_key} specialist={specialist} " f"unresolved={tok}" ) for tok, ctx in forbidden: errors.append( f"[verify] skill={ownership_key} specialist={specialist} " f"FORBIDDEN token={tok} context=\"{ctx}\"" ) summary.append( f"[verify] skill={ownership_key} specialist={specialist} " f"resolved={len(prescribed) - len(unresolved)}/{len(prescribed)} " f"forbidden={len(forbidden)}" ) pairs_checked += 1 # Built-in-tool rule + WebFetch rule: sweep all agent template .md files. # Covers platform specialists (SPECIALISTS_DIR) and premium-plugin agent # templates (premium-plugins/*/agents/). These rules are orthogonal to the # skill-pairing loop above — they check each agent's own prompt body for # directives that require built-in CC tools. known_skills = collect_skill_names() builtin_checked = 0 agent_template_dirs: list[str] = [] if os.path.isdir(SPECIALISTS_DIR): agent_template_dirs.append(SPECIALISTS_DIR) pp_root = os.path.join(REPO_ROOT, "premium-plugins") if os.path.isdir(pp_root): for plugin in sorted(os.listdir(pp_root)): adir = os.path.join(pp_root, plugin, "agents") if os.path.isdir(adir): agent_template_dirs.append(adir) for adir in agent_template_dirs: for fname in sorted(os.listdir(adir)): if not fname.endswith(".md"): continue spath = os.path.join(adir, fname) fm = parse_frontmatter(spath) if fm is None: continue specialist = fm.get("name") or fname[:-3] raw = fm.get("tools", "") if isinstance(raw, list): tools = {t for t in raw if t} else: tools = {t.strip() for t in str(raw).split(",") if t.strip()} body = specialist_body(spath) for err in check_builtin_tool_rule(specialist, body, tools, known_skills): errors.append(err) for err in check_webfetch_rule(specialist, body, tools): errors.append(err) builtin_checked += 1 summary.append( f"[verify] builtin-tool-check specialists_checked={builtin_checked}" ) for line in summary: print(line) if errors: for line in errors: print(line, file=sys.stderr) print( f"[verify] FAIL pairs_checked={pairs_checked} errors={len(errors)}", file=sys.stderr, ) return 1 print(f"[verify] OK pairs_checked={pairs_checked}") return 0 sys.exit(main()) PYEOF