What vexjoy-agent runs automatically, and when. A hook is a command Claude Code fires at a fixed moment, without you asking for it.
Claude Code
$ npx -y skills add notque/vexjoy-agent --agent claude-code
Cursor
Cursor also runs Claude Code hooks from .claude/settings.json, so copy these hooks there to use them in Cursor.
Where it lives
# Claude Code hooks for event-driven automation # hook-version: 1.0.0
#!/usr/bin/env python3
# hook-version: 1.0.2
"""
PostToolUse Hook: ADR Compliance Enforcement
After every Write or Edit tool call on a pipeline component file, automatically
run adr-compliance.py and inject violations as feedback into the next context.
Design Principles:
- Non-blocking (always exits 0)
- Silent on non-pipeline files (no noise)
- Graceful degradation when adr-compliance.py not yet deployed
- Skips when no active ADR session (.adr-session.json absent)
"""
import sys # vexjoy-pyguard
if sys.version_info < (3, 10): # vexjoy-pyguard advisory
import os
if not os.environ.get("VEXJOY_PYGUARD_REEXEC"):
os.environ["VEXJOY_PYGUARD_REEXEC"] = "1"
for _d in [*os.environ.get("PATH", "").split(os.pathsep), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]:
for _v in ("14", "13", "12", "11", "10"):
_p = os.path.join(_d, "python3." + _v)
if _d and "/shims" not in _d and os.access(_p, os.X_OK):
os.execv(_p, [_p, *sys.argv])
sys.stderr.write(
"[vexjoy] hook %s needs Python 3.10+, got %s (%s); run ./install.sh in vexjoy-agent from a terminal"
" outside Claude Code (gated tool calls stay blocked until then)\n"
% (__file__.rsplit("/", 1)[-1], sys.version.split()[0], sys.executable)
)
sys.exit(0)
import json
import os
import re
import subprocess
import sys
from pathlib import Path
# Add lib directory to path for imports
sys.path.insert(0, str(Path(__file__).parent / "lib"))
from hook_utils import context_output, empty_output, hook_error, log_warning
from stdin_timeout import read_stdin
_EVENT_NAME = "PostToolUse"
_TRUSTED_ROOT = Path(__file__).resolve().parent.parent
# Pipeline component files that trigger enforcement (matched against repo-relative paths)
_PIPELINE_COMPONENT_PATTERNS = [
r"^skills/(?:[^/]+/)?[^/]+/SKILL\.md$",
r"^agents/[^/]+\.md$",
r"^scripts/[^/]+\.py$",
r"^hooks/[^/]+\.py$",
]
# Repo-relative paths to exclude even if they match a component pattern
_EXCLUDE_PATTERNS = [
r"^agents/INDEX\.json",
r"^hooks/lib/",
r"^hooks/adr-enforcement\.py$",
r"^scripts/tests/",
r"^scripts/__pycache__/",
]
# Reference files used by adr-compliance.py
_STEP_MENU = "skills/process/workflow/references/pipeline-scaffolder/references/step-menu.md"
_SPEC_FORMAT = "skills/process/workflow/references/pipeline-scaffolder/references/pipeline-spec-format.md"
def _project_root(event: dict) -> Path:
"""Resolve the repository the hook event is operating on."""
candidate = event.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or Path.cwd()
return Path(candidate).resolve()
def _project_relative_path(file_path: str, project_root: Path) -> Path | None:
"""Return ``file_path`` relative to the event project, or None if outside it."""
try:
path = Path(file_path)
absolute = path.resolve() if path.is_absolute() else (project_root / path).resolve()
return absolute.relative_to(project_root)
except (OSError, ValueError):
return None
def is_pipeline_component(file_path: str, project_root: Path | None = None) -> bool:
"""Check if the file is a pipeline component that should be compliance-checked.
Normalizes the path to be relative to the repo root before matching, so
patterns like agents/ only match the top-level agents/ directory and not
arbitrary path components.
"""
root = project_root or Path(os.environ.get("CLAUDE_PROJECT_DIR", Path.cwd())).resolve()
rel_path = _project_relative_path(file_path, root)
if rel_path is None:
return False
rel_str = rel_path.as_posix()
for exclude in _EXCLUDE_PATTERNS:
if re.search(exclude, rel_str):
return False
return any(re.match(pattern, rel_str) for pattern in _PIPELINE_COMPONENT_PATTERNS)
def load_session(cwd: str) -> dict | None:
"""Load .adr-session.json from cwd. Returns None if absent or invalid."""
session_path = Path(cwd) / ".adr-session.json"
if not session_path.exists():
return None
try:
with open(session_path) as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return None
def run_compliance_check(
compliance_script: Path,
file_path: str,
project_root: Path,
) -> dict | None:
"""
Run adr-compliance.py check on the file.
Returns parsed JSON output dict, or None on subprocess failure.
"""
cmd = [
sys.executable,
str(compliance_script),
"check",
"--file",
str(project_root / file_path),
"--step-menu",
str(project_root / _STEP_MENU),
"--spec-format",
str(project_root / _SPEC_FORMAT),
]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=10,
cwd=str(_TRUSTED_ROOT),
)
output = result.stdout.strip()
if not output:
return None
return json.loads(output)
except subprocess.TimeoutExpired:
log_warning("adr-compliance.py timed out after 10s")
return None
except (json.JSONDecodeError, OSError):
return None
def format_violations(file_path: str, check_result: dict) -> str:
"""Format violation output for context injection."""
violations = check_result.get("violations", [])
lines = []
# Use a relative display path when possible
display_path = file_path
_chk = "COMPLIANCE CHECK"
lines.append(f"[adr-enforcement] {_chk}: {display_path}")
count = len(violations)
_vf = "VIOLATIONS FOUND"
lines.append(f"[adr-enforcement] {_vf} ({count}):")
for v in violations:
line_num = v.get("line", "?")
v_type = v.get("type", "unknown")
value = v.get("value", "")
suggestion = v.get("suggestion", "")
entry = f' Line {line_num}: {v_type} "{value}"'
if suggestion#!/usr/bin/env python3
# hook-version: 1.1.0
"""PostToolUse Hook: ADR lifecycle checker on merge.
Fires after a Bash tool use. Detects merge commands (gh pr merge / git merge)
and scans the branch name and recent commit messages for ADR references.
For each matched ADR it:
1. Reads the ## Implementation section and extracts numbered steps.
2. Diffs HEAD~1 to get changed files.
3. Checks each step for keyword matches in the changed-file list.
4. Reports a checklist with PARTIAL / COMPLETE status.
5. If all steps matched, updates the ADR file status to "Completed" and moves
it to adr/completed/ (shutil, not git mv — adr/ is gitignored).
Always exits 0 (non-blocking, informational hook).
Target execution time: <200ms (exits early when not a merge command).
"""
import sys # vexjoy-pyguard
if sys.version_info < (3, 10): # vexjoy-pyguard advisory
import os
if not os.environ.get("VEXJOY_PYGUARD_REEXEC"):
os.environ["VEXJOY_PYGUARD_REEXEC"] = "1"
for _d in [*os.environ.get("PATH", "").split(os.pathsep), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]:
for _v in ("14", "13", "12", "11", "10"):
_p = os.path.join(_d, "python3." + _v)
if _d and "/shims" not in _d and os.access(_p, os.X_OK):
os.execv(_p, [_p, *sys.argv])
sys.stderr.write(
"[vexjoy] hook %s needs Python 3.10+, got %s (%s); run ./install.sh in vexjoy-agent from a terminal"
" outside Claude Code (gated tool calls stay blocked until then)\n"
% (__file__.rsplit("/", 1)[-1], sys.version.split()[0], sys.executable)
)
sys.exit(0)
import json
import os
import re
import shutil
import subprocess
import sys
from datetime import date
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent / "lib"))
from hook_utils import append_hook_debug_log, context_output, empty_output, get_tool_input
from stdin_timeout import read_stdin
DEBUG_LOG = Path("/tmp/claude_hook_debug.log")
EVENT_NAME = "PostToolUse"
# Patterns to identify ADR references in branch names / commit messages
ADR_PATTERNS = [
re.compile(r"ADR-(\d+)", re.IGNORECASE),
re.compile(r"adr/(\d+)-"),
re.compile(r"adr-(\d+)", re.IGNORECASE),
]
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
def debug_log(msg: str) -> None:
"""Append to the shared debug log (0600) only when CLAUDE_HOOKS_DEBUG is set."""
append_hook_debug_log(DEBUG_LOG, f"[adr-lifecycle] {msg}")
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def run_git(args: list[str], cwd: str | None = None, timeout: int = 5) -> str:
"""Run a git command and return stdout. Returns '' on failure."""
try:
result = subprocess.run(
["git"] + args,
capture_output=True,
text=True,
timeout=timeout,
cwd=cwd,
)
return result.stdout.strip() if result.returncode == 0 else ""
except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
return ""
def extract_adr_numbers(text: str) -> list[str]:
"""Return deduplicated ADR numbers found in *text*."""
seen: set[str] = set()
numbers: list[str] = []
for pattern in ADR_PATTERNS:
for m in pattern.finditer(text):
num = m.group(1).lstrip("0") or "0"
if num not in seen:
seen.add(num)
numbers.append(num)
return numbers
def find_adr_file(adr_number: str) -> Path | None:
"""Locate adr/{NNN}-*.md, zero-padded searches 1-4 digits."""
adr_dir = Path(os.environ.get("CLAUDE_PROJECT_DIR", ".")) / "adr"
if not adr_dir.is_dir():
return None
# Try zero-padded variants: 1 → 001, 179 → 179, etc.
num_int = int(adr_number)
padded_variants = [
str(num_int),
f"{num_int:03d}",
f"{num_int:04d}",
]
for variant in padded_variants:
matches = list(adr_dir.glob(f"{variant}-*.md"))
if matches:
return matches[0]
return None
def extract_implementation_steps(adr_content: str) -> list[str]:
"""Extract numbered steps from ## Implementation section."""
steps: list[str] = []
in_section = False
for line in adr_content.splitlines():
if re.match(r"^#{1,3}\s+Implementation", line, re.IGNORECASE):
in_section = True
continue
if in_section:
# Stop at next heading
if re.match(r"^#{1,3}\s+", line):
break
# Match numbered list items: "1. " or "1) "
m = re.match(r"^\s*\d+[.)]\s+(.+)", line)
if m:
steps.append(m.group(1).strip())
return steps
def get_changed_files() -> list[str]:
"""Return list of files changed in the last commit (HEAD~1 diff)."""
output = run_git(["diff", "HEAD~1", "--name-only"])
if not output:
# Fallback: files in most recent commit
output = run_git(["show", "--name-only", "--format=", "HEAD"])
return [f.strip() for f in output.splitlines() if f.strip()]
def keywords_from_step(step: str) -> list[str]:
"""Extract meaningful lowercase keywords (>3 chars) from a step description."""
words = re.findall(r"[a-zA-Z0-9_\-/\.]+", step)
return [w.lower() for w in words if len(w) > 3]
def step_matches_files(step: str, changed_files: list[str]) -> tuple[bool, str]:
"""Return (matched, matching_file_or_empty)."""
keywords = keywords_from_step(step)
if not keywords:
return False, ""
lower_files = [f.lower() for f in changed_files]
for keyword in keywords:
for i, lf in enumerate(lower_files):
if keyword in lf:
return True, changed_files[i]
return False, ""
def update_adr_status(adr_path: Pat#!/usr/bin/env python3
# hook-version: 1.1.0
"""SessionStart hook: inject autonomous behavioral context for AFK sessions.
Injects a posture block that instructs Claude to work proactively without
requiring confirmation (ADR-143). In the default 'always' mode it injects on
every session without detecting anything, so the block states the configured
mode; it never claims the user is or is not watching. In 'auto' mode it
injects only for SSH/tmux/screen sessions and says that is what it saw.
Environment:
CLAUDE_AFK_MODE - Control override: 'always' (default), 'auto', or 'never'
SSH_CONNECTION - Set by SSH daemon when connected via SSH
SSH_TTY - Set when SSH allocates a TTY
SSH_CLIENT - Set by SSH daemon (older form)
TMUX - Set by tmux when inside a tmux session
STY - Set by GNU screen when inside a screen session
Always exits 0 (advisory, never blocking).
"""
import sys # vexjoy-pyguard
if sys.version_info < (3, 10): # vexjoy-pyguard advisory
import os
if not os.environ.get("VEXJOY_PYGUARD_REEXEC"):
os.environ["VEXJOY_PYGUARD_REEXEC"] = "1"
for _d in [*os.environ.get("PATH", "").split(os.pathsep), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]:
for _v in ("14", "13", "12", "11", "10"):
_p = os.path.join(_d, "python3." + _v)
if _d and "/shims" not in _d and os.access(_p, os.X_OK):
os.execv(_p, [_p, *sys.argv])
sys.stderr.write(
"[vexjoy] hook %s needs Python 3.10+, got %s (%s); run ./install.sh in vexjoy-agent from a terminal"
" outside Claude Code (gated tool calls stay blocked until then)\n"
% (__file__.rsplit("/", 1)[-1], sys.version.split()[0], sys.executable)
)
sys.exit(0)
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "lib"))
from hook_utils import context_output, empty_output, hook_error
EVENT_NAME = "SessionStart"
_AFK_GUIDANCE = (
"Work proactively. Complete multi-step tasks without asking for confirmation.\n"
"If you can determine the next logical step, take it.\n"
"Produce concise task-completion summaries when finishing long-running work.\n"
"Actions that CLAUDE.md reserves for owner approval still need that approval."
)
# 'always' mode detects nothing, so the block states the setting.
AFK_CONTEXT = (
"<afk-mode>\n"
'AFK mode is set to "always" (CLAUDE_AFK_MODE), so work as if the user may not be watching '
"the terminal. This is the configured mode; nothing checked whether anyone is watching.\n"
f"{_AFK_GUIDANCE}\n"
"</afk-mode>"
)
# 'auto' mode saw an SSH/tmux/screen session, which suggests but does not prove
# that nobody is watching.
AFK_AUTO_CONTEXT = (
"<afk-mode>\n"
'AFK mode is "auto" (CLAUDE_AFK_MODE) and this session runs over SSH, tmux or screen, so work as '
"if the user may not be watching the terminal. Nothing checked whether anyone is watching.\n"
f"{_AFK_GUIDANCE}\n"
"</afk-mode>"
)
def is_afk() -> bool:
"""Return True if the session appears to be AFK (unattended).
NOTE: Do NOT use sys.stdin.isatty() or sys.stdout.isatty() here.
Claude Code pipes stdin (event JSON) and captures stdout (hook output),
so both are always non-TTY in hook context — the check would always
return True, activating AFK mode on every session regardless of type.
"""
# SSH indicators
if os.environ.get("SSH_CONNECTION"):
return True
if os.environ.get("SSH_TTY"):
return True
if os.environ.get("SSH_CLIENT"):
return True
# Multiplexer indicators
if os.environ.get("TMUX"):
return True
if os.environ.get("STY"):
return True
return False
_BYPASS_ENV = "AFK_MODE_BYPASS"
def main() -> None:
if os.environ.get(_BYPASS_ENV) == "1":
if os.environ.get("CLAUDE_HOOKS_DEBUG"):
print("[afk-mode] Bypassed via AFK_MODE_BYPASS=1", file=sys.stderr)
sys.exit(0)
mode = os.environ.get("CLAUDE_AFK_MODE", "always").strip().lower()
if mode == "always":
context_output(EVENT_NAME, AFK_CONTEXT).print_and_exit()
if mode == "never":
empty_output(EVENT_NAME).print_and_exit()
# Default: auto-detect
if is_afk():
context_output(EVENT_NAME, AFK_AUTO_CONTEXT).print_and_exit()
else:
empty_output(EVENT_NAME).print_and_exit()
if __name__ == "__main__":
try:
main()
except Exception as e:
hook_error("afk-mode", e)
finally:
sys.exit(0)
#!/usr/bin/env python3
# hook-version: 1.0.1
"""Run a VexJoy Claude hook against the Antigravity CLI (agy) hooks.json protocol.
Translates Antigravity's camelCase hook stdin payloads (`conversationId`,
`workspacePaths`, `toolCall`, `lastUserInput`, `transcriptPath`, `modelName`)
into VexJoy's Claude-compatible hook input contract (`session_id`, `cwd`,
`tool_name`, `tool_input`, `prompt`, `transcript_path`), executes the target
VexJoy hook script, and converts its exit code and stdout into Antigravity's
per-event JSON output schema:
- SessionStart / PreInvocation: `{"injectSteps": [{"systemMessage": {"systemMessage": "..."}}]}` or `{}`
- PostInvocation: `{"injectSteps": [...], "terminationBehavior": "force_continue"|"terminate"}` or `{}`
- PreToolUse: `{"decision": "allow"|"deny"|"ask"|"force_ask", "reason": "...", "overwrite": {...}}`
- PostToolUse: `{}` (advisory outputs are queued per conversationId and drained on PreInvocation)
- Stop: `{"decision": "continue"|"allow", "reason": "..."}`
stdin is read with the --timeout bound; a pipe left open past it gets the
--failure-policy answer. The deferred queue is read and rewritten under a
bounded lock with per-writer temp files. Its last-resort directory is a
per-user, owner-only (0700) one under the system temp dir; when that path is a
symlink, another user's, or open to others, nothing is queued there.
"""
from __future__ import annotations
import sys # vexjoy-pyguard
if sys.version_info < (3, 10): # vexjoy-pyguard adapter
import os
if not os.environ.get("VEXJOY_PYGUARD_REEXEC"):
os.environ["VEXJOY_PYGUARD_REEXEC"] = "1"
for _d in [*os.environ.get("PATH", "").split(os.pathsep), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]:
for _v in ("14", "13", "12", "11", "10"):
_p = os.path.join(_d, "python3." + _v)
if _d and "/shims" not in _d and os.access(_p, os.X_OK):
os.execv(_p, [_p, *sys.argv])
sys.stderr.write(
"[vexjoy] hook %s needs Python 3.10+, got %s (%s); run ./install.sh in vexjoy-agent from a terminal"
" outside Claude Code (gated tool calls stay blocked until then)\n"
% (__file__.rsplit("/", 1)[-1], sys.version.split()[0], sys.executable)
)
_a = sys.argv[sys.argv.index("--failure-policy") + 1 :] if "--failure-policy" in sys.argv else []
sys.exit(2 if _a[:1] == ["closed"] else 0)
import argparse
import copy
import json
import os
import re
import signal
import stat
import subprocess
import tempfile
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Any
try:
import fcntl
except ImportError: # Windows
fcntl = None
import msvcrt
AGY_EVENTS = (
"SessionStart",
"PreInvocation",
"PostInvocation",
"PreToolUse",
"PostToolUse",
"Stop",
)
CLAUDE_EVENT_FOR_AGY = {
"SessionStart": "SessionStart",
"PreInvocation": "UserPromptSubmit",
"PostInvocation": "Stop",
"PreToolUse": "PreToolUse",
"PostToolUse": "PostToolUse",
"Stop": "Stop",
}
FAILURE_POLICIES = ("open", "closed")
_MAX_CHILD_OUTPUT = 1_048_576
_SAFE_ID_RE = re.compile(r"[^A-Za-z0-9._-]+")
class AdapterError(RuntimeError):
"""A conversion or child-hook result cannot be represented safely."""
# Queue read-modify-writes and drains hold an exclusive lock on this file in the
# queue directory. Hooks can run in parallel; unlocked, two writers each kept
# only their own message and a drain could delete one queued after its read.
# The wait is capped: past it the operation runs unlocked rather than stall.
_QUEUE_LOCK_NAME = "deferred.lock"
_LOCK_WAIT_SECONDS = 1.0
_LOCK_POLL_SECONDS = 0.01
def _private_fallback_dir() -> Path | None:
"""The per-user queue directory under the system temp dir, made 0700.
Returns None when the path is a symlink or not a directory, belongs to
another user, or is open to group or other: advisories are then dropped
rather than written where someone else can read or redirect them.
"""
getuid = getattr(os, "getuid", None)
name = "vexjoy-agy-deferred" + (f"-{getuid()}" if getuid is not None else "")
path = Path(tempfile.gettempdir()) / name
try:
path.mkdir(mode=0o700, exist_ok=True)
info = os.lstat(path)
except OSError:
return None
if not stat.S_ISDIR(info.st_mode):
return None
if getuid is not None and (info.st_uid != getuid() or info.st_mode & 0o077):
return None
return path
def _deferred_dir() -> Path | None:
env_dir = os.environ.get("VEXJOY_AGY_DEFERRED_DIR")
if env_dir:
return Path(env_dir)
state_dir = os.environ.get("VEXJOY_STATE_DIR")
if state_dir:
return Path(state_dir) / "agy-deferred"
gemini_dir = Path.home() / ".gemini" / "state" / "agy-deferred"
try:
gemini_dir.mkdir(parents=True, exist_ok=True)
return gemini_dir
except OSError:
return _private_fallback_dir()
def _deferred_path(conversation_id: str) -> Path | None:
directory = _deferred_dir()
if directory is None:
return None
safe_id = _SAFE_ID_RE.sub("_", conversation_id.strip() or "default")[:80]
return directory / f"deferred-{safe_id}.json"
def _try_lock(fd: int) -> bool:
"""One non-blocking exclusive-lock attempt; True when the lock is held."""
try:
if fcntl is not None:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
else:
os.lseek(fd, 0, os.SEEK_SET)
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
except BlockingIOError:
return False
except OSError:
if fcntl is not None:
raise # locking unsupported here: the caller goes ahead unlocked
return False # msvcrt reports contention as a plain OSError
return True
@contextmanager
def _queue_lock(directory: Path):
"""Hold the queue lock for one operation, waiting at most _LOCK_WAIT_SECONDS.
On timeout, or where#!/usr/bin/env python3
# hook-version: 1.0.2
"""PreToolUse hook: Block gh pr merge when CI checks haven't passed.
Intercepts Bash tool calls containing 'gh pr merge' and checks GitHub
Actions status before allowing the merge. Blocks if any checks are
failing or still pending.
"""
import sys # vexjoy-pyguard
if sys.version_info < (3, 10): # vexjoy-pyguard gate
import os
if not os.environ.get("VEXJOY_PYGUARD_REEXEC"):
os.environ["VEXJOY_PYGUARD_REEXEC"] = "1"
for _d in [*os.environ.get("PATH", "").split(os.pathsep), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"]:
for _v in ("14", "13", "12", "11", "10"):
_p = os.path.join(_d, "python3." + _v)
if _d and "/shims" not in _d and os.access(_p, os.X_OK):
os.execv(_p, [_p, *sys.argv])
sys.stderr.write(
"[vexjoy] hook %s needs Python 3.10+, got %s (%s); run ./install.sh in vexjoy-agent from a terminal"
" outside Claude Code (gated tool calls stay blocked until then)\n"
% (__file__.rsplit("/", 1)[-1], sys.version.split()[0], sys.executable)
)
sys.exit(2)
import json
import os
import re
import shlex
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent / "lib"))
from hook_utils import deny_tool_use, record_governance
from stdin_timeout import read_stdin
_MERGE_RE = re.compile(r"\bgh\s+pr\s+merge\b")
# Tokens that end the merge simple command.
_SEGMENT_END = {"&&", "||", ";", "|", "&", "\n", "(", ")"}
# Merge flags that take a value; the value is never the PR number.
_VALUE_FLAGS = {
"-t",
"--subject",
"-b",
"--body",
"-F",
"--body-file",
"-A",
"--author-email",
"--match-head-commit",
"-R",
"--repo",
}
_PULL_URL_RE = re.compile(r"/pull/(\d+)")
def extract_pr_number(command: str) -> str | None:
"""Return the PR number from the first merge command's argument list.
Parses only the tokens after the merge subcommand up to the first shell
separator or redirection. Returns None when no numeric or URL selector
is given (the caller then falls back to the current branch).
"""
m = _MERGE_RE.search(command)
if not m:
return None
rest = command[m.end() :]
# Drop fd prefixes on redirects (`2>&1`) so the fd is not read as an argument.
rest = re.sub(r"(?<!\S)\d+(?=[<>])", " ", rest)
lex = shlex.shlex(rest, posix=True, punctuation_chars=True)
lex.whitespace_split = True
lex.commenters = "" # `#55` is a PR selector, not a comment
try:
tokens = list(lex)
except ValueError:
tokens = rest.split()
skip_next = False
for tok in tokens:
if tok in _SEGMENT_END or tok[:1] in "<>" or tok.startswith(("&", "|", ";")):
break
if skip_next:
skip_next = False
continue
if tok.startswith("-"):
if tok in _VALUE_FLAGS:
skip_next = True
continue
# First positional argument is the PR selector.
if tok.lstrip("#").isdigit():
return tok.lstrip("#")
url = _PULL_URL_RE.search(tok)
return url.group(1) if url else None
return None
_BYPASS_ENV = "CI_MERGE_GATE_BYPASS"
def main() -> None:
if os.environ.get(_BYPASS_ENV) == "1":
if os.environ.get("CLAUDE_HOOKS_DEBUG"):
print("[ci-merge-gate] Bypassed via CI_MERGE_GATE_BYPASS=1", file=sys.stderr)
sys.exit(0)
data = json.loads(read_stdin(timeout=2))
# tool_name filter removed — matcher "Bash" in settings.json prevents
# this hook from spawning for non-Bash tools.
command = data.get("tool_input", {}).get("command", "")
# Only intercept gh pr merge commands
if "gh pr merge" not in command and "gh pr merge" not in command.replace(" ", " "):
return
parts = command.split()
# --- Block --admin before any CI check ---
if "--admin" in parts:
if os.environ.get("ALLOW_ADMIN_MERGE") == "1":
print("[ci-merge-gate] WARNING: --admin override allowed via ALLOW_ADMIN_MERGE=1", file=sys.stderr)
else:
print("[ci-merge-gate] BLOCKED: --admin bypasses branch protection", file=sys.stderr)
deny_tool_use("PreToolUse", "Use of --admin bypasses CI checks. Remove --admin and wait for CI to pass.")
sys.exit(0)
# --- Block --force before any CI check ---
if "--force" in parts:
if os.environ.get("ALLOW_FORCE_MERGE") == "1":
print("[ci-merge-gate] WARNING: --force override allowed via ALLOW_FORCE_MERGE=1", file=sys.stderr)
else:
print("[ci-merge-gate] BLOCKED: --force bypasses merge safeguards", file=sys.stderr)
deny_tool_use("PreToolUse", "Use of --force bypasses merge safeguards. Remove --force and merge normally.")
sys.exit(0)
# PR number comes from the merge-command arguments only, never from
# other commands in the chain (`sleep 5 && <merge> 1012`).
pr_number = extract_pr_number(command)
if not pr_number:
# No PR number found — might be merging current branch PR
# Try to get it from current branch
try:
result = subprocess.run(
["gh", "pr", "view", "--json", "number", "--jq", ".number"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0 and result.stdout.strip().isdigit():
pr_number = result.stdout.strip()
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
if not pr_number:
# Can't determine PR number — let it through with a warning
print("[ci-merge-gate] WARNING: Could not determine PR number. Skipping CI check.", file=sys.stderr)
return
# Check CI status
try:
result = subprocess.run(
["gh", "pr", "checks", pr_number, "--json", "name,All 83 scripts are listed above. The source is inlined for 6 of them, starting with whatever hooks.json actually runs. See all of them in the repo.
Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.
Essays and writing behind this toolkit live at vexjoy.com. VexJoy Agent connects plain-English requests to specialist agents, skills, and workflows. /do selects the knowledge and tools needed for your task.
Stars on GitHub
A development methodology for coding agents that makes the agent brainstorm before building, follow plans, run parallel work and finish branches in a disciplined way. You get fewer skipped steps and more consistent results from your coding assistant.
@obraA Claude Code plugin with skills for auditing a whole repo, reviewing a change, listing deferred shortcut comments, and showing measured savings and a quick command reference.
@dietrichgebertWalks a software change from a written spec through design, testing, code review, CI/CD, and release, applying the matching engineering skill at each stage. Useful for developers using Claude Code on full development workflows.
@addyosmaniPersistent file-based planning for AI coding agents on long-running, multi-step tasks. It keeps task_plan.md, findings.md and progress.md on disk, so work can resume after /clear or context compaction.
@othmanadiA spec-driven development workflow for Claude Code that guides a project through planning, phases, milestone audits and fixes. Includes skills to generate tests from UAT criteria and to audit completed work.
@open-gsdChinese-language skill pack for Claude Code that adds workflow guidance for brainstorming, TDD, debugging, and code review, plus Chinese conventions for commits, documentation, and domestic Git platforms like Gitee and Coding.net.
@jnmetacodeGives Claude Code semantic search, dependency graph and architecture context over a codebase, so an AI assistant can explore and understand a project's code. Includes skills for indexing a project and managing its file watching.
@giancarloerraA lightweight code search engine that uses AST-based semantic search with local embeddings and incremental indexing. It helps Claude Code users find code faster and more efficiently.
@cocoindex-io