hooks/enforce-git-workflow.py

#!/usr/bin/env python3
"""
PreToolUse:Bash hook to enforce GitHub Issues workflow.

Classification (plan.md §5 Epic 1): bypass-retained. Protected-branch
protections are NOT permission noise — they are the workflow contract.
`deny()` migrated to `hook_utils.hard_block()` (exit 2) so the rules
survive bypass mode. `ALLOW_MAIN_COMMIT=1` is the explicit escape hatch.

BLOCKS (via hard_block, bypass-proof):
1. Commits on protected branches (must use feature branch)
2. Commit messages without issue number prefix (^#\\d+:)
3. Direct pushes to protected branches (must use PR workflow)

ALLOWS:
- sync: prefix for log/coordination commits (no issue number needed)
- Merge commits and --amend without -m skip validation
- ALLOW_MAIN_COMMIT=1 env var for emergencies
- Not in a git repo → allow all
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_utils  # noqa: E402

# ─── Protected branches ──────────────────────────────────────────────
# Any branch matching one of these names (case-insensitive) is blocked
# from direct commits and pushes. Use feature branches + PRs instead.
#
# To add custom branches, append to this list or use the setup script
# which writes them to a config file.
PROTECTED_BRANCHES = [
    "dev",
    "develop",
    "main",
    "master",
    "prod",
    "production",
    "release",
    "stag",
    "staging",
    "trunk",
]

# Load user-defined protected branches from config file (written by setup.sh)
_CUSTOM_BRANCHES_FILE = os.path.expanduser(
    "~/.claude/git-flow-protected-branches.json"
)
try:
    with open(_CUSTOM_BRANCHES_FILE) as f:
        _custom = json.load(f)
        if isinstance(_custom, list):
            PROTECTED_BRANCHES.extend(_custom)
except (FileNotFoundError, json.JSONDecodeError, TypeError):
    pass

# Normalize to lowercase set for O(1) lookup
PROTECTED_BRANCHES_SET = {b.lower() for b in PROTECTED_BRANCHES}

# ─── Direct-to-main repos ────────────────────────────────────────────
# Repos that use direct-to-main workflow (no feature branches, no issue numbers).
# Matched as substrings against the git remote URL. Populate via
# ~/.claude/git-flow-direct-to-main-repos.json: e.g. ["myorg/dotfiles"].
DIRECT_TO_MAIN_REPOS: list[str] = []

_DIRECT_TO_MAIN_FILE = os.path.expanduser(
    "~/.claude/git-flow-direct-to-main-repos.json"
)
try:
    with open(_DIRECT_TO_MAIN_FILE) as f:
        _d2m = json.load(f)
        if isinstance(_d2m, list):
            DIRECT_TO_MAIN_REPOS.extend(str(x) for x in _d2m)
except (FileNotFoundError, json.JSONDecodeError, TypeError):
    pass


def is_direct_to_main_repo() -> bool:
    """Check if the current repo is allowlisted for direct-to-main workflow."""
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=2,
        )
        if result.returncode == 0:
            remote = result.stdout.strip()
            return any(repo in remote for repo in DIRECT_TO_MAIN_REPOS)
    except Exception:
        pass
    return False


def is_protected_branch(branch: str) -> bool:
    """Check if a branch name is in the protected list (case-insensitive)."""
    return branch.lower() in PROTECTED_BRANCHES_SET


def get_current_branch() -> str | None:
    """Get current git branch. Returns None if not in a git repo."""
    try:
        result = subprocess.run(
            ["git", "branch", "--show-current"],
            capture_output=True,
            text=True,
            timeout=2,
        )
        if result.returncode == 0:
            return result.stdout.strip()
        return None
    except Exception:
        return None


def extract_commit_message(command: str) -> str | None:
    """Extract commit message from git commit -m '...' or --message '...'."""
    # Heredoc patterns (e.g. -m "$(cat <<'EOF' ... EOF)") can't be reliably
    # parsed from the raw command string — skip validation and let git handle it.
    # This check MUST come before the regex patterns, which would otherwise
    # match the outer quotes and capture the raw heredoc syntax as the "message".
    if "-m" in command and "<<" in command:
        return None

    # Try quoted patterns first (single or double quotes)
    patterns = [
        r"""-m\s+"([^"]+)" """,
        r"""-m\s+'([^']+)' """,
        r'''--message\s+"([^"]+)"''',
        r"""--message\s+'([^']+)'""",
        # Unquoted (grab until next flag or end)
        r"""-m\s+(\S+)""",
    ]
    # Append a space to command to make patterns match at end of string
    cmd = command + " "
    for pattern in patterns:
        match = re.search(pattern, cmd)
        if match:
            return match.group(1)

    return None


def is_commit_command(command: str) -> bool:
    """Check if this is a git commit command."""
    # Match: git commit, git commit -m, git commit --amend, etc.
    # But NOT: git commit-graph, git commit-tree
    return bool(re.match(r"git\s+commit(\s|$)", command))


def is_push_command(command: str) -> bool:
    """Check if this is a git push command."""
    return bool(re.match(r"(SKIP_CHECKS=\S+\s+)?git\s+push(\s|$)", command))


def is_merge_commit(command: str) -> bool:
    """Check if this is a merge commit (auto-generated message)."""
    return "Merge branch" in command or "Merge pull request" in command


def is_amend_without_message(command: str) -> bool:
    """Check if this is --amend that preserves existing message."""
    return "--amend" in command and "-m" not in command and "--message" not in command


def is_sync_commit(message: str | None) -> bool:
    """Check if this is a sync/coordination commit (no issue number needed)."""
    if not message:
        return False
    return bool(re.match(r"^sync:", message, re.IGNORECASE))


def is_auto_commit(message: str | None) -> bool:
    """Check if this is an autoheal apply-path commit (`#auto:` prefix).

    autoheal's `lib/apply-proposal.py` (and the future /autoheal-apply
    command) write commits with `#auto: apply autoheal proposal {id}`
    when a proposal is mechanically applied to a feature branch. These
    are not issue commits and never run on main; the format is
    distinguishable enough to allow without weakening the issue-number
    rule for human commits.
    """
    if not message:
        return False
    return bool(re.match(r"^#auto:", message))


def validate_commit_message_format(message: str | None) -> bool:
    """Check if commit message starts with issue number: '#42: description'."""
    if not message:
        return True  # No message to validate (e.g., heredoc), skip
    return bool(re.match(r"^#\d+:", message))


def deny(reason: str) -> None:
    """Bypass-proof hard block. Exits 2 with reason on stderr.

    `permissionDecision: deny` from a PreToolUse hook is silently
    overridden by any later `ask` decision (GitHub issue #39344). The
    only signal Claude Code honors regardless of `permission_mode` is
    `exit 2`. `hook_utils.hard_block()` wraps that primitive.
    """
    hook_utils.hard_block(reason)


def check_commit(command: str, branch: str) -> None:
    """Validate git commit commands against workflow rules."""
    # Direct-to-main repos skip all checks
    if is_direct_to_main_repo():
        return

    # Emergency bypass
    if os.environ.get("ALLOW_MAIN_COMMIT") == "1":
        print(
            "WARNING: ALLOW_MAIN_COMMIT bypass active - skipping all branch protections",
            file=sys.stderr,
        )
        return

    # Rule 1: No commits on protected branches
    if is_protected_branch(branch):
        deny(
            f"Cannot commit on protected branch '{branch}'. "
            "Create a feature branch first:\n"
            "  1. gh issue create  (if no issue exists)\n"
            "  2. git checkout -b {issue#}-{description} origin/main\n"
            "  3. Then commit with '#<issue>: description'\n\n"
            "Emergency bypass: ALLOW_MAIN_COMMIT=1 git commit ..."
        )

    # Skip message validation for merge commits and amends without new message
    if is_merge_commit(command) or is_amend_without_message(command):
        return

    # Rule 2: Commit message must have issue number prefix (or sync: / #auto: prefix)
    message = extract_commit_message(command)
    if (
        message
        and not is_sync_commit(message)
        and not is_auto_commit(message)
        and not validate_commit_message_format(message)
    ):
        deny(
            f'Commit message must start with issue number.\n'
            f'  Required format: "#{{issue_number}}: {{description}}"\n'
            f'  Example: "#42: Fix the login bug"\n'
            f'  Also allowed: "sync: update session log"\n'
            f'  Also allowed (autoheal apply path): "#auto: apply autoheal proposal {{id}}"\n'
            f'  Your message: "{message}"'
        )


def check_push(command: str, branch: str) -> None:
    """Validate git push commands against workflow rules."""
    # Direct-to-main repos skip all checks
    if is_direct_to_main_repo():
        return

    # Emergency bypass
    if os.environ.get("ALLOW_MAIN_COMMIT") == "1":
        print(
            "WARNING: ALLOW_MAIN_COMMIT bypass active - skipping all branch protections",
            file=sys.stderr,
        )
        return

    # Only block pushes when on a protected branch
    if not is_protected_branch(branch):
        return

    # Parse what's being pushed to determine if it's actually pushing
    # the protected branch or an explicit different branch.
    # Allow: git push origin feature-branch (explicitly pushing a different branch)
    parts = command.split()

    # Find the refspec (what's being pushed)
    # git push origin branch-name -> pushing branch-name
    # git push -u origin HEAD -> pushing current branch
    # git push -> pushing current branch
    push_target = None
    skip_next = False
    found_remote = False

    for i, part in enumerate(parts):
        if skip_next:
            skip_next = False
            continue
        if part in ("-u", "--set-upstream", "--force", "-f", "--no-verify",
                     "--force-with-lease", "--delete", "-d"):
            if part in ("-u", "--set-upstream"):
                skip_next = False  # -u takes no argument by itself
            continue
        if part.startswith("-"):
            continue
        if part in ("git", "push"):
            continue
        # Skip env var prefix like SKIP_CHECKS=1
        if "=" in part:
            continue
        if not found_remote:
            found_remote = True  # First non-flag arg is the remote
            continue
        # Second non-flag arg is the refspec
        push_target = part
        break

    # HEAD means current branch, explicit branch name could differ
    if push_target == "HEAD":
        push_target = branch  # Resolves to current (protected) branch
    elif push_target is None:
        # No explicit target -> defaults to current branch
        push_target = branch
    elif not is_protected_branch(push_target):
        return  # Pushing a specific non-protected branch, allow

    deny(
        f"Cannot push to protected branch '{branch}'. "
        "Use the feature branch + PR workflow:\n"
        "  1. git checkout -b {issue-number}-{description}\n"
        "  2. Make changes and commit\n"
        "  3. git push -u origin HEAD\n"
        "  4. gh pr create\n"
        "  5. gh pr merge --squash --delete-branch\n\n"
        "Emergency bypass: ALLOW_MAIN_COMMIT=1 git push ..."
    )


def main() -> None:
    input_data = hook_utils.read_hook_input()

    tool_name = input_data.get("tool_name", "")
    tool_input = input_data.get("tool_input", {})

    if tool_name != "Bash":
        sys.exit(0)

    command = tool_input.get("command", "").strip()
    if not command:
        sys.exit(0)

    # Only check git commit and git push commands
    if not (is_commit_command(command) or is_push_command(command)):
        sys.exit(0)

    # Get current branch (None if not in git repo)
    branch = get_current_branch()
    if not branch:
        sys.exit(0)  # Not in git repo, allow

    if is_commit_command(command):
        check_commit(command, branch)
    elif is_push_command(command):
        check_push(command, branch)

    # No deny → exit silently, let other hooks handle allow/deny
    sys.exit(0)


if __name__ == "__main__":
    main()

hooks/enforce-issue-workflow.py

#!/usr/bin/env python3
"""
UserPromptSubmit hook to enforce issue-first workflow.

This hook detects work requests and injects a reminder into Claude's context
to ensure the issue-first workflow is followed before making changes.

Scoped to ~/code/ — the reminder only fires when cwd is under that directory.
Outside that scope (e.g., a note-taking vault or other non-code working
directory), the hook stays silent. Coordination injection is additionally
conditional on .claude/logs/ existing in the current working directory.
"""

from __future__ import annotations

import json
import os
import re
import sys


def is_work_request(prompt: str) -> bool:
    """Detect if the prompt is a work request vs a question or research task."""
    prompt_lower = prompt.lower()

    # Work action verbs that indicate implementation tasks
    work_patterns = [
        r"\b(update|fix|add|create|implement|build|change|modify|refactor)\b",
        r"\b(write|make|set up|setup|configure|migrate|convert|move)\b",
        r"\b(delete|remove|rename|replace|upgrade|downgrade)\b",
        r"\b(enable|disable|install|uninstall)\b",
    ]

    # Patterns that indicate it's NOT a work request (questions, research)
    question_patterns = [
        r"^(what|why|how|where|when|which|who|can you explain|tell me)\b",
        r"\?$",  # Ends with question mark
        r"\b(explain|describe|show me|list|find|search|look for|check)\b",
    ]

    # Check if it looks like a question first
    for pattern in question_patterns:
        if re.search(pattern, prompt_lower):
            return False

    # Check if it matches work patterns
    for pattern in work_patterns:
        if re.search(pattern, prompt_lower):
            return True

    return False


def has_logs_directory() -> bool:
    """Check if .claude/logs/ exists in the current working directory."""
    return os.path.isdir(os.path.join(os.getcwd(), ".claude", "logs"))


def is_in_code_dir() -> bool:
    """Scope check: only fire the reminder when cwd is under ~/code/."""
    code_root = os.path.realpath(os.path.expanduser("~/code"))
    cwd = os.path.realpath(os.getcwd())
    return cwd == code_root or cwd.startswith(code_root + os.sep)


def build_reminder() -> str:
    """Build the workflow reminder, with optional coordination section."""
    if has_logs_directory():
        return """
<workflow-reminder>
STOP - Before making ANY code or file changes, you MUST:

1. CHECK: Does a GitHub issue exist for this work?
   - If NO: Create one first with `gh issue create`
   - If YES: Note the issue number

2. CREATE BRANCH: `git checkout -b {issue#}-{description} origin/main`

3. CHECK COORDINATION: Read `.claude/logs/` for other active sessions
   - Look for today's date directory: `.claude/logs/YYYYMMDD/`
   - Read other agents' logs to check for file conflicts
   - If overlap detected, note it but proceed (advisory, not blocking)

4. LOG YOUR SESSION: Create/update `.claude/logs/YYYYMMDD/agent-N.md`

5. IMPLEMENT: Make your changes

6. COMMIT & PR:
   - `git commit -m "#{issue_number}: {description}"`
   - `gh pr create --title "#{issue_number}: ..." --body "Closes #{issue_number}"`

This applies to ALL changes including documentation, config, and "trivial" fixes.
Do NOT skip this workflow. The user has explicitly requested strict adherence.
</workflow-reminder>
"""
    else:
        return """
<workflow-reminder>
STOP - Before making ANY code or file changes, you MUST:

1. CHECK: Does a GitHub issue exist for this work?
   - If NO: Create one first with `gh issue create`
   - If YES: Note the issue number

2. CREATE BRANCH: `git checkout -b {issue#}-{description} origin/main`

3. IMPLEMENT: Make your changes

4. COMMIT & PR:
   - `git commit -m "#{issue_number}: {description}"`
   - `gh pr create --title "#{issue_number}: ..." --body "Closes #{issue_number}"`

This applies to ALL changes including documentation, config, and "trivial" fixes.
Do NOT skip this workflow. The user has explicitly requested strict adherence.
</workflow-reminder>
"""


def main() -> None:
    try:
        input_data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)  # Silent failure, don't block

    prompt = input_data.get("prompt", "")

    if is_work_request(prompt) and is_in_code_dir():
        print(build_reminder())

    sys.exit(0)


if __name__ == "__main__":
    main()

hooks/auto-approve-bash.py

#!/usr/bin/env python3
"""
PreToolUse hook that enforces Bash permissions from settings.json.

Classification (plan.md §5 Epic 1): bypass-suppressible for allow/deny pattern
matching; the curated destructive set and smart-rules run OUTSIDE the bypass
short-circuit so they survive bypass mode via `hook_utils.hard_block()`.

Execution order in main():
  1. curated destructive set (always run; hard_block, bypass-proof)
  2. force branch delete (always run; hard_block, bypass-proof)
  3. smart-rules (always run; destructive-reset → hard_block, bypass-proof)
  4. bypass-mode short-circuit (exit 0)
  5. settings.json allow/deny pattern matching (per-segment)

Deny/allow matching is PER SEGMENT: the command is decomposed on &&, ||, ;,
|, newlines, and command substitution before matching, so a chained command
like "echo hi && rm -rf /" cannot smuggle a denied/dangerous segment past
the list (GitHub issue #660).

This hook exists because Claude Code's built-in permission system has bugs:
- Issue #15921: VSCode extension ignores Bash permissions
- Issue #13340: Piped commands bypass permission allowlist

This hook properly implements the allow/deny logic that settings.json SHOULD provide.
"""
from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_utils  # noqa: E402

# Settings file locations in precedence order (highest first)
SETTINGS_FILES = [
    Path.home() / ".claude" / "settings.json",
    # Add project-level settings if needed
]

def load_settings() -> tuple[list[str], list[str]]:
    """Load and merge settings from all settings files."""
    allow_patterns: list[str] = []
    deny_patterns: list[str] = []

    for settings_file in SETTINGS_FILES:
        if settings_file.exists():
            try:
                with open(settings_file, 'r') as f:
                    settings = json.load(f)
                    permissions = settings.get("permissions", {})

                    # Extract Bash patterns from allow list
                    for rule in permissions.get("allow", []):
                        if rule.startswith("Bash(") and rule.endswith(")"):
                            pattern = rule[5:-1]  # Extract pattern from Bash(...)
                            allow_patterns.append(pattern)

                    # Extract Bash patterns from deny list
                    for rule in permissions.get("deny", []):
                        if rule.startswith("Bash(") and rule.endswith(")"):
                            pattern = rule[5:-1]  # Extract pattern from Bash(...)
                            deny_patterns.append(pattern)
            except (json.JSONDecodeError, IOError):
                continue

    return allow_patterns, deny_patterns

def pattern_matches_command(pattern: str, command: str) -> bool:
    """
    Check if a permission pattern matches a command.

    Patterns use prefix matching with :* as wildcard suffix.
    Examples:
        - "mkdir:*" matches "mkdir -p /foo/bar"
        - "git status:*" matches "git status"
        - "npm run lint" matches exactly "npm run lint" or "npm run lint ..."

    NOTE: This matches a SINGLE command segment. Whole-command matching is
    done by check_pattern_decision(), which first decomposes a chained
    command into segments via split_command_segments() and applies this
    matcher to each. A bare prefix match against a whole chained string
    (e.g. "echo hi && rm -rf /") would let a benign leading command smuggle
    a dangerous trailing one past the deny list — see GitHub issue #660.
    """
    command = command.strip()

    # Handle :* wildcard suffix (matches anything after the prefix)
    if pattern.endswith(":*"):
        prefix = pattern[:-2]  # Remove :*
        return command.startswith(prefix)

    # Handle patterns with space before * (e.g., "mkdir *")
    if pattern.endswith(" *"):
        prefix = pattern[:-2]  # Remove " *"
        return command.startswith(prefix)

    # Exact match or prefix match
    return command.startswith(pattern)


# Shell tokens that separate one command from the next. A deny pattern must
# be evaluated against every segment, not just the head of the whole string,
# or chaining slips dangerous commands past the deny list (issue #660).
_SEGMENT_SEPARATORS = re.compile(r"&&|\|\||[;|\n]")

# Command-substitution forms: $( ... ) and `...`. Their inner command runs,
# so it must be decomposed and checked just like a top-level segment.
_SUBSTITUTION = re.compile(r"\$\(([^()]*)\)|`([^`]*)`")


def split_command_segments(command: str) -> list[str]:
    """
    Decompose a shell command string into independently-runnable segments.

    Splits on the operators that chain or pipe commands (`&&`, `||`, `;`,
    `|`, newlines) and lifts out command substitutions (`$(...)`, backticks)
    so their inner commands are evaluated too. Every returned segment is a
    candidate command that the shell would actually execute; the deny list
    must be checked against each one.

    This is a best-effort lexical split, not a full shell parser. It does not
    track quoting, so a separator inside a quoted string is still treated as a
    separator. For a SECURITY deny check that bias is correct: over-splitting
    only produces extra segments to inspect, never fewer.

    Returns a flat list of non-empty, stripped segments. Always returns at
    least one element for a non-empty command.
    """
    work = [command]
    out: list[str] = []
    while work:
        current = work.pop()
        # Lift any command substitutions into their own segments, replacing
        # the substitution text with a space so the surrounding command is
        # still inspected as its own segment.
        subs = list(_SUBSTITUTION.finditer(current))
        if subs:
            for m in subs:
                inner = m.group(1) if m.group(1) is not None else m.group(2)
                if inner and inner.strip():
                    work.append(inner)
            current = _SUBSTITUTION.sub(" ", current)
        for part in _SEGMENT_SEPARATORS.split(current):
            part = part.strip()
            if part:
                out.append(part)
    return out or ([command.strip()] if command.strip() else [])


# Curated set of destructive command shapes that must be hard-blocked even in
# bypass-permission mode. These mirror the destructive-git-reset smart-rule:
# each is checked ABOVE the bypass short-circuit and promoted to
# hook_utils.hard_block() (exit 2), the only mechanism that survives bypass
# mode (GitHub issue #39344). Patterns are deliberately narrow — they target
# whole-disk / root destruction, not ordinary cleanup like `rm -rf ./build`.
_DESTRUCTIVE_PATTERNS: list[tuple[str, "re.Pattern[str]"]] = [
    # rm with recursive+force flags targeting the filesystem root or a
    # top-level system dir. Matches combined (-rf/-fr) and split (-r ... -f)
    # flag forms; the flag run must contain both r and f.
    (
        "recursive root delete",
        re.compile(
            r"\brm\b(?=(?:\s+-[a-zA-Z]*)*\s+-[a-zA-Z]*r[a-zA-Z]*\b)"
            r"(?=(?:\s+-[a-zA-Z]*)*\s+-[a-zA-Z]*f[a-zA-Z]*\b)"
            r".*?\s(?:/|/\*|~|\$HOME|/etc|/usr|/bin|/var|/boot|/lib|/sys|/dev)"
            r"(?:/\S*)?\s*$"
        ),
    ),
    # Filesystem creation over a block device (mkfs, mkfs.ext4, etc.).
    ("filesystem format (mkfs)", re.compile(r"\bmkfs(?:\.[a-z0-9]+)?\b")),
    # Raw disk write via dd to a device node.
    ("raw disk write (dd of=/dev)", re.compile(r"\bdd\b[^\n]*\bof=/dev/")),
    # Overwriting a partition / whole disk device directly.
    ("device overwrite", re.compile(r"\bof=/dev/(?:sd|hd|nvme|disk|mapper)\S*")),
    # Forking bomb.
    ("fork bomb", re.compile(r":\s*\(\s*\)\s*\{\s*:\s*\|\s*:")),
    # Overwriting a whole disk via shred.
    ("disk shred", re.compile(r"\bshred\b[^\n]*\s/dev/")),
]


def check_destructive(command: str) -> tuple[str | None, str | None]:
    """
    Detect curated destructive command shapes in any segment of `command`.

    Run ABOVE the bypass short-circuit in main(); a hit is promoted to
    hook_utils.hard_block() (exit 2) so it cannot be bypassed by
    --dangerously-skip-permissions. Decomposes the command first so a
    destructive segment hidden behind a benign one (e.g.
    "echo go && rm -rf /") is still caught.

    Returns (label, reason) on the first destructive segment, else (None, None).
    """
    for segment in split_command_segments(command):
        for label, regex in _DESTRUCTIVE_PATTERNS:
            if regex.search(segment):
                return (
                    label,
                    f"Refusing destructive command ({label}): {segment!r}. "
                    "This is hard-blocked even in bypass-permission mode. "
                    "If this is intentional, run it manually outside Claude Code.",
                )
    return (None, None)


# Escape hatch for a genuinely-intended force delete of an unmerged branch.
# Honored from the environment or inline on the command, matching the
# ALLOW_MAIN_COMMIT convention in branch-guard.py / enforce-git-workflow.py.
_FORCE_DELETE_HATCH = "ALLOW_BRANCH_FORCE_DELETE"
_TRUTHY = frozenset({"1", "true", "yes"})


def _git_branch_args(segment: str) -> list[str] | None:
    """Return the argument tokens of a `git branch` segment, else None.

    Skips the global options that may sit between `git` and the subcommand
    (`-C <path>`, `-c key=val`, `--git-dir=...`), so `git -C /repo branch -D x`
    is recognized. Anything that is not a `git branch` invocation — including a
    segment that merely mentions the string in a grep pattern or an echo —
    returns None.
    """
    tokens = segment.split()
    if not tokens or tokens[0] != "git":
        return None
    i = 1
    while i < len(tokens):
        tok = tokens[i]
        if tok in ("-C", "-c"):
            i += 2  # consumes its value
            continue
        if tok.startswith("--") or tok.startswith("-"):
            i += 1
            continue
        break
    if i >= len(tokens) or tokens[i] != "branch":
        return None
    return tokens[i + 1:]


def is_force_branch_delete(segment: str) -> bool:
    """True if `segment` is a `git branch` invocation that force-deletes.

    Covers every spelling git accepts, not just the literal `-D` prefix the
    settings.json deny pattern matches: `-D`, `-d -f`, `--delete --force`,
    combined short flags (`-Df`, `-fd`), and any order.
    """
    args = _git_branch_args(segment)
    if args is None:
        return False
    has_delete = has_force = False
    for tok in args:
        if tok == "--":
            break
        if tok.startswith("--"):
            if tok == "--delete":
                has_delete = True
            elif tok == "--force":
                has_force = True
        elif tok.startswith("-") and len(tok) > 1:
            letters = tok[1:]
            if "D" in letters:
                has_delete = has_force = True
            if "d" in letters:
                has_delete = True
            if "f" in letters:
                has_force = True
    return has_delete and has_force


def check_force_branch_delete(command: str) -> tuple[str | None, str | None]:
    """Block `git branch` force-deletes, naming the segment and the way out.

    Runs ABOVE the bypass short-circuit in main() so the reason reaches the
    agent in bypassPermissions sessions. There, the settings.json pattern
    check never runs, and Claude Code's own deny enforcement emits a generic
    "Permission to use Bash with command <whole chain> has been denied" — which
    reads as if worktree removal were blocked and makes agents abandon teardown
    (GitHub issue #907).

    Returns (segment, reason) on the first force-delete segment, else
    (None, None).
    """
    if os.environ.get(_FORCE_DELETE_HATCH, "").strip().lower() in _TRUTHY:
        return (None, None)
    if f"{_FORCE_DELETE_HATCH}=1" in command:
        return (None, None)

    for segment in split_command_segments(command):
        if not is_force_branch_delete(segment):
            continue
        return (
            segment,
            f"Force-deleting a branch is blocked. ONLY this segment is at fault: {segment!r}\n"
            "Every other segment of your command is permitted — `git worktree remove` and "
            "`git worktree prune` are both on the allow list. Re-run them without this segment "
            "and they will succeed.\n"
            "\n"
            "`git branch -D` can discard commits that exist nowhere else, so it is never "
            "permitted ad hoc. Use the janitor, which deletes a branch only after verifying "
            "the default branch already contains its work (merged OR squash-merged):\n"
            "\n"
            "  # tear down one worktree: remove it, prune, delete its branch if absorbed\n"
            "  bash ~/.claude/lib/worktree-sweep.sh --worktree <path-to-worktree>\n"
            "\n"
            "  # worktree already gone? sweep leftover absorbed branches\n"
            "  bash ~/.claude/lib/worktree-sweep.sh --merged-branches\n"
            "\n"
            "(That script ships with the git-worktrees module. If it is not installed, "
            "`git branch -d` still deletes a normally-merged branch; a squash-merged one "
            "needs the hatch below.)\n"
            "\n"
            "If the branch is genuinely unmerged and you intend to discard its commits, say so "
            f"explicitly: `{_FORCE_DELETE_HATCH}=1 git branch -D <branch>`.",
        )
    return (None, None)


def check_smart_rules(command: str) -> tuple[str | None, str | None]:
    """
    Context-aware rules for commands that are safe in some forms but dangerous in others.
    These run BEFORE settings.json patterns AND before the bypass-mode short-circuit.

    Returns: (decision, reason) where decision is:
      - "allow": auto-approve (e.g. reset to a remote ref)
      - "hard_block": destroy-loca-history pattern; main() promotes to hard_block()
      - None: fall through to bypass / pattern matching
    """
    # git reset --hard: allow when targeting a remote ref, hard-block otherwise.
    # Safe: git reset --hard origin/main, git reset --hard origin/development
    # Dangerous: git reset --hard (bare), git reset --hard HEAD~3, git reset --hard <local-ref>
    reset_match = re.search(r'\bgit\s+reset\s+--hard\b', command)
    if reset_match:
        if re.search(r'\bgit\s+reset\s+--hard\s+origin/', command):
            return ("allow", "git reset --hard to remote ref (safe)")
        # Also allow git -C <path> reset --hard origin/
        if re.search(r'\bgit\s+-C\s+\S+\s+reset\s+--hard\s+origin/', command):
            return ("allow", "git reset --hard to remote ref in subdir (safe)")
        return (
            "hard_block",
            "git reset --hard without remote ref is blocked. "
            "Use 'git reset --hard origin/<branch>' to reset to a remote ref, "
            "or 'git pull --ff-only' to sync.",
        )

    return (None, None)


def check_pattern_decision(command: str, allow_patterns: list[str], deny_patterns: list[str]) -> tuple[str | None, str | None]:
    """
    Allow/deny decision from settings.json patterns, evaluated PER SEGMENT.

    The command is decomposed into segments (split_command_segments) so that
    chaining cannot smuggle a denied command past the list. Rules:

      - DENY if ANY segment matches ANY deny pattern. A chain is only as safe
        as its most dangerous link.
      - ALLOW only if EVERY segment matches some allow pattern. One
        un-allowed segment means the whole chain falls through (returns None),
        so the caller's normal permission flow still applies to it.

    Smart-rules and the destructive set are NOT consulted here — main() runs
    those first so their hard_block fires bypass-proof.

    Returns: (decision, reason) where decision is "allow", "deny", or None.
    """
    segments = split_command_segments(command)

    # DENY wins: any denied segment blocks the whole command.
    for segment in segments:
        for pattern in deny_patterns:
            if pattern_matches_command(pattern, segment):
                return (
                    "deny",
                    f"Command segment {segment!r} matches deny pattern: {pattern}",
                )

    # ALLOW only when every segment is explicitly allowed.
    if allow_patterns and segments:
        all_allowed = all(
            any(pattern_matches_command(p, seg) for p in allow_patterns)
            for seg in segments
        )
        if all_allowed:
            return ("allow", "All command segments match allow patterns")

    return (None, None)


def main() -> None:
    input_data = hook_utils.read_hook_input()

    tool_name = input_data.get("tool_name", "")
    tool_input = input_data.get("tool_input", {})

    # Only handle Bash commands
    if tool_name != "Bash":
        sys.exit(0)

    command = tool_input.get("command", "")
    if not command:
        sys.exit(0)

    # 1. Curated destructive set runs first and OUTSIDE the bypass
    #    short-circuit so whole-disk / root destruction hard-blocks even in
    #    bypass mode. Checked per-segment so chaining cannot hide it.
    destructive_label, destructive_reason = check_destructive(command)
    if destructive_label:
        hook_utils.hard_block(destructive_reason or "destructive command blocked")

    # 2. Force branch-delete runs OUTSIDE the bypass short-circuit too. In
    #    bypass mode the pattern check below never runs, so this is the only
    #    place an actionable reason can reach the agent (issue #907).
    fbd_segment, fbd_reason = check_force_branch_delete(command)
    if fbd_segment:
        hook_utils.hard_block(fbd_reason or "force branch delete blocked")

    # 3. Smart-rules run next, also OUTSIDE the bypass short-circuit so the
    #    destructive-reset hard-blocks even in bypass mode.
    smart_decision, smart_reason = check_smart_rules(command)
    if smart_decision == "hard_block":
        hook_utils.hard_block(smart_reason or "destructive smart-rule matched")
    if smart_decision == "allow":
        hook_utils.emit_decision("allow", smart_reason or "smart-rule allow")
    # "deny" via smart-rule is unused now (legacy path).

    # 4. Bypass mode: skip pattern matching. The session has opted out
    #    of permission noise, and smart-rule hard-blocks have already
    #    fired for the genuinely dangerous cases.
    if hook_utils.is_bypass_mode(input_data):
        sys.exit(0)

    # 5. Pattern matching against settings.json allow/deny lists.
    allow_patterns, deny_patterns = load_settings()
    decision, reason = check_pattern_decision(command, allow_patterns, deny_patterns)

    if decision:
        hook_utils.emit_decision(decision, reason or "")

    sys.exit(0)

if __name__ == "__main__":
    main()

hooks/auto-approve-file-ops.py

#!/usr/bin/env python3
"""
PreToolUse hook that auto-approves Read, Edit, and Write operations for allowed paths.

This hook exists because Claude Code's built-in permission system has bugs:
- Issue #15921: VSCode extension ignores Edit/Write permissions
- The permission patterns in settings.json are ignored

This hook enforces path-based permissions for file operations.
"""
from __future__ import annotations

import json
import os
import sys
from fnmatch import fnmatch
from pathlib import Path

# Settings file location
SETTINGS_FILE = Path.home() / ".claude" / "settings.json"

def load_path_patterns() -> tuple[list[str], list[str], list[str]]:
    """Load Read/Edit/Write path patterns from settings."""
    read_patterns: list[str] = []
    edit_patterns: list[str] = []
    write_patterns: list[str] = []

    if SETTINGS_FILE.exists():
        try:
            with open(SETTINGS_FILE, 'r') as f:
                settings = json.load(f)
                permissions = settings.get("permissions", {})

                for rule in permissions.get("allow", []):
                    if rule.startswith("Read(") and rule.endswith(")"):
                        pattern = rule[5:-1]  # Extract path from Read(...)
                        read_patterns.append(pattern)
                    elif rule.startswith("Edit(") and rule.endswith(")"):
                        pattern = rule[5:-1]  # Extract path from Edit(...)
                        edit_patterns.append(pattern)
                    elif rule.startswith("Write(") and rule.endswith(")"):
                        pattern = rule[6:-1]  # Extract path from Write(...)
                        write_patterns.append(pattern)
        except (json.JSONDecodeError, IOError):
            pass

    return read_patterns, edit_patterns, write_patterns

def path_matches_pattern(file_path: str, pattern: str) -> bool:
    """
    Check if a file path matches a glob pattern.

    Patterns use glob syntax:
        - "$HOME/code/**" matches anything under $HOME/code/
        - "/tmp/**" matches anything under /tmp/
    """
    # Normalize paths
    file_path = os.path.normpath(file_path)

    # Handle ** glob pattern (match any depth)
    if "**" in pattern:
        # Convert $HOME/code/** to $HOME/code/ prefix match
        base_path = pattern.replace("**", "").rstrip("/")
        return file_path.startswith(base_path)

    # Standard glob matching
    return fnmatch(file_path, pattern)

def check_file_path(file_path: str, patterns: list[str]) -> tuple[str | None, str | None]:
    """
    Check if a file path matches any allowed pattern.

    Returns: (decision, reason)
        decision: "allow" or None (let default system handle)
    """
    for pattern in patterns:
        if path_matches_pattern(file_path, pattern):
            return ("allow", f"Path matches allowed pattern: {pattern}")

    # No match - let the default permission system handle it
    return (None, None)

def main() -> None:
    try:
        input_data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)  # Non-blocking exit on invalid input

    tool_name = input_data.get("tool_name", "")
    tool_input = input_data.get("tool_input", {})

    # Load patterns from settings
    read_patterns, edit_patterns, write_patterns = load_path_patterns()

    # Handle Read tool
    if tool_name == "Read":
        file_path = tool_input.get("file_path", "")
        if file_path:
            decision, reason = check_file_path(file_path, read_patterns)
            if decision:
                output = {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": decision,
                        "permissionDecisionReason": reason
                    }
                }
                print(json.dumps(output))
        sys.exit(0)

    # Handle Edit tool
    if tool_name == "Edit":
        file_path = tool_input.get("file_path", "")
        if file_path:
            decision, reason = check_file_path(file_path, edit_patterns)
            if decision:
                output = {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": decision,
                        "permissionDecisionReason": reason
                    }
                }
                print(json.dumps(output))
        sys.exit(0)

    # Handle Write tool
    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")
        if file_path:
            decision, reason = check_file_path(file_path, write_patterns)
            if decision:
                output = {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": decision,
                        "permissionDecisionReason": reason
                    }
                }
                print(json.dumps(output))
        sys.exit(0)

    sys.exit(0)

if __name__ == "__main__":
    main()

hooks/ccgm-update-check.py

#!/usr/bin/env python3
"""
UserPromptSubmit hook: daily CCGM health check (upstream + install drift).

On the first prompt of each day:
  1. Check the CCGM remote for new commits and warn if updates are available.
  2. Audit install drift: verify every file the manifest claims is installed
     actually exists on disk (and symlinks resolve).

Both checks print warnings to stderr. A single daily flag file gates both so
subsequent prompts skip the work entirely.

Disable via CCGM_AUTO_UPDATE_CHECK=false in ~/.claude/.ccgm.env.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
from datetime import date
from pathlib import Path

MANIFEST_FILE = Path.home() / ".claude" / ".ccgm-manifest.json"
ENV_FILE = Path.home() / ".claude" / ".ccgm.env"
FLAG_DIR = Path(tempfile.gettempdir())


def is_enabled() -> bool:
    """Check if auto-update check is enabled in .ccgm.env."""
    if not ENV_FILE.exists():
        return False
    try:
        with open(ENV_FILE) as f:
            for line in f:
                line = line.strip()
                if line.startswith("CCGM_AUTO_UPDATE_CHECK="):
                    value = line.split("=", 1)[1].strip().lower()
                    return value in ("true", "1", "yes")
    except (OSError, IOError):
        pass
    return False


def already_checked_today() -> bool:
    """Check if we already ran the update check today."""
    flag_file = FLAG_DIR / f".ccgm-update-check-{date.today().isoformat()}"
    if flag_file.exists():
        return True
    # Create flag file for today, clean up old ones
    for old_flag in FLAG_DIR.glob(".ccgm-update-check-*"):
        try:
            old_flag.unlink()
        except OSError:
            pass
    try:
        flag_file.touch()
    except OSError:
        pass
    return False


def get_ccgm_root() -> str | None:
    """Read the CCGM clone path from the manifest."""
    if not MANIFEST_FILE.exists():
        return None
    try:
        with open(MANIFEST_FILE) as f:
            manifest = json.load(f)
            return manifest.get("ccgmRoot")
    except (json.JSONDecodeError, OSError):
        return None


def check_install_drift() -> list[str]:
    """Return a list of manifest-claimed files that are missing from disk.

    A file is "missing" if its path does not exist. For symlinks, the
    symlink itself must resolve to an existing target. Returns file paths
    (max 10) so output stays readable.
    """
    if not MANIFEST_FILE.exists():
        return []
    try:
        with open(MANIFEST_FILE) as f:
            manifest = json.load(f)
    except (json.JSONDecodeError, OSError):
        return []

    missing: list[str] = []
    for entry in manifest.get("files", []):
        p = Path(entry)
        # Path.exists() follows symlinks, so a dangling symlink shows as missing.
        if not p.exists():
            missing.append(entry)
            if len(missing) >= 10:
                break
    return missing


def check_for_updates(ccgm_root: str) -> int:
    """Fetch remote and count new commits on main."""
    try:
        # Fetch latest (quiet, fast)
        subprocess.run(
            ["git", "fetch", "origin", "--quiet"],
            cwd=ccgm_root,
            capture_output=True,
            timeout=10,
        )
        # Count commits ahead on remote
        result = subprocess.run(
            ["git", "rev-list", "HEAD..origin/main", "--count"],
            cwd=ccgm_root,
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode == 0:
            count = int(result.stdout.strip())
            return count
    except (subprocess.TimeoutExpired, subprocess.SubprocessError, ValueError):
        pass
    return 0


def main() -> None:
    # Read stdin (required by hook contract) but we don't use it
    try:
        json.load(sys.stdin)
    except (json.JSONDecodeError, EOFError):
        pass

    # Quick exit paths
    if not is_enabled():
        return

    if already_checked_today():
        return

    ccgm_root = get_ccgm_root()
    if not ccgm_root or not Path(ccgm_root).is_dir():
        return

    count = check_for_updates(ccgm_root)
    if count > 0:
        s = "s" if count != 1 else ""
        print(
            f"\n  CCGM: {count} update{s} available. "
            f"Run: cd {ccgm_root} && ./update.sh\n",
            file=sys.stderr,
        )

    drift = check_install_drift()
    if drift:
        more = ""
        if len(drift) >= 10:
            more = " (more truncated)"
        print(
            f"\n  CCGM install drift: {len(drift)} manifest-claimed "
            f"file(s) missing from disk{more}:",
            file=sys.stderr,
        )
        for path in drift:
            print(f"    - {path}", file=sys.stderr)
        print(
            "  Resolve by re-running the relevant module install, or "
            "prune the manifest via /ccgm-sync.\n",
            file=sys.stderr,
        )


if __name__ == "__main__":
    main()

hooks/port-check.py

#!/usr/bin/env python3
"""
PreToolUse:Bash hook that intercepts dev server commands and ensures correct
port allocation based on the port registry and .env.clone identity.

Classification (plan.md §5 Epic 1): bypass-suppressible. Warnings are advisory
only — they never block. In bypass-mode sessions the hook short-circuits to
keep dev-server launches quiet. Outside bypass mode it prints to stderr (not
as a permission decision) so the agent can see the warning without blocking.

DETECTS: Commands that launch dev servers (vite, wrangler dev, npm run dev,
pnpm dev, next dev, browser-sync, etc.)

CHECKS:
1. Resolves the correct port from port-registry.json + .env.clone
2. Checks if that port is already in use by another process
3. If a collision is found, warns with the PID and suggests action
4. If the command uses a wrong port (hardcoded or default), warns

OUTPUT: Prints a status message to stderr for the agent to see.
Does NOT block - only warns. The agent should act on the warning.
"""

from __future__ import annotations

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

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_utils  # noqa: E402

REGISTRY_PATH = Path.home() / ".claude" / "port-registry.json"

# Patterns that indicate a dev server is being launched
DEV_SERVER_PATTERNS = [
    r'\bvite\b',
    r'\bwrangler\s+dev\b',
    r'\bnpm\s+run\s+dev\b',
    r'\bpnpm\s+(run\s+)?dev\b',
    r'\bnext\s+dev\b',
    r'\bbrowser-sync\s+start\b',
    r'\bastro\s+dev\b',
    r'\btsx\s+watch\b',
    r'\bnode\s+.*server',
    r'\bconcurrently\b.*\bdev\b',
]

# Patterns to extract --port from command
PORT_FLAG_PATTERN = re.compile(r'--port[=\s]+(\d+)')
PORT_EXPR_PATTERN = re.compile(r'--port\s+\$\(\(([^)]+)\)\)')


def load_registry() -> dict | None:
    """Load port registry."""
    try:
        with open(REGISTRY_PATH) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return None


def get_repo_name() -> str | None:
    """Derive repo name from git remote."""
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=2,
        )
        if result.returncode == 0:
            url = result.stdout.strip()
            name = url.rstrip("/").split("/")[-1]
            if name.endswith(".git"):
                name = name[:-4]
            return name
    except Exception:
        pass
    return None


def get_env_clone() -> dict[str, str]:
    """Read .env.clone from current directory."""
    env_clone: dict[str, str] = {}
    env_path = Path.cwd() / ".env.clone"
    if env_path.exists():
        try:
            with open(env_path) as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith("#") and "=" in line:
                        key, val = line.split("=", 1)
                        env_clone[key.strip()] = val.strip()
        except Exception:
            pass
    return env_clone


def get_port_offset(env_clone: dict[str, str]) -> int:
    """Get port offset from .env.clone."""
    if "PORT_OFFSET" in env_clone:
        try:
            return int(env_clone["PORT_OFFSET"])
        except ValueError:
            pass
    if "CLONE_NUMBER" in env_clone:
        try:
            return int(env_clone["CLONE_NUMBER"])
        except ValueError:
            pass
    # Try deriving from directory name
    dirname = Path.cwd().name
    # Workspace model: extract w and c numbers
    wc_match = re.search(r'w(\d+)-c(\d+)$', dirname)
    if wc_match:
        w, c = int(wc_match.group(1)), int(wc_match.group(2))
        return w * 4 + c  # Assumes 4 clones per workspace
    # Flat clone model: extract trailing number
    num_match = re.search(r'(\d+)$', dirname)
    if num_match:
        return int(num_match.group(1))
    return 0


def check_port_in_use(port: int) -> tuple[str, str] | None:
    """Check if a port is in use. Returns (pid, process_name) or None."""
    try:
        result = subprocess.run(
            ["lsof", "-iTCP:" + str(port), "-sTCP:LISTEN", "-P", "-n", "-t"],
            capture_output=True, text=True, timeout=3,
        )
        if result.returncode == 0 and result.stdout.strip():
            pid = result.stdout.strip().split("\n")[0]
            # Get process name
            ps_result = subprocess.run(
                ["ps", "-p", pid, "-o", "comm="],
                capture_output=True, text=True, timeout=2,
            )
            proc_name = ps_result.stdout.strip() if ps_result.returncode == 0 else "unknown"
            return (pid, proc_name)
    except Exception:
        pass
    return None


def is_dev_server_command(command: str) -> bool:
    """Check if command launches a dev server."""
    for pattern in DEV_SERVER_PATTERNS:
        if re.search(pattern, command, re.IGNORECASE):
            return True
    return False


def extract_port_from_command(command: str) -> int | None:
    """Try to extract a hardcoded port from the command."""
    match = PORT_FLAG_PATTERN.search(command)
    if match:
        return int(match.group(1))
    return None


def determine_service_type(command: str) -> str:
    """Determine if this is a frontend or backend service."""
    cmd_lower = command.lower()
    if "wrangler" in cmd_lower:
        return "backend"
    if "tsx" in cmd_lower and ("server" in cmd_lower or "watch" in cmd_lower):
        return "backend"
    if "api" in cmd_lower:
        return "backend"
    # Default to frontend
    return "frontend"


def main() -> None:
    # The earlier shape passed only `tool_input` here; the modern Claude
    # Code shape passes the full hook envelope (tool_name + tool_input +
    # permission_mode). Read both for compatibility.
    data = hook_utils.read_hook_input()

    # Accept either shape: the modern envelope (with tool_input nested)
    # or the older direct-tool-input payload some tests may still send.
    if "tool_input" in data:
        tool_input = data.get("tool_input", {})
    else:
        tool_input = data

    # Bypass mode: stay completely silent. The user has opted out of
    # permission noise, and port checks are advisory not safety.
    if hook_utils.is_bypass_mode(data):
        return

    command = tool_input.get("command", "")

    # Only care about dev server commands
    if not is_dev_server_command(command):
        return

    registry = load_registry()
    if not registry:
        print("WARNING: Port registry (~/.claude/port-registry.json) not found. "
              "Cannot validate port allocation.", file=sys.stderr)
        return

    repo_name = get_repo_name()
    if not repo_name or repo_name not in registry.get("repos", {}):
        # Not a registered repo, skip
        return

    repo_config = registry["repos"][repo_name]
    env_clone = get_env_clone()
    port_offset = get_port_offset(env_clone)
    service_type = determine_service_type(command)
    base_port = repo_config.get(service_type)

    if base_port is None:
        return

    expected_port = base_port + port_offset
    agent_id = env_clone.get("AGENT_ID", f"offset-{port_offset}")

    # Check if the command specifies a different port
    cmd_port = extract_port_from_command(command)

    messages = []

    if cmd_port is not None and cmd_port != expected_port:
        messages.append(
            f"PORT MISMATCH: Command uses port {cmd_port} but registry assigns "
            f"port {expected_port} for {repo_name} {service_type} ({agent_id}). "
            f"Use --port {expected_port} instead."
        )

    # Check if expected port is already in use
    in_use = check_port_in_use(expected_port)
    if in_use:
        pid, proc_name = in_use
        messages.append(
            f"PORT CONFLICT: Port {expected_port} ({repo_name} {service_type}, "
            f"{agent_id}) is already in use by {proc_name} (PID {pid}). "
            f"Kill it with: kill {pid}"
        )

    # Also check if the command port is in use (if different from expected)
    if cmd_port is not None and cmd_port != expected_port:
        in_use_cmd = check_port_in_use(cmd_port)
        if in_use_cmd:
            pid, proc_name = in_use_cmd
            messages.append(
                f"PORT CONFLICT: Command port {cmd_port} is already in use by "
                f"{proc_name} (PID {pid})."
            )

    # If no explicit port in command and not using expected, suggest it
    if cmd_port is None and port_offset > 0:
        messages.append(
            f"PORT INFO: {repo_name} {service_type} ({agent_id}) should use "
            f"port {expected_port} (base {base_port} + offset {port_offset}). "
            f"Ensure --port {expected_port} is passed or .env.clone is read by the dev config."
        )

    if messages:
        # Warnings go to stderr so the agent sees them in tool output;
        # this hook never blocks via a permission decision (advisory only).
        print("PORT-CHECK: " + " | ".join(messages), file=sys.stderr)
    # If no messages, silently allow


if __name__ == "__main__":
    main()

hooks/agent-tracking-pre.py

#!/usr/bin/env python3
"""
PreToolUse:Bash hook for multi-agent issue tracking.

ADVISORY ONLY - this hook NEVER writes to tracking CSV.
It only emits warnings when an agent is about to work on an
issue that's already claimed by another agent.

Intercepts:
- git checkout -b {N}-* : Warn if issue N is already claimed
"""

from __future__ import annotations

import json
import os
import re
import sys

# Import tracking module
sys.path.insert(0, os.path.expanduser("~/.claude/lib"))


def is_multi_clone_repo() -> bool:
    """Check if current directory has .env.clone (multi-clone repo)."""
    return os.path.isfile(os.path.join(os.getcwd(), ".env.clone"))


def get_repo_name() -> str | None:
    """Get repo name from git remote."""
    import subprocess
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=5,
        )
        if result.returncode == 0:
            name = os.path.basename(result.stdout.strip())
            return name[:-4] if name.endswith(".git") else name
    except Exception:
        pass
    return None


def extract_issue_from_branch(command: str) -> str | None:
    """Extract issue number from git checkout -b {N}-* command."""
    match = re.search(r"git\s+checkout\s+-b\s+(\d+)-", command)
    if match:
        return match.group(1)
    return None


def warn(message: str) -> None:
    """Emit a warning (advisory, does not block)."""
    output = {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecisionReason": f"WARNING: {message}",
        }
    }
    print(json.dumps(output))


def check_live_session_in_cwd() -> dict | None:
    """Check if another live Claude session is running in this working directory."""
    try:
        from agent_sessions import get_active_sessions
        my_cwd = os.getcwd()
        sessions = get_active_sessions(exclude_cwd=my_cwd)
        # Check if any session is in the same directory
        for s in sessions:
            if s.get("cwd") and os.path.realpath(s["cwd"]) == os.path.realpath(my_cwd):
                return s
    except Exception:
        pass
    return None


def main() -> None:
    try:
        data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)

    tool_name = data.get("tool_name", "")
    tool_input = data.get("tool_input", {})

    if tool_name != "Bash":
        sys.exit(0)

    command = tool_input.get("command", "").strip()
    if not command:
        sys.exit(0)

    # Early exit: not a multi-clone repo
    if not is_multi_clone_repo():
        sys.exit(0)

    # Only intercept git checkout -b
    if not re.match(r"git\s+checkout\s+-b\s+", command):
        sys.exit(0)

    # Check for live session in current directory (highest priority warning)
    live_session = check_live_session_in_cwd()
    if live_session:
        pid = live_session.get("pid", "?")
        uptime = live_session.get("uptime", "?")
        branch = live_session.get("branch") or "unknown branch"
        warn(
            f"A live Claude session (PID {pid}, up {uptime}) is already running in this directory "
            f"on branch '{branch}'. Creating a new branch here may conflict with that session's work. "
            f"Consider using a different clone directory."
        )

    issue_num = extract_issue_from_branch(command)
    if not issue_num:
        sys.exit(0)

    repo = get_repo_name()
    if not repo:
        sys.exit(0)

    # Check if issue is already claimed
    try:
        from agent_tracking import check_claim, get_agent_id
        agent, status = check_claim(repo, issue_num)
        my_id = get_agent_id()

        if agent and agent != my_id:
            warn(f"Issue #{issue_num} is already claimed by {agent} (status: {status})")
    except Exception:
        pass  # Never block on tracking errors

    sys.exit(0)


if __name__ == "__main__":
    main()

hooks/agent-tracking-post.py

#!/usr/bin/env python3
"""
PostToolUse:Bash hook for multi-agent issue tracking.

All tracking CSV mutations happen here, AFTER the command succeeds.
This prevents orphaned claims from failed commands.

Intercepts:
- git checkout -b {N}-*     : Register claim for issue N
- git commit -m "#N: ..."   : Update heartbeat (throttled to 30 min)
- gh pr create              : Update status to pr-created
- gh pr merge               : Update status to merged
- gh issue close            : Update status to closed

PostToolUse input schema (verified empirically):
{
    "tool_name": "Bash",
    "tool_input": {"command": "...", "description": "..."},
    "tool_response": {"stdout": "...", "stderr": "...", "interrupted": false},
    "cwd": "/path/to/working/dir",
    ...
}
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys

# Import tracking module
sys.path.insert(0, os.path.expanduser("~/.claude/lib"))


def _find_log_repo() -> str:
    """Find the agent log repo directory."""
    env = os.environ.get("CLAUDE_LOG_REPO")
    if env and os.path.isdir(env):
        return env
    code_dir = os.path.expanduser("~/code")
    if os.path.isdir(code_dir):
        for entry in sorted(os.listdir(code_dir)):
            if entry.endswith("agent-logs") and os.path.isdir(os.path.join(code_dir, entry)):
                return os.path.join(code_dir, entry)
    return os.path.expanduser("~/code/agent-logs")


LOG_REPO_DIR = _find_log_repo()


def is_multi_clone_repo(cwd: str | None = None) -> bool:
    """Check if directory has .env.clone (multi-clone repo)."""
    wd = cwd or os.getcwd()
    return os.path.isfile(os.path.join(wd, ".env.clone"))


def is_log_repo(cwd: str | None = None) -> bool:
    """Check if we're in the log repo (skip heartbeats for log commits)."""
    wd = cwd or os.getcwd()
    try:
        return os.path.realpath(wd).startswith(os.path.realpath(LOG_REPO_DIR))
    except Exception:
        return False


def get_repo_name(cwd: str | None = None) -> str | None:
    """Get repo name from git remote."""
    wd = cwd or os.getcwd()
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=5, cwd=wd,
        )
        if result.returncode == 0:
            name = os.path.basename(result.stdout.strip())
            return name[:-4] if name.endswith(".git") else name
    except Exception:
        pass
    return None


def extract_issue_from_branch_cmd(command: str) -> str | None:
    """Extract issue number from git checkout -b {N}-* command."""
    match = re.search(r"git\s+checkout\s+-b\s+(\d+)-", command)
    if match:
        return match.group(1)
    return None


def extract_issue_from_commit_msg(command: str) -> str | None:
    """Extract issue number from git commit -m '#N: ...' command."""
    match = re.search(r'-m\s+["\']?#(\d+):', command)
    if match:
        return match.group(1)
    return None


def extract_branch_name(command: str) -> str | None:
    """Extract branch name from git checkout -b command."""
    match = re.search(r"git\s+checkout\s+-b\s+(\S+)", command)
    if match:
        return match.group(1)
    return None


def extract_pr_number(stdout: str) -> str | None:
    """Extract PR number from gh pr create output."""
    # gh pr create outputs a URL like https://github.com/user/repo/pull/123
    match = re.search(r"/pull/(\d+)", stdout)
    if match:
        return match.group(1)
    return None


def get_issue_title(issue_num: str, cwd: str | None = None) -> str:
    """Fetch issue title from GitHub (best-effort)."""
    try:
        result = subprocess.run(
            ["gh", "issue", "view", str(issue_num), "--json", "title", "--jq", ".title"],
            capture_output=True, text=True, timeout=10,
            cwd=cwd,
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception:
        pass
    return ""


def get_current_branch(cwd: str | None = None) -> str | None:
    """Get current git branch name."""
    try:
        result = subprocess.run(
            ["git", "branch", "--show-current"],
            capture_output=True, text=True, timeout=5,
            cwd=cwd,
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception:
        pass
    return None


def extract_issue_from_branch_name(branch: str | None) -> str | None:
    """Extract issue number from branch name like 42-fix-auth."""
    if branch:
        match = re.match(r"^(\d+)-", branch)
        if match:
            return match.group(1)
    return None


def main() -> None:
    try:
        data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)

    tool_name = data.get("tool_name", "")
    tool_input = data.get("tool_input", {})
    tool_response = data.get("tool_response", {})
    cwd = data.get("cwd", os.getcwd())

    if tool_name != "Bash":
        sys.exit(0)

    command = tool_input.get("command", "").strip()
    stdout = tool_response.get("stdout", "")
    stderr = tool_response.get("stderr", "")
    interrupted = tool_response.get("interrupted", False)

    if not command or interrupted:
        sys.exit(0)

    # Early exit: not a multi-clone repo
    if not is_multi_clone_repo(cwd):
        sys.exit(0)

    try:
        from agent_tracking import (
            claim_issue, update_status, update_heartbeat,
            check_claim, get_agent_id, get_repo_name as at_get_repo_name,
        )

        agent_id = get_agent_id(cwd)
        repo = get_repo_name(cwd)
        if not repo:
            sys.exit(0)

        # --- git checkout -b {N}-* : Register claim ---
        if re.match(r"git\s+checkout\s+-b\s+\d+-", command):
            issue_num = extract_issue_from_branch_cmd(command)
            branch = extract_branch_name(command)
            if issue_num:
                # Check if we already have this claim (idempotent)
                existing_agent, _ = check_claim(repo, issue_num)
                if existing_agent == agent_id:
                    sys.exit(0)  # Already claimed by us

                title = get_issue_title(issue_num, cwd)
                claim_issue(
                    repo, issue_num,
                    agent_id=agent_id,
                    title=title,
                    branch=branch or "",
                )

        # --- git commit -m "#N: ..." : Heartbeat ---
        elif re.match(r"git\s+commit(\s|$)", command) and not is_log_repo(cwd):
            issue_num = extract_issue_from_commit_msg(command)
            if not issue_num:
                # Try to get issue from branch name
                branch = get_current_branch(cwd)
                issue_num = extract_issue_from_branch_name(branch)

            if issue_num:
                # Also transition from claimed -> in-progress on first commit
                existing_agent, status = check_claim(repo, issue_num)
                if existing_agent == agent_id and status == "claimed":
                    update_status(repo, issue_num, "in-progress", agent_id=agent_id)
                else:
                    update_heartbeat(repo, issue_num, agent_id=agent_id)

        # --- gh pr create : Update to pr-created ---
        elif re.match(r"gh\s+pr\s+create", command):
            branch = get_current_branch(cwd)
            issue_num = extract_issue_from_branch_name(branch)
            pr_num = extract_pr_number(stdout)
            if issue_num:
                update_status(
                    repo, issue_num, "pr-created",
                    agent_id=agent_id, pr=pr_num,
                )

        # --- gh pr merge : Update to merged ---
        elif re.match(r"gh\s+pr\s+merge", command):
            branch = get_current_branch(cwd)
            issue_num = extract_issue_from_branch_name(branch)
            if issue_num:
                update_status(repo, issue_num, "merged", agent_id=agent_id)

        # --- gh issue close : Update to closed ---
        elif re.match(r"gh\s+issue\s+close\s+(\d+)", command):
            match = re.search(r"gh\s+issue\s+close\s+(\d+)", command)
            if match:
                issue_num = match.group(1)
                update_status(repo, issue_num, "closed", agent_id=agent_id)

    except Exception as e:
        # Never block on tracking errors - write warning to stderr
        sys.stderr.write(f"agent-tracking-post: {e}\n")

    sys.exit(0)


if __name__ == "__main__":
    main()

hooks/pretooluse-bash-dispatch.py

#!/usr/bin/env python3
"""Single-process PreToolUse:Bash dispatcher (composition entry point).

This is the BACKWARD-COMPATIBLE composition layer for the PreToolUse:Bash
event. It replaces the six-process legacy chain
(enforce-git-workflow → auto-approve-bash → port-check → agent-tracking-pre
→ check-migration-timestamps → check-careful) with ONE process that runs the
same checks in-process, by priority, through hook_dispatcher.

It is the DEFAULT PreToolUse:Bash handler: settings.partial.json wires this
single entry in place of the six legacy per-process entries. The decisions are
identical because the handlers (lib/pretooluse_bash_checks.py) call the legacy
hooks' own pure functions; the dispatcher only resolves precedence. The six
standalone hook scripts remain installed and individually runnable, so a
deployment can revert by restoring the six PreToolUse:Bash entries.

DECLARATIVE MANIFEST (priority order mirrors the legacy registration order).
Precedence among the decisions they return is governed by hook_dispatcher's
DECISION_RANK: hard_block > deny > allow > ask. The curated destructive set is
short_circuit so it is emitted the instant it fires, nothing can soften it.

  priority  check                       decision kinds       bypass-safe  short
  --------  --------------------------  -------------------  -----------  -----
  10        git_workflow_check          hard_block / adv     yes          no
  20        destructive_check           hard_block           yes          YES
  25        force_branch_delete_check   hard_block           yes          YES
  30        smart_rules_check           hard_block / allow   yes          YES
  40        port_advisory_check         advisory             no           no
  50        agent_tracking_check        advisory             no           no
  60        migration_timestamp_check   hard_block           yes          no
  70        force_push_main_check       hard_block           yes          no
  80        careful_check               ask                  no           no
  90        pattern_check               deny / allow         no           no

bypass-safe=no means the dispatcher skips it in bypass mode, exactly as the
legacy hook exits 0 before its suppressible logic when is_bypass_mode() is
true. bypass-safe=yes means it runs even in bypass mode (the only path that
can produce a bypass-proof exit-2 block).
"""
from __future__ import annotations

import os
import sys

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_dispatcher as hd  # noqa: E402
import pretooluse_bash_checks as checks  # noqa: E402


def build_manifest() -> "hd.Manifest":
    """Construct the declarative PreToolUse:Bash manifest."""
    bash_only = hd.tool_matcher("Bash")
    m = hd.Manifest(event="PreToolUse")
    m.add(hd.Check(10, "git_workflow", bash_only, checks.git_workflow_check,
                   runs_in_bypass=True, short_circuit=False))
    m.add(hd.Check(20, "destructive", bash_only, checks.destructive_check,
                   runs_in_bypass=True, short_circuit=True))
    m.add(hd.Check(25, "force_branch_delete", bash_only, checks.force_branch_delete_check,
                   runs_in_bypass=True, short_circuit=True))
    m.add(hd.Check(30, "smart_rules", bash_only, checks.smart_rules_check,
                   runs_in_bypass=True, short_circuit=True))
    m.add(hd.Check(40, "port_advisory", bash_only, checks.port_advisory_check,
                   runs_in_bypass=False, short_circuit=False))
    m.add(hd.Check(50, "agent_tracking", bash_only, checks.agent_tracking_check,
                   runs_in_bypass=False, short_circuit=False))
    m.add(hd.Check(60, "migration_timestamp", bash_only, checks.migration_timestamp_check,
                   runs_in_bypass=True, short_circuit=False))
    m.add(hd.Check(70, "force_push_main", bash_only, checks.force_push_main_check,
                   runs_in_bypass=True, short_circuit=False))
    m.add(hd.Check(80, "careful", bash_only, checks.careful_check,
                   runs_in_bypass=False, short_circuit=False))
    m.add(hd.Check(90, "pattern", bash_only, checks.pattern_check,
                   runs_in_bypass=False, short_circuit=False))
    return m


def main() -> None:
    hd.dispatch(build_manifest())


if __name__ == "__main__":
    main()

hooks/check-migration-timestamps.py

#!/usr/bin/env python3
"""
PreToolUse:Bash hook to prevent duplicate Supabase migration timestamps.

Classification (plan.md §5 Epic 1): bypass-retained. This is a data-integrity
check, not permission noise — a duplicate timestamp leaves a migration
permanently stuck as "local only" in Supabase, which is hard to recover from.
Block uses `hook_utils.hard_block()` so it survives bypass mode.

BLOCKS git commit (via hard_block, bypass-proof) when migration files have
duplicate numeric prefixes. Duplicate timestamps break `supabase db push`
because the CLI cannot distinguish files that share the same timestamp.

Only runs when:
1. The command is a git commit
2. A supabase/migrations/ directory exists in the working directory
"""

from __future__ import annotations

import os
import re
import sys
from collections import Counter

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_utils  # noqa: E402


def main() -> None:
    data = hook_utils.read_hook_input()

    # Only handle Bash commands
    if data.get("tool_name", "") != "Bash":
        return

    command = data.get("tool_input", {}).get("command", "")

    # Only check git commit commands
    if not re.search(r"\bgit\s+commit\b", command):
        return

    # Check if we're in a git repo with supabase/migrations/
    migrations_dir = None
    for candidate in ["supabase/migrations", "supabase/migrations/"]:
        if os.path.isdir(candidate):
            migrations_dir = candidate
            break

    if not migrations_dir:
        return

    # Check ALL migration files for duplicate timestamps
    try:
        all_files = sorted(os.listdir(migrations_dir))
    except OSError:
        return

    sql_files = [f for f in all_files if f.endswith(".sql")]
    timestamps = []
    for f in sql_files:
        match = re.match(r"^(\d+)", f)
        if match:
            timestamps.append(match.group(1))

    # Find duplicates
    counts = Counter(timestamps)
    duplicates = {ts: count for ts, count in counts.items() if count > 1}

    if not duplicates:
        return

    # Build error message
    dup_details = []
    for ts in sorted(duplicates.keys()):
        files = [f for f in sql_files if f.startswith(ts)]
        dup_details.append(f"  Timestamp {ts} used by {duplicates[ts]} files:")
        for f in files:
            dup_details.append(f"    - {f}")

    error_msg = (
        "BLOCKED: Duplicate Supabase migration timestamps detected.\n"
        "Duplicate timestamps break `supabase db push` - the CLI can't distinguish files\n"
        "that share the same numeric prefix. Rename one file to a unique timestamp.\n\n"
        + "\n".join(dup_details)
        + "\n\nFix: rename one file in each group to increment the timestamp by 1 "
        "(e.g., 20260325900000 -> 20260325900001)."
    )

    hook_utils.hard_block(error_msg)


if __name__ == "__main__":
    main()

hooks/orphan-process-check.py

#!/usr/bin/env python3
"""
Check for orphaned test worker processes (vitest, jest) at session start.

Orphaned workers occur when a Claude Code session exits mid-test-run. The forked
worker processes get re-parented to PID 1 (launchd) and run indefinitely, consuming
RAM and CPU.

This hook runs during startup and warns if orphans are found.
"""

import json
import os
import subprocess
import sys


def find_orphaned_test_workers():
    """Find node processes with PPID 1 that look like test workers."""
    try:
        result = subprocess.run(
            ["ps", "-eo", "pid,ppid,rss,command"],
            capture_output=True, text=True, timeout=5
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return []

    orphans = []
    for line in result.stdout.strip().split("\n")[1:]:  # skip header
        parts = line.split(None, 3)
        if len(parts) < 4:
            continue

        pid, ppid, rss_kb, command = parts[0], parts[1], parts[2], parts[3]

        # Only orphaned processes (PPID 1)
        if ppid != "1":
            continue

        # Only node processes that look like test workers
        test_patterns = ["vitest", "jest-worker", "jest_worker", "test-worker"]
        if not any(p in command.lower() for p in test_patterns):
            continue

        try:
            orphans.append({
                "pid": int(pid),
                "rss_mb": int(rss_kb) / 1024,
                "command": command[:80]
            })
        except ValueError:
            continue

    return orphans


def main():
    orphans = find_orphaned_test_workers()

    if not orphans:
        sys.exit(0)

    total_mb = sum(o["rss_mb"] for o in orphans)
    pids = [str(o["pid"]) for o in orphans]

    # Output warning via hook result
    msg = (
        f"WARNING: {len(orphans)} orphaned test worker(s) found "
        f"({total_mb:.0f} MB RAM). "
        f"PIDs: {', '.join(pids[:10])}. "
        f"Run: kill {' '.join(pids[:10])}"
    )

    print(json.dumps({
        "decision": "approve",
        "reason": msg
    }))


if __name__ == "__main__":
    main()

hooks/check-careful.py

#!/usr/bin/env python3
"""
PreToolUse hook that pauses on destructive Bash commands.

Classification (plan.md §5 Epic 1): bypass-suppressible. In bypass-mode
sessions (bypassPermissions / dontAsk / auto) the routine "ask" decisions
short-circuit to exit 0 — the user already opted out of permission noise.

The exception is force-push-to-`main`: that case is migrated to
`hook_utils.hard_block()` so it survives bypass mode. Pushing over a
protected branch's history is the kind of destructive op `ALLOW_MAIN_COMMIT`
exists to gate, not a permission-noise prompt.

Inspects Bash commands for destructive patterns:
  - rm -rf (with smart whitelist for build artifacts)
  - SQL DROP / TRUNCATE
  - git push --force (and --force-with-lease) — escalates to hard_block on `main`
  - git reset --hard (history destroying)
  - git checkout . (dirty discard)
  - kubectl delete
  - docker rm -f / docker system prune

Build-artifact directories are whitelisted so `rm -rf node_modules` does not
trigger a prompt. Whitelist: node_modules, dist, .next, build, __pycache__,
.cache, .turbo, coverage.

Ported to Python from gstack `careful/bin/check-careful.sh` to match CCGM's
hook style (JSON stdin/stdout, consistent with auto-approve-bash.py).
"""
from __future__ import annotations

import os
import re
import sys

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_utils  # noqa: E402

# Build-artifact directory names that are safe to `rm -rf` without asking.
# If every path on an rm -rf line matches one of these (exactly or as a
# trailing component), we let it pass without warning.
BUILD_ARTIFACT_WHITELIST = {
    "node_modules",
    "dist",
    ".next",
    "build",
    "__pycache__",
    ".cache",
    ".turbo",
    "coverage",
}


def _is_whitelisted_rm_rf(command: str) -> bool:
    """
    Return True if every target of `rm -rf` in `command` is a build artifact.

    Walks tokens after `rm` (in any order of flags) and verifies each
    non-flag argument is a recognised build-artifact directory name.
    Conservative: any unknown target disables the whitelist.
    """
    # Extract the argument list after `rm`. We only check the first rm
    # occurrence; chained commands are handled separately by the caller.
    m = re.search(r"\brm\s+([^\|&;]+)", command)
    if not m:
        return False

    tokens = m.group(1).strip().split()
    targets: list[str] = []
    for tok in tokens:
        if tok.startswith("-"):
            # Flag (e.g. -rf, -r, -f, --force, -R)
            continue
        targets.append(tok)

    if not targets:
        return False

    for target in targets:
        # Strip trailing slash, quotes
        t = target.strip().strip("'\"").rstrip("/")
        # Extract the last path component (e.g. "apps/web/dist" -> "dist")
        last = t.rsplit("/", 1)[-1]
        if last not in BUILD_ARTIFACT_WHITELIST:
            return False

    return True


def check_careful(command: str) -> tuple[bool, str]:
    """
    Determine whether the command is destructive and should prompt the user.

    Returns (is_destructive, warning_reason).
    """
    # Normalize whitespace for easier matching
    cmd = command

    # rm -rf  (any order of -r/-R/-f/--recursive/--force)
    # Matches: rm -rf, rm -fr, rm -r -f, rm --recursive --force, rm -Rf
    rm_rf_patterns = [
        r"\brm\s+[^\|&;]*-[rRf]*r[rRf]*f",  # -rf, -fr, -Rf, -fR (r and f together)
        r"\brm\s+[^\|&;]*-[rRf]*f[rRf]*r",  # -fr ordering
        r"\brm\s+[^\|&;]*--recursive[^\|&;]*--force",
        r"\brm\s+[^\|&;]*--force[^\|&;]*--recursive",
        r"\brm\s+[^\|&;]*-r\b[^\|&;]*-f\b",
        r"\brm\s+[^\|&;]*-f\b[^\|&;]*-r\b",
    ]
    for pattern in rm_rf_patterns:
        if re.search(pattern, cmd):
            if _is_whitelisted_rm_rf(cmd):
                return (False, "")
            return (True, "Destructive: `rm -rf` deletes recursively. Confirm targets are correct.")

    # SQL DROP statements (DROP TABLE, DROP DATABASE, DROP SCHEMA, etc.)
    if re.search(r"\bDROP\s+(TABLE|DATABASE|SCHEMA|INDEX|VIEW|FUNCTION|TRIGGER|ROLE|USER)\b", cmd, re.IGNORECASE):
        return (True, "Destructive SQL: DROP removes objects permanently. Confirm target and environment.")

    # SQL TRUNCATE
    if re.search(r"\bTRUNCATE\s+(TABLE\s+)?\w+", cmd, re.IGNORECASE):
        return (True, "Destructive SQL: TRUNCATE empties a table. Confirm target and environment.")

    # git push --force / -f. The safer --force-with-lease only rewrites the
    # remote when our local ref still matches it, which is the recommended
    # flow in git-workflow.md when rebasing a feature branch onto main.
    # Don't prompt on --force-with-lease.
    if re.search(r"\bgit\s+push\s+[^\|&;]*(--force\b(?!-with-lease)|-f\b)", cmd):
        return (True, "History-rewriting: `git push --force` overwrites remote commits. Confirm the branch is yours.")

    # git reset --hard (any target). Even `origin/main` can discard uncommitted work.
    if re.search(r"\bgit\s+(-C\s+\S+\s+)?reset\s+--hard\b", cmd):
        return (True, "History-destroying: `git reset --hard` discards local changes and commits. Confirm no work is lost.")

    # git checkout . (discard working tree changes)
    if re.search(r"\bgit\s+checkout\s+\.(\s|$)", cmd):
        return (True, "Destructive: `git checkout .` discards all uncommitted changes. Confirm before proceeding.")

    # git restore . / git restore --staged .
    if re.search(r"\bgit\s+restore\s+(--staged\s+)?\.(\s|$)", cmd):
        return (True, "Destructive: `git restore .` discards uncommitted changes. Confirm before proceeding.")

    # git clean -f (removes untracked files)
    if re.search(r"\bgit\s+clean\s+[^\|&;]*-[fdx]*f", cmd):
        return (True, "Destructive: `git clean -f` removes untracked files. Confirm before proceeding.")

    # kubectl delete
    if re.search(r"\bkubectl\s+delete\b", cmd):
        return (True, "Destructive: `kubectl delete` removes cluster resources. Confirm target and namespace.")

    # docker rm -f / docker rmi -f
    if re.search(r"\bdocker\s+rmi?\s+[^\|&;]*-f", cmd):
        return (True, "Destructive: `docker rm/rmi -f` force-removes containers/images. Confirm before proceeding.")

    # docker system prune / docker volume prune
    if re.search(r"\bdocker\s+(system|volume|image|container|network)\s+prune\b", cmd):
        return (True, "Destructive: `docker ... prune` removes unused resources. Confirm before proceeding.")

    return (False, "")


def _is_force_push_to_main(command: str) -> bool:
    """Specifically: `git push --force[-f|--force-with-lease] ... main` (and refs that resolve to it).

    Distinct from `_is_force_push_to_branch` because main is the one branch
    a bypass-mode session should NOT be able to overwrite without an explicit
    `ALLOW_MAIN_COMMIT=1` escape hatch.
    """
    if not re.search(r"\bgit\s+push\b", command):
        return False
    if not re.search(r"(--force\b|--force-with-lease\b|\s-f\b)", command):
        return False
    # Match: git push --force origin main, git push -f origin main,
    # git push --force-with-lease origin main, git push origin +main,
    # git push origin HEAD:main, git push origin main:main, etc.
    if re.search(r"\b(origin\s+\+?main\b|main:main\b|HEAD:main\b|\s\+main\b)", command):
        return True
    # `git push --force` with no remote/refspec defaults to current branch.
    # Catch that when the current branch is main via env var, but the
    # hook can't reliably know the branch — leave that case to
    # enforce-git-workflow.py which DOES read the branch.
    return False


def main() -> None:
    data = hook_utils.read_hook_input()

    tool_name = data.get("tool_name", "")
    tool_input = data.get("tool_input", {})

    if tool_name != "Bash":
        sys.exit(0)

    command = tool_input.get("command", "")
    if not command:
        sys.exit(0)

    # Bypass-proof: force-push to main is a hard block regardless of mode.
    # `ALLOW_MAIN_COMMIT=1` opens the explicit emergency channel.
    if _is_force_push_to_main(command) and os.environ.get("ALLOW_MAIN_COMMIT") != "1":
        hook_utils.hard_block(
            "BLOCKED: force-pushing to `main` overwrites shared history. "
            "If this is truly intended (recovering from a bad merge, etc.), "
            "re-run with `ALLOW_MAIN_COMMIT=1` set."
        )

    is_destructive, reason = check_careful(command)
    if not is_destructive:
        sys.exit(0)

    # In bypass mode the user has explicitly opted out of permission noise.
    # Routine `ask` decisions are suppressed; the hard-block path above
    # still fires for genuinely dangerous ops.
    if hook_utils.is_bypass_mode(data):
        sys.exit(0)

    hook_utils.emit_decision("ask", reason)


if __name__ == "__main__":
    main()

hooks/check-freeze.py

#!/usr/bin/env python3
"""
PreToolUse hook that scope-locks Edit/Write to a frozen directory.

When a freeze is active (state file `~/.claude/freeze-dir.txt` exists and
contains a directory path), Edit and Write operations outside that directory
are denied. Use `/freeze <dir>` to set the scope and `/unfreeze` to clear it.

Paths are normalised (symlinks resolved, `..` collapsed) before the
containment check so trivial escape attempts (`../foo`, symlinked parent) are
caught POSIX-portably.

Ported to Python from gstack `freeze/bin/check-freeze.sh` to match CCGM's
hook style (JSON stdin/stdout, consistent with auto-approve-file-ops.py).
"""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path

FREEZE_FILE = Path.home() / ".claude" / "freeze-dir.txt"


def read_freeze_dir() -> str | None:
    """Return the current freeze directory (absolute, resolved) or None."""
    if not FREEZE_FILE.exists():
        return None
    try:
        raw = FREEZE_FILE.read_text().strip()
    except OSError:
        return None
    if not raw:
        return None
    # Expand ~ and environment variables, then resolve.
    expanded = os.path.expandvars(os.path.expanduser(raw))
    try:
        return str(Path(expanded).resolve())
    except OSError:
        return None


def resolve_path(path: str) -> str | None:
    """
    Resolve `path` to an absolute canonical form.

    Uses Path.resolve(strict=False) so new files (Write) that do not yet
    exist on disk still get their parent hierarchy resolved and `..` collapsed.
    """
    if not path:
        return None
    expanded = os.path.expandvars(os.path.expanduser(path))
    try:
        return str(Path(expanded).resolve())
    except OSError:
        return None


def is_within(target: str, parent: str) -> bool:
    """Return True if `target` is the same path as or nested under `parent`."""
    # Ensure trailing slash semantics so "/a/b" does not match "/a/bcd".
    parent_norm = parent.rstrip(os.sep)
    target_norm = target.rstrip(os.sep)
    if target_norm == parent_norm:
        return True
    return target_norm.startswith(parent_norm + os.sep)


def main() -> None:
    try:
        input_data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)

    tool_name = input_data.get("tool_name", "")
    if tool_name not in ("Edit", "Write"):
        sys.exit(0)

    freeze_dir = read_freeze_dir()
    if not freeze_dir:
        # No freeze active. Fall through to default permission handling.
        sys.exit(0)

    tool_input = input_data.get("tool_input", {})
    file_path = tool_input.get("file_path", "")
    resolved = resolve_path(file_path)
    if resolved is None:
        # Could not resolve; fall through rather than block on malformed input.
        sys.exit(0)

    if is_within(resolved, freeze_dir):
        # In scope - let other hooks / default system decide.
        sys.exit(0)

    reason = (
        f"Freeze active: writes are scoped to {freeze_dir}. "
        f"{resolved} is outside the frozen directory. "
        f"Run `/unfreeze` to clear the scope, or stay within the frozen path."
    )
    output = {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": reason,
        }
    }
    print(json.dumps(output))
    sys.exit(0)


if __name__ == "__main__":
    main()

hooks/session-start-enforce.py

#!/usr/bin/env python3
"""
SessionStart hook that injects a rule-enforcement meta-instruction.

Adapted from obra/superpowers `hooks/session-start`. The idea: CCGM installs
discipline rules (TDD, systematic-debugging, verification, confusion-protocol,
etc.) to `~/.claude/rules/` where Claude Code auto-loads them, but there is no
*meta-instruction* that forces the agent to route through those rules under
pressure. This hook injects a short reminder at fresh session start so the
agent treats the Iron Laws as real gates, not background reading.

Experimental: OFF by default. Opt in by setting CCGM_RULE_ENFORCEMENT=true in
`~/.claude/.ccgm.env`. Fires only on source == "startup" so it does not fire
on resume or compaction.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

ENV_FILE = Path.home() / ".claude" / ".ccgm.env"

# The meta-instruction injected into the session. Kept short on purpose:
# long context at session start competes with the user's first prompt for
# attention. The goal is to bias routing toward loaded rules, not to restate
# them.
META_INSTRUCTION = """\
<ccgm-rule-enforcement>
Before your first response in this session, scan the loaded rules in
~/.claude/rules/ and in any CLAUDE.md files for Iron Laws (all-caps "NO X
WITHOUT Y" declarations). For any task with a plausible match, route through
the relevant rule before acting:

- Writing or modifying code -> test-driven-development (failing test first).
- Fixing a bug or unexpected behavior -> systematic-debugging (root cause before fix).
- Claiming a task is done, tests pass, or a build works -> verification (fresh evidence).
- Dispatching work to subagents -> subagent-patterns (spec + status protocol).
- Unclear requirements, contradictions, or missing context -> confusion-protocol (stop and ask).

"Violating the letter of a rule is violating the spirit." Do not negotiate
Iron Laws under time pressure, user pressure, or sunk-cost pressure. If a rule
seems to block the task, surface the conflict instead of routing around it.
</ccgm-rule-enforcement>
"""


def is_enabled() -> bool:
    """Check if rule-enforcement injection is enabled in .ccgm.env."""
    if not ENV_FILE.exists():
        return False
    try:
        with open(ENV_FILE) as f:
            for line in f:
                line = line.strip()
                if line.startswith("CCGM_RULE_ENFORCEMENT="):
                    value = line.split("=", 1)[1].strip().lower()
                    return value in ("true", "1", "yes")
    except (OSError, IOError):
        pass
    return False


def main() -> None:
    # Read stdin (required by hook contract)
    try:
        hook_input = json.load(sys.stdin)
    except (json.JSONDecodeError, EOFError):
        hook_input = {}

    # Only fire on fresh sessions, not resume or compact. Resume already has
    # the rules in-context from the prior session; compact has its own
    # (different) context-preservation mechanism.
    source = hook_input.get("source", "")
    if source != "startup":
        return

    if not is_enabled():
        return

    # Print to stdout - Claude Code injects this as additionalContext for the
    # SessionStart event.
    sys.stdout.write(META_INSTRUCTION)


if __name__ == "__main__":
    main()

hooks/sync-ccgm-canonical.py

#!/usr/bin/env python3
"""
PostToolUse:Bash hook — sync the canonical CCGM clone after a CCGM PR merges.

Why: ~/.claude/ symlinks point at one canonical CCGM checkout. When PRs merge
in workspace clones, that canonical checkout drifts unless something pulls it.
This hook removes the manual sync step.

Triggers when:
- The Bash command invokes `gh pr merge ...` in any segment (it is usually
  `cd <repo>` first, or chained with && / piped to tail -- not the first token)
- The cwd's git remote points at a repo named "ccgm" (any owner)
- The canonical clone exists at $CCGM_CANONICAL_DIR (default ~/code/ccgm)

Behavior:
- Runs `git fetch origin main && git pull --ff-only origin main` in the
  canonical clone
- Logs success/failure to stderr
- Never blocks on errors (always exit 0)
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys


CANONICAL_DIR_ENV = "CCGM_CANONICAL_DIR"
DEFAULT_CANONICAL_DIR = os.path.expanduser("~/code/ccgm")
CCGM_REPO_NAME = "ccgm"


def get_origin_url(cwd: str) -> str | None:
    try:
        result = subprocess.run(
            ["git", "-C", cwd, "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=3, check=False,
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except (subprocess.SubprocessError, OSError):
        pass
    return None


def command_triggers_merge(command: str) -> bool:
    """True if any segment of `command` invokes `gh pr merge`.

    `gh pr merge` is almost never the first token of the command string -- it is
    typically `cd <repo>` first (often on its own line) or chained with && / piped
    to tail. A start-anchored match therefore misses real merges and leaves the
    canonical clone stale (#728). Split on shell separators (newline ; | &) and
    check each segment. A false positive only costs one harmless, idempotent
    ff-only pull, so erring toward matching is safe.
    """
    for segment in re.split(r"[\n;|&]+", command):
        if re.match(r"\s*gh\s+pr\s+merge(\s|$)", segment):
            return True
    return False


def is_ccgm_repo(cwd: str) -> bool:
    url = get_origin_url(cwd)
    if not url:
        return False
    # Extract repo name from URL (last path segment, strip .git)
    repo_name = re.sub(r"\.git$", "", url.rstrip("/").rsplit("/", 1)[-1])
    return repo_name == CCGM_REPO_NAME


def sync_canonical(canonical_dir: str) -> tuple[bool, str]:
    """Pull origin/main into canonical_dir. Returns (success, message)."""
    if not os.path.isdir(os.path.join(canonical_dir, ".git")):
        return False, f"canonical dir not a git repo: {canonical_dir}"

    try:
        fetch = subprocess.run(
            ["git", "-C", canonical_dir, "fetch", "origin", "main"],
            capture_output=True, text=True, timeout=30, check=False,
        )
        if fetch.returncode != 0:
            return False, f"fetch failed: {fetch.stderr.strip()}"

        pull = subprocess.run(
            ["git", "-C", canonical_dir, "pull", "--ff-only", "origin", "main"],
            capture_output=True, text=True, timeout=30, check=False,
        )
        if pull.returncode != 0:
            return False, f"pull failed (not fast-forward?): {pull.stderr.strip()}"

        return True, pull.stdout.strip().splitlines()[-1] if pull.stdout.strip() else "up to date"
    except subprocess.TimeoutExpired:
        return False, "timeout"
    except (subprocess.SubprocessError, OSError) as e:
        return False, str(e)


def main() -> None:
    try:
        payload = json.load(sys.stdin)
    except (json.JSONDecodeError, ValueError):
        sys.exit(0)

    if payload.get("tool_name") != "Bash":
        sys.exit(0)

    command = (payload.get("tool_input") or {}).get("command", "")
    if not command_triggers_merge(command):
        sys.exit(0)

    cwd = payload.get("cwd") or os.getcwd()
    if not is_ccgm_repo(cwd):
        sys.exit(0)

    canonical_dir = os.environ.get(CANONICAL_DIR_ENV, DEFAULT_CANONICAL_DIR)
    if not os.path.isdir(canonical_dir):
        sys.stderr.write(
            f"sync-ccgm-canonical: skipped — {canonical_dir} does not exist "
            f"(set {CANONICAL_DIR_ENV} or create the dir)\n"
        )
        sys.exit(0)

    if os.path.realpath(cwd) == os.path.realpath(canonical_dir):
        sys.exit(0)

    ok, msg = sync_canonical(canonical_dir)
    prefix = "sync-ccgm-canonical"
    if ok:
        sys.stderr.write(f"{prefix}: {canonical_dir} → {msg}\n")
    else:
        sys.stderr.write(f"{prefix}: FAILED — {msg}\n")

    sys.exit(0)


if __name__ == "__main__":
    main()

lib/agent_tracking.py

#!/usr/bin/env python3
"""
Multi-agent issue tracking system.

Provides structured issue lifecycle tracking via a single CSV file per repo
in the agent log repository. Replaces the label-based claiming system.

Storage: ~/code/{log-repo}/{repo}/tracking.csv
Format: CSV with fields: issue,agent,status,branch,pr,epic,title,claimed_at,updated_at
Concurrency: Standard git flow (commit, pull --rebase, push). Different-row
edits auto-resolve via rebase.

Usage as CLI:
    python agent_tracking.py claim <repo> <issue> [--title "..."] [--epic N]
    python agent_tracking.py check <repo> <issue>
    python agent_tracking.py update <repo> <issue> --status <status> [--pr N]
    python agent_tracking.py release <repo> <issue>
    python agent_tracking.py list [--repo <repo>] [--status <status>] [--agent <id>]
    python agent_tracking.py gc [--days N]
    python agent_tracking.py import <repo>
    python agent_tracking.py init <repo>

Usage as import:
    from agent_tracking import claim_issue, check_claim, update_status
"""

from __future__ import annotations

import argparse
import csv
import io
import os
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

CSV_FIELDS = [
    "issue", "agent", "status", "branch", "pr",
    "epic", "title", "claimed_at", "updated_at",
]

ACTIVE_STATUSES = {"claimed", "in-progress", "pr-created", "blocked"}
TERMINAL_STATUSES = {"merged", "closed", "released"}
ALL_STATUSES = ACTIVE_STATUSES | TERMINAL_STATUSES


def _find_log_repo() -> str:
    """Find the agent log repo directory."""
    env = os.environ.get("CLAUDE_LOG_REPO")
    if env and os.path.isdir(env):
        return env
    code_dir = os.path.expanduser("~/code")
    if os.path.isdir(code_dir):
        for entry in sorted(os.listdir(code_dir)):
            if entry.endswith("agent-logs") and os.path.isdir(os.path.join(code_dir, entry)):
                return os.path.join(code_dir, entry)
    return os.path.expanduser("~/code/agent-logs")


LOG_REPO_DIR = _find_log_repo()

# ---------------------------------------------------------------------------
# Agent identity helpers
# ---------------------------------------------------------------------------

def get_agent_id(working_dir: str | None = None) -> str:
    """Derive agent ID from the working directory or .env.clone."""
    wd = working_dir or os.getcwd()

    # Try .env.clone first
    env_clone = os.path.join(wd, ".env.clone")
    if os.path.isfile(env_clone):
        with open(env_clone) as f:
            for line in f:
                if line.startswith("AGENT_ID="):
                    return line.strip().split("=", 1)[1]

    # Workspace model: directory name ends with w{N}-c{M}
    import re
    basename = os.path.basename(wd)
    wc = re.search(r"w(\d+)-c(\d+)$", basename)
    if wc:
        return f"agent-w{wc.group(1)}-c{wc.group(2)}"

    # Flat clone model: directory name ends with -{N}
    num = re.search(r"-(\d+)$", basename)
    if num:
        return f"agent-{num.group(1)}"

    return "agent-0"


def get_repo_name(working_dir: str | None = None) -> str | None:
    """Derive repo name from git remote origin URL."""
    wd = working_dir or os.getcwd()
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=5, cwd=wd,
        )
        if result.returncode == 0:
            url = result.stdout.strip()
            name = os.path.basename(url)
            if name.endswith(".git"):
                name = name[:-4]
            return name
    except Exception:
        pass
    return None


def is_multi_clone_repo(working_dir: str | None = None) -> bool:
    """Check if the current directory is a multi-clone repo (has .env.clone)."""
    wd = working_dir or os.getcwd()
    return os.path.isfile(os.path.join(wd, ".env.clone"))


# ---------------------------------------------------------------------------
# CSV helpers
# ---------------------------------------------------------------------------

def get_tracking_path(repo: str) -> str:
    """Return the path to tracking.csv for a repo."""
    return os.path.join(LOG_REPO_DIR, repo, "tracking.csv")


def read_tracking(repo: str) -> list[dict[str, str]]:
    """Read tracking.csv and return a list of dicts."""
    path = get_tracking_path(repo)
    if not os.path.isfile(path):
        return []
    rows = []
    with open(path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            rows.append(row)
    return rows


def write_tracking(repo: str, rows: list[dict[str, str]]) -> None:
    """Write a list of dicts to tracking.csv."""
    path = get_tracking_path(repo)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS, quoting=csv.QUOTE_MINIMAL)
        writer.writeheader()
        for row in rows:
            writer.writerow(row)


def now_iso() -> str:
    """Return current local time as ISO 8601 string (minute precision)."""
    return datetime.now().strftime("%Y-%m-%dT%H:%M")


# ---------------------------------------------------------------------------
# Git operations on the log repo
# ---------------------------------------------------------------------------

def commit_and_push(agent_id: str, message: str) -> bool:
    """Commit tracking changes to the log repo and push.

    Uses standard git flow: add, commit, pull --rebase, push.
    Returns True on success, False on failure (non-blocking).
    """
    try:
        cmds = [
            ["git", "add", "-A"],
            ["git", "commit", "-m", f"{agent_id}: {message}"],
            ["git", "pull", "--rebase"],
            ["git", "push"],
        ]
        for cmd in cmds:
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=30,
                cwd=LOG_REPO_DIR,
            )
            # git commit returns 1 if nothing to commit - that's ok
            if result.returncode != 0 and cmd[1] != "commit":
                sys.stderr.write(
                    f"WARNING: tracking git {cmd[1]} failed: {result.stderr.strip()}\n"
                )
                return False
        return True
    except Exception as e:
        sys.stderr.write(f"WARNING: tracking commit/push failed: {e}\n")
        return False


# ---------------------------------------------------------------------------
# Core operations
# ---------------------------------------------------------------------------

def claim_issue(repo: str, issue: int | str, agent_id: str | None = None, title: str = "", epic: int | str = "", branch: str = "") -> tuple[bool, str]:
    """Claim an issue. Returns (success: bool, message: str)."""
    agent_id = agent_id or get_agent_id()
    issue = str(issue)

    rows = read_tracking(repo)

    # Check for existing active claim on this issue
    for row in rows:
        if row["issue"] == issue and row["status"] in ACTIVE_STATUSES:
            if row["agent"] == agent_id:
                return False, f"You already have issue #{issue} claimed"
            return False, f"Issue #{issue} is already claimed by {row['agent']}"

    # Add claim
    rows.append({
        "issue": issue,
        "agent": agent_id,
        "status": "claimed",
        "branch": branch,
        "pr": "",
        "epic": str(epic) if epic else "",
        "title": title,
        "claimed_at": now_iso(),
        "updated_at": now_iso(),
    })

    write_tracking(repo, rows)
    commit_and_push(agent_id, f"claim #{issue}")
    return True, f"Claimed issue #{issue}"


def update_status(repo: str, issue: int | str, status: str, agent_id: str | None = None, pr: int | str | None = None, branch: str | None = None) -> tuple[bool, str]:
    """Update the status of a claimed issue. Returns (success, message)."""
    agent_id = agent_id or get_agent_id()
    issue = str(issue)

    if status not in ALL_STATUSES:
        return False, f"Invalid status: {status}. Valid: {', '.join(sorted(ALL_STATUSES))}"

    rows = read_tracking(repo)
    updated = False

    for row in rows:
        if row["issue"] == issue and row["agent"] == agent_id and row["status"] in ACTIVE_STATUSES:
            row["status"] = status
            row["updated_at"] = now_iso()
            if pr is not None:
                row["pr"] = str(pr)
            if branch is not None:
                row["branch"] = branch
            updated = True
            break

    if not updated:
        return False, f"No active claim found for issue #{issue} by {agent_id}"

    write_tracking(repo, rows)
    commit_and_push(agent_id, f"update #{issue} -> {status}")
    return True, f"Updated issue #{issue} to {status}"


def update_heartbeat(repo: str, issue: int | str, agent_id: str | None = None, throttle_minutes: int = 30) -> tuple[bool, str]:
    """Update the heartbeat (updated_at) for an issue, throttled.

    Only updates if the current updated_at is older than throttle_minutes.
    Returns (updated: bool, message: str).
    """
    agent_id = agent_id or get_agent_id()
    issue = str(issue)

    rows = read_tracking(repo)
    threshold = datetime.now() - timedelta(minutes=throttle_minutes)

    for row in rows:
        if row["issue"] == issue and row["agent"] == agent_id and row["status"] in ACTIVE_STATUSES:
            try:
                last_updated = datetime.strptime(row["updated_at"], "%Y-%m-%dT%H:%M")
                if last_updated > threshold:
                    return False, "Heartbeat throttled (too recent)"
            except ValueError:
                pass  # Can't parse, just update

            row["updated_at"] = now_iso()
            write_tracking(repo, rows)
            # Don't commit/push for heartbeats - they'll be included in the next
            # regular log repo commit to avoid excessive pushes
            return True, f"Heartbeat updated for #{issue}"

    return False, f"No active claim found for issue #{issue}"


def release_issue(repo: str, issue: int | str, agent_id: str | None = None) -> tuple[bool, str]:
    """Release a claimed issue. Returns (success, message)."""
    return update_status(repo, issue, "released", agent_id)


def check_claim(repo: str, issue: int | str) -> tuple[str | None, str | None]:
    """Check if an issue is claimed. Returns (agent_id, status) or (None, None)."""
    issue = str(issue)
    rows = read_tracking(repo)
    for row in rows:
        if row["issue"] == issue and row["status"] in ACTIVE_STATUSES:
            return row["agent"], row["status"]
    return None, None


def list_claims(repo: str | None = None, status: str | None = None, agent: str | None = None) -> list[dict[str, str]]:
    """List claims, optionally filtered. Returns list of dicts."""
    results: list[dict[str, str]] = []

    if repo:
        repos = [repo]
    else:
        # Find all repos with tracking.csv
        repos = []
        if os.path.isdir(LOG_REPO_DIR):
            for entry in sorted(os.listdir(LOG_REPO_DIR)):
                tracking = os.path.join(LOG_REPO_DIR, entry, "tracking.csv")
                if os.path.isfile(tracking):
                    repos.append(entry)

    for r in repos:
        rows = read_tracking(r)
        for row in rows:
            if status and row["status"] != status:
                continue
            if agent and row["agent"] != agent:
                continue
            row["_repo"] = r
            results.append(row)

    return results


def gc_stale(repo: str | None = None, days: int = 1) -> list[dict]:
    """Find stale claims (active status + old updated_at). Returns list of stale rows."""
    threshold = datetime.now() - timedelta(days=days)
    stale = []

    claims = list_claims(repo=repo)
    for row in claims:
        if row["status"] not in ACTIVE_STATUSES:
            continue
        try:
            updated = datetime.strptime(row["updated_at"], "%Y-%m-%dT%H:%M")
            if updated < threshold:
                row["_stale_hours"] = int((datetime.now() - updated).total_seconds() / 3600)
                stale.append(row)
        except ValueError:
            # Can't parse date - consider it stale
            row["_stale_hours"] = "unknown"
            stale.append(row)

    return stale


def import_from_labels(repo: str, agent_id: str | None = None) -> tuple[int, str]:
    """Import existing GitHub issues with agent-* labels into tracking.csv.

    Scans for open issues with in-progress or agent-* labels and creates
    tracking entries for them.
    """
    agent_id = agent_id or get_agent_id()
    imported = 0

    try:
        # Get all open issues with labels
        result = subprocess.run(
            ["gh", "issue", "list", "--state", "open", "--limit", "100",
             "--json", "number,title,labels"],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return 0, f"Failed to list issues: {result.stderr.strip()}"

        import json
        issues = json.loads(result.stdout)
        rows = read_tracking(repo)
        existing_issues = {row["issue"] for row in rows if row["status"] in ACTIVE_STATUSES}

        for issue in issues:
            issue_num = str(issue["number"])
            if issue_num in existing_issues:
                continue

            labels = [l["name"] for l in issue.get("labels", [])]
            # Find agent label
            agent_label = None
            for label in labels:
                if label.startswith("agent-"):
                    agent_label = label
                    break

            if agent_label or "in-progress" in labels:
                status = "in-progress" if "in-progress" in labels else "claimed"
                rows.append({
                    "issue": issue_num,
                    "agent": agent_label or agent_id,
                    "status": status,
                    "branch": "",
                    "pr": "",
                    "epic": "",
                    "title": issue["title"],
                    "claimed_at": now_iso(),
                    "updated_at": now_iso(),
                })
                imported += 1

        if imported > 0:
            write_tracking(repo, rows)
            commit_and_push(agent_id, f"import {imported} issues from labels")

        return imported, f"Imported {imported} issues"

    except Exception as e:
        return 0, f"Import failed: {e}"


def init_tracking(repo: str) -> tuple[bool, str]:
    """Create tracking.csv with header row for a repo."""
    path = get_tracking_path(repo)
    if os.path.isfile(path):
        return False, f"tracking.csv already exists for {repo}"

    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS, quoting=csv.QUOTE_MINIMAL)
        writer.writeheader()

    return True, f"Created tracking.csv for {repo}"


# ---------------------------------------------------------------------------
# Display helpers
# ---------------------------------------------------------------------------

def format_claims_table(claims: list[dict[str, str]], show_repo: bool = False) -> str:
    """Format claims as a readable table string."""
    if not claims:
        return "  (no claims)"

    lines = []
    if show_repo:
        header = f"  {'Repo':<20} {'Issue':>6} {'Agent':<16} {'Status':<14} {'Branch':<30} {'PR':>4} {'Updated'}"
        lines.append(header)
        lines.append("  " + "-" * (len(header) - 2))
        for c in claims:
            repo_name = c.get("_repo", "")
            lines.append(
                f"  {repo_name:<20} #{c['issue']:>5} {c['agent']:<16} {c['status']:<14} "
                f"{c.get('branch', ''):<30} {c.get('pr', ''):>4} {c.get('updated_at', '')}"
            )
    else:
        header = f"  {'Issue':>6} {'Agent':<16} {'Status':<14} {'Branch':<30} {'PR':>4} {'Updated'}"
        lines.append(header)
        lines.append("  " + "-" * (len(header) - 2))
        for c in claims:
            lines.append(
                f"  #{c['issue']:>5} {c['agent']:<16} {c['status']:<14} "
                f"{c.get('branch', ''):<30} {c.get('pr', ''):>4} {c.get('updated_at', '')}"
            )

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Multi-agent issue tracking",
        prog="agent-tracking",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    # claim
    p = sub.add_parser("claim", help="Claim an issue")
    p.add_argument("repo", help="Repo name (e.g., darkly-suite)")
    p.add_argument("issue", type=int, help="Issue number")
    p.add_argument("--title", default="", help="Issue title")
    p.add_argument("--epic", type=int, default=None, help="Parent epic number")
    p.add_argument("--branch", default="", help="Branch name")

    # update
    p = sub.add_parser("update", help="Update issue status")
    p.add_argument("repo", help="Repo name")
    p.add_argument("issue", type=int, help="Issue number")
    p.add_argument("--status", required=True, help="New status")
    p.add_argument("--pr", type=int, default=None, help="PR number")
    p.add_argument("--branch", default=None, help="Branch name")

    # release
    p = sub.add_parser("release", help="Release an issue")
    p.add_argument("repo", help="Repo name")
    p.add_argument("issue", type=int, help="Issue number")

    # check
    p = sub.add_parser("check", help="Check if an issue is claimed")
    p.add_argument("repo", help="Repo name")
    p.add_argument("issue", type=int, help="Issue number")

    # list
    p = sub.add_parser("list", help="List claims")
    p.add_argument("--repo", default=None, help="Filter by repo")
    p.add_argument("--status", default=None, help="Filter by status")
    p.add_argument("--agent", default=None, help="Filter by agent")

    # gc
    p = sub.add_parser("gc", help="Find stale claims")
    p.add_argument("--repo", default=None, help="Filter by repo")
    p.add_argument("--days", type=int, default=1, help="Stale threshold in days")

    # import
    p = sub.add_parser("import", help="Import issues from GitHub labels")
    p.add_argument("repo", help="Repo name")

    # init
    p = sub.add_parser("init", help="Initialize tracking.csv for a repo")
    p.add_argument("repo", help="Repo name")

    args = parser.parse_args()

    # Defensive: empty-string --repo is almost always a caller bug (e.g., REPO_NAME
    # failed to derive in a workspace coordinator dir). Without this guard, list/gc
    # would silently fall through to scanning ALL repos and dump unrelated noise.
    if getattr(args, "repo", None) == "":
        print("Error: --repo cannot be empty (omit --repo to scan all repos intentionally)",
              file=sys.stderr)
        sys.exit(2)

    if args.command == "claim":
        ok, msg = claim_issue(args.repo, args.issue, title=args.title,
                              epic=args.epic or "", branch=args.branch)
        print(msg)
        sys.exit(0 if ok else 1)

    elif args.command == "update":
        ok, msg = update_status(args.repo, args.issue, args.status,
                                pr=args.pr, branch=args.branch)
        print(msg)
        sys.exit(0 if ok else 1)

    elif args.command == "release":
        ok, msg = release_issue(args.repo, args.issue)
        print(msg)
        sys.exit(0 if ok else 1)

    elif args.command == "check":
        agent, status = check_claim(args.repo, args.issue)
        if agent:
            print(f"Issue #{args.issue} is claimed by {agent} (status: {status})")
        else:
            print(f"Issue #{args.issue} is unclaimed")

    elif args.command == "list":
        claims = list_claims(repo=args.repo, status=args.status, agent=args.agent)
        show_repo = args.repo is None
        print(format_claims_table(claims, show_repo=show_repo))

    elif args.command == "gc":
        stale = gc_stale(repo=args.repo, days=args.days)
        if stale:
            print(f"Found {len(stale)} stale claims:")
            for s in stale:
                repo_name = s.get("_repo", "")
                hours = s.get("_stale_hours", "?")
                print(f"  WARNING: {s['agent']} has stale claim on #{s['issue']} "
                      f"in {repo_name} (status: {s['status']}, stale for {hours}h)")
        else:
            print("No stale claims found")

    elif args.command == "import":
        count, msg = import_from_labels(args.repo)
        print(msg)
        sys.exit(0 if count >= 0 else 1)

    elif args.command == "init":
        ok, msg = init_tracking(args.repo)
        print(msg)
        sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()

lib/agent_sessions.py

#!/usr/bin/env python3
"""
Discover active Claude Code CLI sessions on this machine.

Uses ps + lsof + git to find all running claude CLI sessions with their
working directory, repo, and branch context. No files, no daemons - just
reads live OS state. Process exit = session gone, no stale data possible.

Usage as library:
    from agent_sessions import get_active_sessions
    sessions = get_active_sessions()
    # [{pid, tty, uptime, cwd, repo, branch, agent_id}, ...]

Usage as CLI:
    python3 agent_sessions.py          # JSON output
    python3 agent_sessions.py --text   # Human-readable table
    python3 agent_sessions.py --repo habitpro-ai  # Filter by repo
"""

from __future__ import annotations

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


# ---------------------------------------------------------------------------
# Core discovery
# ---------------------------------------------------------------------------

def _run(cmd: list[str], cwd: str | None = None, timeout: int = 5) -> str:
    """Run a command, return stdout or empty string on failure."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=cwd,
        )
        return result.stdout.strip() if result.returncode == 0 else ""
    except Exception:
        return ""


def _get_claude_pids() -> list[tuple[int, str, str]]:
    """
    Return list of (pid, tty, etime) for running claude CLI processes.

    Filters out Claude Desktop (Electron app) and helper processes.
    Matches only the bare 'claude' command or 'claude /startup' etc.
    """
    ps_out = _run(["ps", "-eo", "pid,tty,etime,command"])
    pids: list[tuple[int, str, str]] = []
    for line in ps_out.splitlines():
        # Match lines where the command is exactly 'claude' (with optional args)
        # Exclude: Claude.app, Claude Helper, grep, python running this script
        m = re.search(r"^\s*(\d+)\s+(\S+)\s+(\S+)\s+claude(\s|$)", line)
        if not m:
            continue
        full_line = line.strip()
        # Exclude Electron app and helpers
        if any(skip in full_line for skip in ["Claude.app", "Claude Helper", "Claude.framework"]):
            continue
        pid, tty, etime = m.group(1), m.group(2), m.group(3)
        pids.append((int(pid), tty, etime))
    return pids


def _get_cwd(pid: int) -> str | None:
    """Get the current working directory of a process via lsof."""
    out = _run(["lsof", "-a", "-p", str(pid), "-d", "cwd", "-Fn"], timeout=5)
    for line in out.splitlines():
        if line.startswith("n"):
            path = line[1:]
            if os.path.isdir(path):
                return path
    return None


def _get_git_context(cwd: str) -> tuple[str | None, str | None]:
    """Return (repo_name, branch) for a directory, or (None, None)."""
    branch = _run(["git", "-C", cwd, "branch", "--show-current"])
    if not branch:
        return None, None
    remote_url = _run(["git", "-C", cwd, "remote", "get-url", "origin"])
    repo_name = os.path.basename(remote_url) if remote_url else None
    if repo_name and repo_name.endswith(".git"):
        repo_name = repo_name[:-4]
    repo = repo_name
    return repo, branch


def _get_tmux_pane_pids() -> dict[int, str]:
    """Return {pane_pid: session_name} for all tmux panes, or {} if tmux unavailable."""
    out = _run(["tmux", "list-panes", "-a", "-F", "#{pane_pid} #{session_name}"])
    panes: dict[int, str] = {}
    for line in out.splitlines():
        parts = line.split(None, 1)
        if len(parts) == 2 and parts[0].isdigit():
            panes[int(parts[0])] = parts[1]
    return panes


def _get_tmux_attached_sessions() -> set[str]:
    """Return set of tmux session names with at least one attached client."""
    out = _run(["tmux", "list-clients", "-F", "#{session_name}"])
    return {line.strip() for line in out.splitlines() if line.strip()}


def _get_ppid_map() -> dict[int, int]:
    """Return {pid: ppid} for all running processes."""
    out = _run(["ps", "-eo", "pid,ppid"])
    ppid_map: dict[int, int] = {}
    for line in out.splitlines()[1:]:
        parts = line.split()
        if len(parts) >= 2 and parts[0].isdigit() and parts[1].isdigit():
            ppid_map[int(parts[0])] = int(parts[1])
    return ppid_map


def _find_tmux_session(
    pid: int,
    ppid_map: dict[int, int],
    pane_pids: dict[int, str],
) -> str | None:
    """Walk parent chain from pid; return tmux session name if an ancestor is a tmux pane."""
    cur = pid
    for _ in range(32):
        if cur in pane_pids:
            return pane_pids[cur]
        parent = ppid_map.get(cur)
        if parent is None or parent <= 1 or parent == cur:
            return None
        cur = parent
    return None


def _get_agent_id(cwd: str) -> str | None:
    """
    Derive agent identity from .env.clone or directory name pattern.

    Workspace model: habitpro-ai-w1-c2 -> agent-w1-c2
    Flat clone model: habitpro-ai-3 -> agent-3
    """
    env_clone = os.path.join(cwd, ".env.clone")
    if os.path.isfile(env_clone):
        try:
            with open(env_clone) as f:
                for line in f:
                    if line.startswith("AGENT_ID="):
                        return line.strip().split("=", 1)[1]
        except Exception:
            pass

    dirname = os.path.basename(cwd)
    # Workspace model pattern: *-wN-cN
    m = re.search(r"w\d+-c\d+$", dirname)
    if m:
        return f"agent-{m.group()}"
    # Flat clone model pattern: *-N (trailing number)
    m = re.search(r"-(\d+)$", dirname)
    if m:
        return f"agent-{m.group(1)}"
    return None


def get_active_sessions(repo_filter: str | None = None, exclude_cwd: str | None = None) -> list[dict]:
    """
    Return list of active Claude Code CLI sessions on this machine.

    Args:
        repo_filter: If set, only return sessions for this repo name.
        exclude_cwd: If set, exclude the session at this working directory
                     (used to exclude the current session from sibling lists).

    Returns:
        List of dicts:
        {
            "pid":      int,    # Process ID
            "tty":      str,    # Terminal (e.g. ttys003, ?? for background)
            "uptime":   str,    # Elapsed time (e.g. "01-02:30:45", "15:30")
            "cwd":      str,    # Working directory
            "repo":       str,    # Git repo name (or None)
            "branch":     str,    # Current git branch (or None)
            "agent_id":   str,    # Derived agent ID (or None)
            "tmux_state": str,    # "attached", "detached", or None (not in tmux)
        }
    """
    sessions: list[dict] = []
    my_cwd = os.path.realpath(exclude_cwd) if exclude_cwd else None

    pane_pids = _get_tmux_pane_pids()
    attached_sessions = _get_tmux_attached_sessions() if pane_pids else set()
    ppid_map = _get_ppid_map() if pane_pids else {}

    for pid, tty, etime in _get_claude_pids():
        cwd = _get_cwd(pid)
        if not cwd:
            continue

        # Normalize for comparison
        real_cwd = os.path.realpath(cwd)
        if my_cwd and real_cwd == my_cwd:
            continue  # Skip current session

        repo, branch = _get_git_context(cwd)

        if repo_filter and repo != repo_filter:
            continue

        tmux_session = _find_tmux_session(pid, ppid_map, pane_pids) if pane_pids else None
        if tmux_session is None:
            tmux_state: str | None = None
        elif tmux_session in attached_sessions:
            tmux_state = "attached"
        else:
            tmux_state = "detached"

        sessions.append({
            "pid":        pid,
            "tty":        tty,
            "uptime":     etime,
            "cwd":        cwd,
            "repo":       repo,
            "branch":     branch,
            "agent_id":   _get_agent_id(cwd),
            "tmux_state": tmux_state,
        })

    return sessions


# ---------------------------------------------------------------------------
# Formatting helpers
# ---------------------------------------------------------------------------

def format_sessions_text(sessions: list[dict], header: bool = True) -> str:
    """Format sessions as a human-readable table for dashboard display."""
    if not sessions:
        return "  (none)"

    lines = []
    for s in sessions:
        pid_str = str(s["pid"])
        repo = s["repo"] or "(no repo)"
        branch = s["branch"] or "(no branch)"
        agent = s["agent_id"] or "(unknown)"
        uptime = s["uptime"]
        tty = s["tty"]
        cwd = s["cwd"]

        tmux_state = s.get("tmux_state")
        suffix = f"  [tmux:{tmux_state}]" if tmux_state else ""

        if s["repo"]:
            lines.append(
                f"  PID {pid_str:6} | {repo:25} | branch: {branch:30} | "
                f"up: {uptime:12} | {tty}{suffix}"
            )
        else:
            lines.append(
                f"  PID {pid_str:6} | (no repo) {cwd:40} | up: {uptime:12} | {tty}{suffix}"
            )
    return "\n".join(lines)


def sessions_by_repo(sessions: list[dict]) -> dict[str, list[dict]]:
    """Group sessions by repo name. Returns {repo: [sessions]}."""
    grouped = {}
    for s in sessions:
        key = s["repo"] or "(no repo)"
        grouped.setdefault(key, []).append(s)
    return grouped


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(
        description="List active Claude Code CLI sessions on this machine."
    )
    parser.add_argument("--text", action="store_true", help="Human-readable output")
    parser.add_argument("--repo", help="Filter by repo name")
    parser.add_argument("--exclude-cwd", help="Exclude session at this CWD")
    args = parser.parse_args()

    sessions = get_active_sessions(
        repo_filter=args.repo,
        exclude_cwd=args.exclude_cwd,
    )

    if args.text:
        print(format_sessions_text(sessions))
    else:
        print(json.dumps(sessions, indent=2))


if __name__ == "__main__":
    main()

lib/hook_utils.py

"""Shared utilities for CCGM Claude Code hooks.

Locked API (referenced from plan.md §5 Epic 1 and §3 architecture):

    read_hook_input() -> dict
    permission_mode(data: dict) -> str
    is_bypass_mode(data: dict) -> bool
    emit_decision(decision: str, reason: str) -> None
    hard_block(reason: str) -> NoReturn
    redact_secrets(text: str) -> str
    file_locked_append(path: str, data: str) -> None
    load_repo_config(cwd: str) -> dict

The module is installed at ~/.claude/lib/hook_utils.py by CCGM's installer.
Hooks import it via `sys.path.insert(0, os.path.expanduser("~/.claude/lib"))`
followed by `import hook_utils`.
"""
from __future__ import annotations

import fcntl
import json
import os
import re
import sys
from typing import NoReturn

__all__ = [
    "read_hook_input",
    "permission_mode",
    "is_bypass_mode",
    "emit_decision",
    "hard_block",
    "redact_secrets",
    "file_locked_append",
    "load_repo_config",
    "BYPASS_MODES",
    "SECRET_PATTERNS",
]


BYPASS_MODES = frozenset({"bypassPermissions", "dontAsk", "auto"})


def read_hook_input() -> dict:
    """Read and parse JSON from stdin. Returns {} on parse failure.

    Hooks that need stdin in a strict mode should check the return value
    themselves; this helper never raises so that a malformed payload
    cannot wedge an enforcement hook into an error state.
    """
    try:
        return json.load(sys.stdin)
    except (json.JSONDecodeError, ValueError, EOFError):
        return {}


def permission_mode(data: dict) -> str:
    """Return the permission_mode field from hook stdin, or 'default'.

    Claude Code passes one of: 'default', 'acceptEdits', 'plan',
    'bypassPermissions', 'dontAsk', 'auto'. Older clients may omit it.
    """
    mode = data.get("permission_mode")
    if isinstance(mode, str) and mode:
        return mode
    return "default"


def is_bypass_mode(data: dict) -> bool:
    """True iff the session is in a bypass-permission mode.

    Treats bypassPermissions, dontAsk, and auto as bypass. Conservative
    default: any unknown / missing mode is treated as NON-bypass so a
    suppressible safety check still fires when in doubt.
    """
    return permission_mode(data) in BYPASS_MODES


def emit_decision(decision: str, reason: str) -> None:
    """Emit a PreToolUse JSON decision and exit 0.

    `decision` must be one of 'allow', 'deny', 'ask'. The hook process
    terminates after this call — callers should not assume control
    returns.
    """
    payload = {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": decision,
            "permissionDecisionReason": reason,
        }
    }
    json.dump(payload, sys.stdout)
    sys.stdout.write("\n")
    sys.stdout.flush()
    sys.exit(0)


def hard_block(reason: str) -> NoReturn:
    """Bypass-proof hard block. Writes reason to stderr and exits 2.

    GitHub issue #39344: `permissionDecision: 'ask'` from any PreToolUse
    hook silently overrides declarative deny rules. The only mechanism
    that survives bypass mode is `exit 2`, which Claude Code treats as a
    hard block regardless of permission_mode.

    Use this for data-integrity invariants and protected-branch
    enforcement that MUST hold even in bypass sessions.
    """
    sys.stderr.write(reason.rstrip() + "\n")
    sys.stderr.flush()
    sys.exit(2)


# 17 secret patterns. Order matters only insofar as longer / more specific
# patterns come first so they win when a substring would match multiple.
# Each entry is (name, compiled regex). The replacement is always
# "[REDACTED:{name}]" so downstream readers can identify the kind without
# re-scanning. Patterns target prefix shapes published by each vendor.
_SECRET_PATTERN_SOURCES: list[tuple[str, str]] = [
    # Anthropic API keys (legacy and api03-prefixed).
    ("anthropic", r"sk-ant-(?:api03-)?[A-Za-z0-9_\-]{32,}"),
    # Stripe live/test secret keys.
    ("stripe_live", r"sk_live_[A-Za-z0-9]{16,}"),
    ("stripe_test", r"sk_test_[A-Za-z0-9]{16,}"),
    # GitHub token families (PAT, OAuth, user-to-server, server-to-server, refresh).
    ("github_pat", r"ghp_[A-Za-z0-9]{30,}"),
    ("github_oauth", r"gho_[A-Za-z0-9]{30,}"),
    ("github_u2s", r"ghu_[A-Za-z0-9]{30,}"),
    ("github_s2s", r"ghs_[A-Za-z0-9]{30,}"),
    ("github_refresh", r"ghr_[A-Za-z0-9]{30,}"),
    # AWS access key.
    ("aws_access_key", r"AKIA[0-9A-Z]{16}"),
    # Google API key.
    ("google_api", r"AIza[0-9A-Za-z_\-]{35}"),
    # Slack tokens.
    ("slack", r"xox[abprs]-[0-9A-Za-z\-]{10,}"),
    # Resend.
    ("resend", r"re_[A-Za-z0-9]{8,}_[A-Za-z0-9]{16,}"),
    # Supabase service-role and publishable keys (modern prefixed shape).
    ("supabase", r"sb_(?:secret|publishable)_[A-Za-z0-9]{20,}"),
    # OpenAI (generic sk- form, after Anthropic and Stripe have eaten theirs).
    ("openai", r"sk-(?!ant-)(?!live_)(?!test_)[A-Za-z0-9]{32,}"),
    # Authorization: Bearer ... header (covers most generic bearer tokens).
    ("authorization_bearer", r"(?i)authorization\s*:\s*bearer\s+[A-Za-z0-9._\-]+"),
    # env-var style KV assignments naming common secret keys.
    (
        "env_var_kv",
        r"(?i)\b(?:api[_-]?key|api[_-]?secret|access[_-]?token|auth[_-]?token|"
        r"client[_-]?secret|secret[_-]?key|password|passwd)\s*[=:]\s*['\"]?[A-Za-z0-9_\-/+=.]{12,}",
    ),
    # CLI password flag (--password / -p with a value following).
    ("password_flag", r"(?<!\S)(?:--password|--passwd|-p)[=\s]+\S{6,}"),
]

# Compile once at import.
SECRET_PATTERNS: list[tuple[str, "re.Pattern[str]"]] = [
    (name, re.compile(pat)) for name, pat in _SECRET_PATTERN_SOURCES
]


def redact_secrets(text: str) -> str:
    """Replace any secret-shaped token in `text` with [REDACTED:{kind}].

    Applied to event log entries BEFORE truncation so the truncation
    point can never lop a redaction marker in half. Conservative: a
    pattern with broad-ish shape (env_var_kv) is fine to over-fire —
    false positives in logs cost much less than leaked secrets.
    """
    if not text:
        return text
    out = text
    for name, regex in SECRET_PATTERNS:
        out = regex.sub(f"[REDACTED:{name}]", out)
    return out


def file_locked_append(path: str, data: str) -> None:
    """Append `data` (with trailing newline) to `path`, fcntl-locked.

    Cross-clone-safe: multiple Claude Code agents in different repo
    clones may race on the same `~/.claude/autoheal/events/*.jsonl`
    path. fcntl.flock(LOCK_EX) on the open file descriptor serializes
    appends so the JSONL never truncates mid-record.

    POSIX-portable. macOS + Linux supported. Creates parent dir if
    missing. Never raises on lock contention — blocks until acquired.
    """
    payload = data if data.endswith("\n") else data + "\n"
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    # Open with O_APPEND so writes always land at end-of-file even
    # under contention. Combined with flock(LOCK_EX), this is the
    # standard POSIX recipe for atomic concurrent append.
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)
        try:
            os.write(fd, payload.encode("utf-8"))
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
    finally:
        os.close(fd)


def load_repo_config(cwd: str | None = None) -> dict:
    """Walk up from cwd looking for `.autoheal/config.json`. Return merged config.

    The walk stops at the first repo root that contains the file or at
    the filesystem root. Returns {} when no config is found.

    Schema (validated lightly here; full schema in
    `modules/autoheal/lib/repo-config-schema.json` for Epic 12):

        {
          "additional_allow_patterns": [str, ...],
          "calibration_days": int,
          "thresholds": {"confidence_min": int, "occurrence_min": int},
          "kind_filters": [str, ...]
        }

    Unknown top-level keys are preserved (forward-compat) but only the
    documented fields are consumed by downstream code.
    """
    here = os.path.abspath(cwd or os.getcwd())
    seen: set[str] = set()
    while here and here not in seen:
        seen.add(here)
        candidate = os.path.join(here, ".autoheal", "config.json")
        if os.path.isfile(candidate):
            try:
                with open(candidate, "r", encoding="utf-8") as fh:
                    cfg = json.load(fh)
                if isinstance(cfg, dict):
                    return cfg
            except (OSError, json.JSONDecodeError):
                # Treat a malformed repo config as if it did not exist;
                # a hook is not the right place to fail loudly on user
                # JSON typos.
                return {}
        parent = os.path.dirname(here)
        if parent == here:
            break
        here = parent
    return {}

lib/hook_dispatcher.py

"""In-process hook composition dispatcher with declarative precedence.

Problem this solves
-------------------
Today every Claude Code event fans out into one OS process per registered
hook script (see modules/*/settings.partial.json). A single PreToolUse:Bash
tool call spawns six python interpreters in sequence, each re-importing
hook_utils, each re-parsing stdin. There is also no explicit, testable
ordering contract: precedence is an emergent property of the array order in
settings.json across several modules that do not know about each other.

This module provides a *backward-compatible* alternative: a single dispatcher
per event that runs a DECLARATIVE manifest of checks in-process, by priority,
with an explicit precedence resolution that exactly mirrors the behavior the
separate-process chain produces today:

    hard_block (exit 2)  >  deny  >  allow  >  ask  >  (advisory / pass)

Precedence guarantees (the safety contract — preserved bit-for-bit):

  * The FIRST hard_block wins and is emitted via hook_utils.hard_block()
    (exit 2). This is the only signal Claude Code honors regardless of
    permission_mode (GitHub issue #39344), so it survives bypass mode.
  * A `deny` beats any `allow`. If any check denies and none hard-block,
    the dispatcher emits deny.
  * An `allow` is emitted only if some check allows and nothing denies or
    hard-blocks.
  * `ask` is the weakest decision; it is emitted only if nothing above it
    fired.
  * Advisory output (stderr warnings) NEVER affects the decision and is
    flushed for every matching check whose result carries it, regardless
    of which decision ultimately wins.
  * Bypass-mode short-circuit: a check may declare `runs_in_bypass=False`.
    Such checks are skipped when hook_utils.is_bypass_mode() is true. Checks
    that must survive bypass (the curated destructive set, data-integrity
    hard blocks, protected-branch enforcement) declare `runs_in_bypass=True`
    and run ABOVE the short-circuit — exactly the current arrangement in
    auto-approve-bash.py and check-careful.py.

The dispatcher does NOT replace hook_utils; it composes its primitives.
redact_secrets-before-truncation, fcntl file locking, and bypass detection
all continue to live in hook_utils and are reused unchanged.

Coexistence
-----------
This is additive. Installing the dispatcher does not remove or rewrite any
existing hook. A module may migrate onto the dispatcher by registering a
single entry hook that calls dispatch(); modules not yet migrated keep their
own settings.partial.json entries and run on the legacy per-process path.
The two paths produce identical decisions because the dispatcher's precedence
rules are derived from the legacy chain's observed behavior.
"""
from __future__ import annotations

import os
import sys
from dataclasses import dataclass, field
from typing import Callable, Optional

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_utils  # noqa: E402

__all__ = [
    "ALLOW",
    "DENY",
    "ASK",
    "HARD_BLOCK",
    "PASS",
    "DECISION_RANK",
    "Result",
    "Check",
    "Manifest",
    "dispatch",
    "tool_matcher",
]

# ─── Decision vocabulary ─────────────────────────────────────────────
# Ranked weakest → strongest. A larger rank wins precedence.
PASS = "pass"            # check did not fire; no opinion
ASK = "ask"              # request a permission prompt (suppressible in bypass)
ALLOW = "allow"          # auto-approve
DENY = "deny"            # block (overridable by hard_block, beats allow)
HARD_BLOCK = "hard_block"  # bypass-proof block via exit 2

DECISION_RANK = {
    PASS: 0,
    ASK: 1,
    ALLOW: 2,
    DENY: 3,
    HARD_BLOCK: 4,
}


@dataclass
class Result:
    """What a single check decided.

    `decision` is one of the decision-vocabulary constants. `reason` is the
    human-readable explanation surfaced to Claude / the user. `advisory` is
    optional stderr text that is ALWAYS printed (it never participates in the
    decision) — this is how port-check / agent-tracking-style warnings are
    represented without giving them blocking power.
    """

    decision: str = PASS
    reason: str = ""
    advisory: str = ""

    def __post_init__(self) -> None:
        if self.decision not in DECISION_RANK:
            raise ValueError(f"unknown decision: {self.decision!r}")


# A handler receives the full hook envelope (the parsed stdin dict) and
# returns a Result. It must be pure with respect to the decision (no
# sys.exit, no printing of permission JSON) — the dispatcher owns emission.
Handler = Callable[[dict], "Result"]


def tool_matcher(*tool_names: str) -> Callable[[dict], bool]:
    """Build a matcher that fires only for the named tools.

    Mirrors settings.json `matcher` semantics: a check registered for "Bash"
    only runs when tool_name == "Bash". An empty matcher (no names) matches
    every tool, mirroring a settings.json block with no `matcher` key.
    """
    wanted = frozenset(tool_names)

    def _match(data: dict) -> bool:
        if not wanted:
            return True
        return data.get("tool_name", "") in wanted

    return _match


@dataclass
class Check:
    """A declarative entry in a dispatcher manifest.

    Fields:
      priority      Lower runs first. Hard-block-capable checks that must beat
                    a downstream allow are given a smaller number so they are
                    evaluated before it. (Precedence resolution does NOT rely
                    on priority alone — DECISION_RANK is authoritative — but
                    priority fixes a deterministic, documented run order and
                    decides ties between same-rank decisions: first wins.)
      name          Stable identifier (used in tests + audit).
      matches       Predicate over the hook envelope; True ⇒ run the handler.
      handler       Callable returning a Result.
      runs_in_bypass  If False, the check is SKIPPED in bypass mode. Safety
                    checks that must survive bypass set this True.
      short_circuit If True, ANY decisive (non-PASS) result from this check is
                    emitted immediately without consulting later checks. This
                    reproduces a legacy hook that calls emit_decision()/
                    hard_block() and exits the instant it fires — e.g. the
                    curated destructive set (hard_block) and the
                    git-reset-to-remote smart-rule (allow), both of which the
                    standalone auto-approve-bash.py emits before any later
                    check runs. A PASS never short-circuits regardless of this
                    flag (a check with no opinion yields to the rest).
    """

    priority: int
    name: str
    matches: Callable[[dict], bool]
    handler: Handler
    runs_in_bypass: bool = True
    short_circuit: bool = False


@dataclass
class Manifest:
    """An ordered set of checks for one event (optionally one tool)."""

    event: str
    checks: list = field(default_factory=list)

    def add(self, check: "Check") -> "Manifest":
        self.checks.append(check)
        return self

    def ordered(self) -> list:
        # Stable sort by priority; registration order breaks ties.
        return sorted(self.checks, key=lambda c: c.priority)


def _emit_and_exit(decision: str, reason: str) -> None:
    """Emit a decision through the hook_utils primitives and terminate.

    hard_block → exit 2 (bypass-proof). deny/allow/ask → JSON on stdout,
    exit 0. The dispatcher routes EVERY winning decision through these same
    primitives so behavior is identical to a standalone hook calling them.
    """
    if decision == HARD_BLOCK:
        hook_utils.hard_block(reason or "hard-blocked")
    if decision in (ALLOW, DENY, ASK):
        hook_utils.emit_decision(decision, reason)
    # PASS: no output, exit 0.
    sys.exit(0)


def dispatch(manifest: "Manifest", data: Optional[dict] = None) -> None:
    """Run the manifest's checks in-process and emit the winning decision.

    This function does not return on a decisive outcome — it calls
    hook_utils.hard_block() (exit 2) or hook_utils.emit_decision() (exit 0),
    exactly as a standalone hook would. On PASS it exits 0.

    Execution model (mirrors the legacy per-process chain):

      1. Read the envelope once (data is read here if not supplied) — replaces
         N separate stdin reads with one.
      2. Determine bypass mode once.
      3. For each check in priority order:
           a. Skip if its matcher does not match the tool.
           b. Skip if bypass mode is active AND the check is not bypass-safe.
           c. Run the handler.
           d. Always flush any advisory text to stderr (never blocks).
           e. If the result is a hard_block AND the check is short_circuit,
              emit it immediately (nothing can override the curated
              destructive set).
           f. Otherwise track the strongest decision seen so far (ties keep
              the earlier check, matching first-wins chain order).
      4. After all checks, emit the strongest tracked decision.

    Precedence is governed by DECISION_RANK, so the final outcome is
    independent of how many checks fired: one hard_block beats any number of
    allows; one deny beats any number of allows; etc.
    """
    if data is None:
        data = hook_utils.read_hook_input()

    bypass = hook_utils.is_bypass_mode(data)

    best_decision = PASS
    best_reason = ""

    for check in manifest.ordered():
        if not check.matches(data):
            continue
        if bypass and not check.runs_in_bypass:
            continue

        result = check.handler(data)

        # Advisory output is decision-independent and always surfaced.
        if result.advisory:
            sys.stderr.write(result.advisory.rstrip("\n") + "\n")
            sys.stderr.flush()

        if result.decision == PASS:
            continue

        # A short-circuit check emits its decisive result immediately — this
        # reproduces a legacy hook that calls emit_decision()/hard_block() and
        # exits the instant it fires (the curated destructive set's hard_block
        # and the smart-rule's allow both behave this way standalone). No later
        # check can override a short-circuited decision.
        if check.short_circuit:
            _emit_and_exit(result.decision, result.reason)

        # Otherwise, keep the strongest decision. First-wins on ties keeps
        # the priority-ordered chain's behavior (an earlier deny's reason is
        # the one surfaced).
        if DECISION_RANK[result.decision] > DECISION_RANK[best_decision]:
            best_decision = result.decision
            best_reason = result.reason

    _emit_and_exit(best_decision, best_reason)

lib/pretooluse_bash_checks.py

"""Dispatcher handlers for the PreToolUse:Bash check chain.

These wrap the ALREADY-TESTED pure functions of the legacy per-process hooks
into the dispatcher's Result contract. They deliberately do NOT re-implement
any regex or branch-protection logic — each handler imports the legacy hook
module and calls its existing functions, so the dispatched path and the
legacy path share one source of truth and cannot drift.

Mapping (legacy hook → dispatcher check), in the order the legacy
settings.partial.json registers them for PreToolUse:Bash:

  enforce-git-workflow.py        → git_workflow_check        (hard_block, bypass-safe)
  auto-approve-bash.py           → destructive_check         (hard_block, bypass-safe, short-circuit)
                                   smart_rules_check          (hard_block/allow, bypass-safe)
                                   pattern_check              (deny/allow, NOT bypass-safe)
  port-check.py                  → port_advisory_check        (advisory only)
  agent-tracking-pre.py          → agent_tracking_check       (advisory only)
  check-migration-timestamps.py  → migration_timestamp_check  (hard_block, bypass-safe)
  check-careful.py               → force_push_main_check      (hard_block, bypass-safe)
                                   careful_check              (ask, NOT bypass-safe)

Precedence inside the dispatcher (DECISION_RANK) reproduces the legacy chain:
any hard_block beats deny beats allow beats ask. The destructive set is
marked short_circuit so it is emitted the instant it fires — identical to
auto-approve-bash.py running it ABOVE everything else.
"""
from __future__ import annotations

import importlib.util
import io
import os
import sys
from contextlib import redirect_stderr

sys.path.insert(0, os.path.expanduser("~/.claude/lib"))
import hook_dispatcher as hd  # noqa: E402

# Where the legacy hook scripts live. Defaults to the installed location but
# is overridable (CCGM_HOOKS_DIR) so the test suite can point at the in-repo
# modules/hooks/hooks directory without installing.
_HOOKS_DIR = os.environ.get(
    "CCGM_HOOKS_DIR", os.path.expanduser("~/.claude/hooks")
)


def _load_hook(module_name: str, filename: str):
    """Import a legacy hook script by path under ~/.claude/hooks.

    The hooks are not on the import path (they are executable scripts), so we
    load them explicitly. Each is side-effect-free at import time (logic lives
    under `def main()` guarded by `if __name__ == '__main__'`).
    """
    path = os.path.join(_HOOKS_DIR, filename)
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load {filename} from {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _command(data: dict) -> str:
    return data.get("tool_input", {}).get("command", "") or ""


# ─── enforce-git-workflow.py ─────────────────────────────────────────
def git_workflow_check(data: dict) -> "hd.Result":
    """Protected-branch + issue-number enforcement (bypass-proof).

    The legacy hook calls hook_utils.hard_block() directly inside
    check_commit/check_push. We capture that exit-2 by running its main() in a
    controlled way: instead, call the underlying validators and translate a
    hard_block into a HARD_BLOCK Result. To avoid re-implementing the parsing,
    we reuse the module's own predicates and let it raise SystemExit(2), which
    we convert to a Result so the dispatcher owns emission.
    """
    egw = _load_hook("egw", "enforce-git-workflow.py")
    command = _command(data).strip()
    if data.get("tool_name", "") != "Bash" or not command:
        return hd.Result()
    if not (egw.is_commit_command(command) or egw.is_push_command(command)):
        return hd.Result()
    branch = egw.get_current_branch()
    if not branch:
        return hd.Result()

    # check_commit / check_push call hook_utils.hard_block() (→ SystemExit 2)
    # on a violation, writing the reason to stderr. Run them with stderr
    # captured so we can lift the reason into a Result rather than letting the
    # process die here — the dispatcher decides when to exit.
    buf = io.StringIO()
    try:
        with redirect_stderr(buf):
            if egw.is_commit_command(command):
                egw.check_commit(command, branch)
            elif egw.is_push_command(command):
                egw.check_push(command, branch)
    except SystemExit as exc:
        if exc.code == 2:
            return hd.Result(hd.HARD_BLOCK, buf.getvalue().rstrip("\n"))
        # Any non-2 exit from the validator is treated as no-decision.
        return hd.Result()
    # The validators may print a WARNING (ALLOW_MAIN_COMMIT bypass) to stderr
    # without blocking; surface it as advisory.
    warned = buf.getvalue()
    if warned.strip():
        return hd.Result(advisory=warned.rstrip("\n"))
    return hd.Result()


# ─── auto-approve-bash.py ────────────────────────────────────────────
def destructive_check(data: dict) -> "hd.Result":
    """Curated destructive set (whole-disk/root destruction). Bypass-proof."""
    aab = _load_hook("aab", "auto-approve-bash.py")
    command = _command(data)
    if data.get("tool_name", "") != "Bash" or not command:
        return hd.Result()
    label, reason = aab.check_destructive(command)
    if label:
        return hd.Result(hd.HARD_BLOCK, reason or "destructive command blocked")
    return hd.Result()


def force_branch_delete_check(data: dict) -> "hd.Result":
    """`git branch` force-delete: hard-block naming the segment and the way out.

    Bypass-safe on purpose. In bypass mode the pattern check never runs, so
    without this the only message an agent sees is Claude Code's own generic
    "Permission to use Bash with command <whole chain> has been denied" —
    which reads as if worktree removal were blocked (issue #907).
    """
    aab = _load_hook("aab", "auto-approve-bash.py")
    command = _command(data)
    if data.get("tool_name", "") != "Bash" or not command:
        return hd.Result()
    segment, reason = aab.check_force_branch_delete(command)
    if segment:
        return hd.Result(hd.HARD_BLOCK, reason or "force branch delete blocked")
    return hd.Result()


def smart_rules_check(data: dict) -> "hd.Result":
    """git reset --hard smart-rule: allow remote-ref resets, hard-block others."""
    aab = _load_hook("aab", "auto-approve-bash.py")
    command = _command(data)
    if data.get("tool_name", "") != "Bash" or not command:
        return hd.Result()
    decision, reason = aab.check_smart_rules(command)
    if decision == "hard_block":
        return hd.Result(hd.HARD_BLOCK, reason or "destructive smart-rule matched")
    if decision == "allow":
        return hd.Result(hd.ALLOW, reason or "smart-rule allow")
    return hd.Result()


def pattern_check(data: dict) -> "hd.Result":
    """settings.json allow/deny pattern matching, per-segment (#660 fix).

    NOT bypass-safe: in bypass mode the dispatcher skips this check, exactly
    as auto-approve-bash.py exits 0 before pattern matching when bypass is on.
    """
    aab = _load_hook("aab", "auto-approve-bash.py")
    command = _command(data)
    if data.get("tool_name", "") != "Bash" or not command:
        return hd.Result()
    allow_patterns, deny_patterns = aab.load_settings()
    decision, reason = aab.check_pattern_decision(command, allow_patterns, deny_patterns)
    if decision == "deny":
        return hd.Result(hd.DENY, reason or "")
    if decision == "allow":
        return hd.Result(hd.ALLOW, reason or "")
    return hd.Result()


# ─── port-check.py ───────────────────────────────────────────────────
def port_advisory_check(data: dict) -> "hd.Result":
    """Dev-server port-allocation warnings. Advisory only, never blocks.

    NOT bypass-safe: port-check.py returns silently in bypass mode.
    """
    pc = _load_hook("pc", "port-check.py")
    if data.get("tool_name", "") != "Bash":
        return hd.Result()
    command = _command(data)
    if not pc.is_dev_server_command(command):
        return hd.Result()
    # port-check.py writes warnings to stderr inside main(). Run main() with
    # stderr captured and surface the text as advisory. main() never emits a
    # permission decision, so nothing else leaks.
    buf = io.StringIO()
    try:
        with redirect_stderr(buf):
            pc.main_with_data(data) if hasattr(pc, "main_with_data") else _run_port_main(pc, data)
    except SystemExit:
        pass
    text = buf.getvalue()
    if text.strip():
        return hd.Result(advisory=text.rstrip("\n"))
    return hd.Result()


def _run_port_main(pc, data: dict) -> None:
    """port-check.py.main() reads its own stdin; feed it our envelope."""
    import json
    saved = sys.stdin
    sys.stdin = io.StringIO(json.dumps(data))
    try:
        pc.main()
    finally:
        sys.stdin = saved


# ─── agent-tracking-pre.py ───────────────────────────────────────────
def agent_tracking_check(data: dict) -> "hd.Result":
    """Multi-agent issue-claim warnings. Advisory only, never blocks."""
    at = _load_hook("atp", "agent-tracking-pre.py")
    if data.get("tool_name", "") != "Bash":
        return hd.Result()
    import json
    buf_out = io.StringIO()
    saved_in, saved_out = sys.stdin, sys.stdout
    sys.stdin = io.StringIO(json.dumps(data))
    sys.stdout = buf_out
    try:
        at.main()
    except SystemExit:
        pass
    finally:
        sys.stdin, sys.stdout = saved_in, saved_out
    # agent-tracking-pre emits its warnings as JSON with only a
    # permissionDecisionReason (no permissionDecision) — that is advisory by
    # Claude Code's contract. Surface it as advisory text.
    text = buf_out.getvalue()
    if text.strip():
        return hd.Result(advisory=text.rstrip("\n"))
    return hd.Result()


# ─── check-migration-timestamps.py ───────────────────────────────────
def migration_timestamp_check(data: dict) -> "hd.Result":
    """Duplicate Supabase migration timestamps. Data-integrity hard_block."""
    cmt = _load_hook("cmt", "check-migration-timestamps.py")
    import json
    buf = io.StringIO()
    saved_in = sys.stdin
    sys.stdin = io.StringIO(json.dumps(data))
    try:
        with redirect_stderr(buf):
            cmt.main()
    except SystemExit as exc:
        if exc.code == 2:
            return hd.Result(hd.HARD_BLOCK, buf.getvalue().rstrip("\n"))
    finally:
        sys.stdin = saved_in
    return hd.Result()


# ─── check-careful.py ────────────────────────────────────────────────
def force_push_main_check(data: dict) -> "hd.Result":
    """Force-push to main. Bypass-proof hard_block (gated by ALLOW_MAIN_COMMIT)."""
    cc = _load_hook("cc", "check-careful.py")
    if data.get("tool_name", "") != "Bash":
        return hd.Result()
    command = _command(data)
    if not command:
        return hd.Result()
    if cc._is_force_push_to_main(command) and os.environ.get("ALLOW_MAIN_COMMIT") != "1":
        return hd.Result(
            hd.HARD_BLOCK,
            "BLOCKED: force-pushing to `main` overwrites shared history. "
            "If this is truly intended (recovering from a bad merge, etc.), "
            "re-run with `ALLOW_MAIN_COMMIT=1` set.",
        )
    return hd.Result()


def careful_check(data: dict) -> "hd.Result":
    """Destructive-command prompt (ask). NOT bypass-safe."""
    cc = _load_hook("cc", "check-careful.py")
    if data.get("tool_name", "") != "Bash":
        return hd.Result()
    command = _command(data)
    if not command:
        return hd.Result()
    is_destructive, reason = cc.check_careful(command)
    if is_destructive:
        return hd.Result(hd.ASK, reason)
    return hd.Result()

lib/sched_platform.py

"""Platform abstraction for scheduled-job installation.

Locked API (referenced from plan.md §5 Epic 1, §3.11 Linux portability):

    install_scheduled_job(label: str, command: str, hour: int, minute: int) -> None
    uninstall_scheduled_job(label: str) -> None
    list_scheduled_jobs() -> list[str]

macOS uses launchd LaunchAgents under ~/Library/LaunchAgents/.
Linux is a documented v2 plug-in seam — the implementation raises
NotImplementedError with a clear message pointing at the planned cron
template.

Note: the plan spells this `lib/platform.py`, but that name shadows the
Python stdlib `platform` module for any hook that puts `~/.claude/lib`
on `sys.path`. Renamed to `sched_platform.py` to keep the stdlib
reachable. The locked API (install_scheduled_job, uninstall_scheduled_job,
list_scheduled_jobs) is unchanged.
"""
from __future__ import annotations

import os
import platform as _platform
import plistlib
import shutil
import subprocess
from typing import NoReturn

__all__ = [
    "install_scheduled_job",
    "uninstall_scheduled_job",
    "list_scheduled_jobs",
    "LAUNCH_AGENTS_DIR",
]


LAUNCH_AGENTS_DIR = os.path.expanduser("~/Library/LaunchAgents")


def _linux_v2(_op: str) -> NoReturn:
    raise NotImplementedError(
        "Linux scheduling is a v2 plug-in point. "
        "See modules/autoheal/lib/autoheal.cron.template for the planned "
        "cron implementation; the macOS launchd path is in lib/sched_platform.py."
    )


def _unsupported(op: str) -> NoReturn:
    raise NotImplementedError(
        f"Platform not supported for {op}: {_platform.system()!r}. "
        "macOS (Darwin) is the only v1 target; Linux is a v2 plug-in seam."
    )


def install_scheduled_job(label: str, command: str, hour: int, minute: int) -> None:
    """Install a daily scheduled job to fire at HH:MM local time.

    macOS: writes a launchd plist at ~/Library/LaunchAgents/{label}.plist
    and bootstraps it via `launchctl bootstrap gui/$UID`. Idempotent — an
    existing job with the same label is bootout'd first.

    Linux: raises NotImplementedError (v2 seam).
    """
    if not (0 <= hour <= 23 and 0 <= minute <= 59):
        raise ValueError(f"Bad schedule: hour={hour}, minute={minute}")

    sysname = _platform.system()
    if sysname == "Darwin":
        _install_launchd(label, command, hour, minute)
        return
    if sysname == "Linux":
        _linux_v2("install_scheduled_job")
    _unsupported("install_scheduled_job")


def uninstall_scheduled_job(label: str) -> None:
    """Remove a previously-installed scheduled job.

    macOS: bootout the launchd job and delete its plist. Tolerates a
    missing plist (treats it as already-uninstalled).

    Linux: raises NotImplementedError.
    """
    sysname = _platform.system()
    if sysname == "Darwin":
        _uninstall_launchd(label)
        return
    if sysname == "Linux":
        _linux_v2("uninstall_scheduled_job")
    _unsupported("uninstall_scheduled_job")


def list_scheduled_jobs() -> list[str]:
    """Return labels of currently-installed scheduled jobs.

    macOS: reads ~/Library/LaunchAgents and returns the basename
    (without .plist) of every plist whose Label matches the file name.
    Cheap and doesn't shell out.

    Linux: raises NotImplementedError.
    """
    sysname = _platform.system()
    if sysname == "Darwin":
        return _list_launchd()
    if sysname == "Linux":
        _linux_v2("list_scheduled_jobs")
    _unsupported("list_scheduled_jobs")


def _plist_path(label: str) -> str:
    return os.path.join(LAUNCH_AGENTS_DIR, f"{label}.plist")


def _gui_target() -> str:
    uid = os.getuid()
    return f"gui/{uid}"


def _install_launchd(label: str, command: str, hour: int, minute: int) -> None:
    os.makedirs(LAUNCH_AGENTS_DIR, exist_ok=True)
    path = _plist_path(label)

    # Best-effort uninstall first so a config change takes effect.
    if os.path.exists(path):
        _uninstall_launchd(label)

    plist = {
        "Label": label,
        "ProgramArguments": ["/bin/sh", "-c", command],
        "StartCalendarInterval": {"Hour": hour, "Minute": minute},
        "RunAtLoad": False,
        "StandardOutPath": os.path.expanduser(f"~/.claude/logs/{label}.out.log"),
        "StandardErrorPath": os.path.expanduser(f"~/.claude/logs/{label}.err.log"),
        "EnvironmentVariables": {
            "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin",
        },
    }
    os.makedirs(os.path.expanduser("~/.claude/logs"), exist_ok=True)
    with open(path, "wb") as fh:
        plistlib.dump(plist, fh)

    if shutil.which("launchctl"):
        # bootstrap is the modern (Big Sur+) way; ignore non-zero on
        # already-loaded (launchctl is grumpy about idempotency).
        subprocess.run(
            ["launchctl", "bootstrap", _gui_target(), path],
            check=False,
            capture_output=True,
        )


def _uninstall_launchd(label: str) -> None:
    path = _plist_path(label)
    if shutil.which("launchctl") and os.path.exists(path):
        subprocess.run(
            ["launchctl", "bootout", _gui_target(), path],
            check=False,
            capture_output=True,
        )
    if os.path.exists(path):
        try:
            os.remove(path)
        except OSError:
            pass


def _list_launchd() -> list[str]:
    if not os.path.isdir(LAUNCH_AGENTS_DIR):
        return []
    labels: list[str] = []
    for entry in os.listdir(LAUNCH_AGENTS_DIR):
        if entry.endswith(".plist"):
            labels.append(entry[: -len(".plist")])
    labels.sort()
    return labels

settings.partial.json

Merged into ~/.claude/settings.json -- a fragment, not a replacement.

{
  "hooks": {
    "SessionStart": [
      {
        "matcher": "startup",
        "hooks": [
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/session-start-enforce.py",
            "timeout": 5000
          }
        ]
      }
    ],
    "UserPromptSubmit": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/enforce-issue-workflow.py",
            "timeout": 5000
          },
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/ccgm-update-check.py",
            "timeout": 15000
          }
        ]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/agent-tracking-post.py",
            "timeout": 15000
          },
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/sync-ccgm-canonical.py",
            "timeout": 35000
          }
        ]
      }
    ],
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/pretooluse-bash-dispatch.py",
            "timeout": 5000
          }
        ]
      },
      {
        "matcher": "Read",
        "hooks": [
          {
            "type": "command",
            "command": "$HOME/.claude/hooks/auto-approve-file-ops.py",
            "timeout": 5000
          }
        ]
      },
      {
        "matcher": "Edit",
        "hooks": [
          {
            "type": "command",
            "command": "$HOME/.claude/hooks/auto-approve-file-ops.py",
            "timeout": 5000
          },
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/check-freeze.py",
            "timeout": 5000
          }
        ]
      },
      {
        "matcher": "Write",
        "hooks": [
          {
            "type": "command",
            "command": "$HOME/.claude/hooks/auto-approve-file-ops.py",
            "timeout": 5000
          },
          {
            "type": "command",
            "command": "python3 $HOME/.claude/hooks/check-freeze.py",
            "timeout": 5000
          }
        ]
      }
    ]
  }
}