FabledScribe/scripts/check_plugin.py

#!/usr/bin/env python3
"""Guards for `plugin/` — the one part of this repo that ships straight to users.

WHY THIS EXISTS. `plugin/` is not built into the Docker image. Installs fetch it
from this git repo via `.claude-plugin/marketplace.json`, so a push IS the
release for plugin content: no build, no gate, immediately fetchable. Two
separate defects have reached a live install through that path:

  - #2198 — all four hook scripts were inert (wrong env-var case, line-oriented
    `jq -rR`, line-oriented `cut -c`). No CI ran, because `plugin/**` wasn't in
    the workflow's `paths:` filter at all.
  - #2209 — the fix for #2198 shipped to `main` and still couldn't reach an
    install, because `plugin.json`'s version wasn't bumped and the installer
    compares versions to decide whether to refresh its cache.

The rule for the second one was already written down and was still missed. A
written rule that depends on being remembered during a long session is not a
control; this is.

shellcheck and jq are NOT in `ci-python` (verified against CI-runner's Dockerfile
and scripts/install-common.sh, not from memory — rule #37). CI installs both
per-job, which is what CI-runner's own docs/process.md prescribes for a dep with
a single consumer: "If only one project needs the dep, prefer that project
installing it per-job in their workflow — at least until a second consumer
arrives." Promotion into the image is filed as an issue there rather than
assumed here.

Both are optional at runtime: without shellcheck the lint step is SKIPPED and
says so, and without jq the smoke test is skipped. A skipped check announces
itself loudly, because a check that quietly no-ops is the failure mode this
whole file exists to prevent.

Usage:
    python3 scripts/check_plugin.py            # all checks
    python3 scripts/check_plugin.py --no-version   # skip the bump check
"""
from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PLUGIN_DIR = ROOT / "plugin"
HOOKS_DIR = PLUGIN_DIR / "hooks"
MANIFEST = PLUGIN_DIR / ".claude-plugin" / "plugin.json"

# Paths whose contents reach an install. Keep in step with the workflow's
# `paths:` filter — a path that ships but isn't checked here is the gap again.
SHIPPED = ("plugin", ".claude-plugin")

failures: list[str] = []


def fail(msg: str) -> None:
    failures.append(msg)
    print(f"FAIL  {msg}")


def ok(msg: str) -> None:
    print(f"ok    {msg}")


def skip(msg: str) -> None:
    # Loud on purpose. A check that quietly does nothing is indistinguishable
    # from a check that passed — the exact confusion that let #2198 survive.
    print(f"SKIP  {msg}")


def hook_scripts() -> list[Path]:
    return sorted(HOOKS_DIR.glob("*.sh"))


# --- syntax ----------------------------------------------------------------

def check_syntax() -> None:
    """`bash -n` every hook. Catches nothing subtle, costs nothing, and a syntax
    error here means a hook that silently never runs."""
    for script in hook_scripts():
        proc = subprocess.run(
            ["bash", "-n", str(script)], capture_output=True, text=True
        )
        if proc.returncode != 0:
            fail(f"{script.relative_to(ROOT)}: bash -n — {proc.stderr.strip()}")
        else:
            ok(f"{script.relative_to(ROOT)}: syntax")


# --- known-bad patterns ----------------------------------------------------

# Each entry: (compiled pattern, short label, why it's wrong).
# These are the exact classes from #2198. They are deliberately specific — a
# broad shell linter belongs in the image, not hand-rolled here.
PATTERNS: list[tuple[re.Pattern, str, str]] = [
    (
        re.compile(r"CLAUDE_PLUGIN_OPTION_[a-z]"),
        "lowercase userConfig env var",
        "Claude Code exports userConfig to hooks as CLAUDE_PLUGIN_OPTION_<KEY> "
        "with the key UPPERCASED. The lowercase spelling reads as empty and the "
        "hook then does nothing, silently.",
    ),
    (
        # -R without -s: reads input line by line, so a multi-line payload is
        # encoded per line and joined with raw newlines. The class is a-r + t-z
        # (i.e. every letter EXCEPT `s`) so `-rR` is caught and `-sRr` is not —
        # an earlier a-q spelling silently excluded `r` and missed the real
        # defect, which is exactly the flag combination that shipped.
        re.compile(r"jq\s+-(?:[a-rt-zA-Z]*R[a-rt-zA-Z]*)\s"),
        "line-oriented jq -R",
        "jq -R reads input LINE BY LINE. Encoding a multi-line payload that way "
        "produces separate encoded lines joined by raw newlines — an invalid "
        "URL. Use -s (slurp) as well, e.g. `jq -sRr '@uri'`.",
    ),
    (
        re.compile(r"\|\s*cut\s+-c"),
        "line-oriented cut for a payload cap",
        "cut -c truncates EACH LINE, so it does not cap total size. Use "
        "`head -c N` to bound a payload.",
    ),
]


def check_patterns() -> None:
    for script in hook_scripts():
        text = script.read_text(encoding="utf-8", errors="replace")
        rel = script.relative_to(ROOT)
        hits = 0
        for line_no, line in enumerate(text.splitlines(), 1):
            # A line that only *documents* the trap is fine — several hooks now
            # carry a comment naming the wrong form so the next reader knows.
            if line.lstrip().startswith("#"):
                continue
            for pattern, label, why in PATTERNS:
                if pattern.search(line):
                    hits += 1
                    fail(f"{rel}:{line_no}: {label}\n        {line.strip()}\n        {why}")
        if not hits:
            ok(f"{rel}: no known-bad patterns")


# --- the version bump ------------------------------------------------------

# --- shellcheck ------------------------------------------------------------

def check_shellcheck() -> None:
    """Real shell linting, where available.

    The hand-rolled patterns above only know the bugs that already happened.
    This is what catches the next one.
    """
    exe = shutil.which("shellcheck")
    if not exe:
        skip("shellcheck not installed — install it to lint the hooks properly")
        return
    for script in hook_scripts():
        proc = subprocess.run(
            [exe, "--severity=warning", "--shell=bash", str(script)],
            capture_output=True, text=True,
        )
        rel = script.relative_to(ROOT)
        if proc.returncode != 0:
            fail(f"{rel}: shellcheck\n{proc.stdout.strip()}")
        else:
            ok(f"{rel}: shellcheck")


# --- the fail-open contract ------------------------------------------------

# Every hook promises never to break the operator's session: unconfigured or
# unreachable, it exits 0. Three of them additionally promise SILENCE, because
# they are pure enrichment. scribe_session_context.sh is the exception by
# design — it always emits a static behavioural floor that needs no credentials
# and no network, so "silent" would be the wrong assertion for it.
#
# This is the contract that made #2198 invisible for weeks, so it is worth
# pinning: the bug and the healthy no-results case look identical from outside.
# Pinning it does NOT make the failure visible; it makes sure the fail-open
# behaviour is deliberate rather than accidental.
SMOKE_EVENTS: dict[str, str] = {
    "scribe_autoinject.sh": json.dumps(
        {"session_id": "smoke", "cwd": ".", "prompt": "a multi-line\nprompt\nhere"}
    ),
    "scribe_prior_art.sh": json.dumps(
        {"session_id": "smoke", "cwd": ".", "tool_name": "Edit",
         "tool_input": {"file_path": "src/x.py", "new_string": "def f():\n    pass\n"}}
    ),
    "scribe_sync_processes.sh": json.dumps({"source": "startup"}),
    "scribe_session_context.sh": json.dumps({"source": "startup"}),
}

# The one hook that legitimately produces output with no credentials.
STATIC_FLOOR = "scribe_session_context.sh"


def _run_hook(script: Path, event: str, env_extra: dict[str, str]) -> subprocess.CompletedProcess:
    env = {k: v for k, v in os.environ.items()
           if not k.startswith(("SCRIBE_", "CLAUDE_PLUGIN_OPTION_"))}
    env.update(env_extra)
    return subprocess.run(
        ["bash", str(script)], input=event, capture_output=True, text=True,
        env=env, timeout=30,
    )


def check_fail_open() -> None:
    if not shutil.which("jq"):
        # Without jq every hook bails at its first line, so this would pass
        # while exercising nothing. Say so rather than bank a green tick.
        skip("jq not installed — the hooks would exit at line 1, so this "
             "check would pass without testing anything")
        return

    scenarios = [
        ("unconfigured", {}),
        # Connection refused immediately — exercises the unreachable-instance
        # path without waiting on a real network timeout.
        ("unreachable", {"SCRIBE_URL": "http://127.0.0.1:1", "SCRIBE_TOKEN": "x"}),
    ]
    for script in hook_scripts():
        rel = script.relative_to(ROOT)
        event = SMOKE_EVENTS.get(script.name)
        if event is None:
            skip(f"{rel}: no smoke event defined")
            continue
        for label, env_extra in scenarios:
            try:
                proc = _run_hook(script, event, env_extra)
            except subprocess.TimeoutExpired:
                fail(f"{rel} [{label}]: hung — a hook must never block a session")
                continue
            if proc.returncode != 0:
                fail(f"{rel} [{label}]: exited {proc.returncode}, must be 0 — "
                     f"a recall aid may never fail the operator's action")
                continue
            out = proc.stdout.strip()
            if script.name == STATIC_FLOOR:
                # Emits its bundled static tier regardless; that floor is the
                # whole point of the two-tier design.
                if not out:
                    fail(f"{rel} [{label}]: emitted nothing — the static "
                         f"behavioural floor must survive having no credentials")
                else:
                    ok(f"{rel} [{label}]: exit 0, static floor present")
            elif out:
                fail(f"{rel} [{label}]: emitted output with no working instance:\n"
                     f"        {out[:200]}")
            else:
                ok(f"{rel} [{label}]: exit 0, silent")


def _git(*args: str) -> tuple[int, str]:
    proc = subprocess.run(
        ["git", *args], capture_output=True, text=True, cwd=ROOT
    )
    return proc.returncode, (proc.stdout or proc.stderr).strip()


def manifest_version(ref: str | None = None) -> str | None:
    """The manifest version at `ref`, or in the working tree when ref is None."""
    if ref is None:
        try:
            return json.loads(MANIFEST.read_text()).get("version")
        except Exception:
            return None
    rel = MANIFEST.relative_to(ROOT).as_posix()
    code, out = _git("show", f"{ref}:{rel}")
    if code != 0:
        return None
    try:
        return json.loads(out).get("version")
    except Exception:
        return None


def check_version_bump(base: str = "origin/main") -> None:
    """If shipped plugin content differs from `base`, the version must too.

    Stated against the BASE BRANCH rather than the last commit on purpose. A
    per-commit rule would demand a bump from every commit in a batch; what
    actually matters is that whatever reaches an install carries a version the
    installer can tell apart from the one already cached. One bump per batch,
    which is also how a human would do it.
    """
    code, _ = _git("rev-parse", "--verify", base)
    if code != 0:
        # Do NOT pass silently — a check that quietly no-ops is how this class
        # of bug survives in the first place.
        fail(
            f"cannot resolve {base}, so the version-bump check could not run. "
            f"Fetch it first — `git fetch --depth=1 origin main:refs/remotes/"
            f"origin/main` is enough, since this diffs two trees and needs no "
            f"common ancestor — or pass --no-version deliberately."
        )
        return

    code, changed = _git("diff", "--name-only", base, "--", *SHIPPED)
    if code != 0:
        fail(f"git diff against {base} failed: {changed}")
        return
    if not changed.strip():
        ok(f"no shipped plugin changes against {base} — version bump not required")
        return

    here, there = manifest_version(), manifest_version(base)
    if here is None:
        fail(f"could not read a version from {MANIFEST.relative_to(ROOT)}")
        return
    if there is None:
        ok(f"no manifest on {base} — treating as a new plugin (version {here})")
        return
    if here == there:
        files = "\n          ".join(changed.splitlines())
        fail(
            f"plugin content changed but the manifest version is still {here}.\n"
            f"        The installer compares versions to decide whether to refresh "
            f"its cache, so an unchanged version means these edits reach the repo "
            f"and stop there — the marketplace clone updates, the cache that "
            f"actually executes does not (issue #2209).\n"
            f"        Bump `version` in {MANIFEST.relative_to(ROOT)}.\n"
            f"        Changed:\n          {files}"
        )
    else:
        ok(f"plugin content changed and version moved {there} -> {here}")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--no-version", action="store_true",
                        help="skip the manifest version-bump check")
    parser.add_argument("--base", default="origin/main",
                        help="branch the version bump is measured against")
    args = parser.parse_args()

    if not HOOKS_DIR.is_dir():
        print(f"FAIL  no hooks directory at {HOOKS_DIR}")
        return 1

    check_syntax()
    check_patterns()
    check_shellcheck()
    check_fail_open()
    if not args.no_version:
        check_version_bump(args.base)

    print()
    if failures:
        print(f"{len(failures)} problem(s) found.")
        return 1
    print("All plugin checks passed.")
    return 0


if __name__ == "__main__":
    sys.exit(main())