Skip to content
Security
Hook

Hooks

What lacp runs automatically, and when. A hook is a command Claude Code fires at a fixed moment, without you asking for it.

@0xnyk0xnyk@0xnyk
From plugin
lacp
2963 skills1 command14 hooks
Install
> /plugin marketplace add 0xNyk/lacp
> /plugin install lacp-hardening@lacp-plugins

Ships with lacp. Installing the plugin gets these hooks.

Where it lives

  • hooks/detect_session_changes.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""Detect file changes made by Claude during a session.

Reads a JSON object from stdin with {"transcript_path": "..."}.
Streams the JSONL transcript line-by-line, extracts file paths from
Write/Edit/NotebookEdit tool_use blocks, and reports which files exist.

Prints a single JSON line to stdout.
"""

import json
import os
import sys

TRACKED_TOOLS = {"Write", "Edit", "NotebookEdit"}


def scan_transcript(path: str) -> dict:
    files_seen: dict[str, set[str]] = {}  # file_path -> set of tool names
    tools_used: dict[str, int] = {}

    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except (json.JSONDecodeError, ValueError):
                    continue

                message = obj.get("message") if isinstance(obj, dict) else None
                if not isinstance(message, dict):
                    continue

                content = message.get("content")
                if not isinstance(content, list):
                    continue

                for block in content:
                    if not isinstance(block, dict):
                        continue
                    if block.get("type") != "tool_use":
                        continue

                    name = block.get("name", "")
                    if name not in TRACKED_TOOLS:
                        continue

                    inp = block.get("input")
                    if not isinstance(inp, dict):
                        continue

                    file_path = inp.get("file_path") or inp.get("notebook_path")
                    if not file_path or not isinstance(file_path, str):
                        continue

                    tools_used[name] = tools_used.get(name, 0) + 1

                    if file_path not in files_seen:
                        files_seen[file_path] = set()
                    files_seen[file_path].add(name)

    except OSError as e:
        return {"files_changed": -1, "files": [], "tools_used": {}, "error": str(e)}

    # Only report files that still exist on disk
    existing = [p for p in files_seen if os.path.exists(p)]
    existing.sort()

    return {
        "files_changed": len(existing),
        "files": existing,
        "tools_used": tools_used,
    }


def main() -> None:
    try:
        raw = sys.stdin.read()
        request = json.loads(raw)
    except (json.JSONDecodeError, ValueError) as e:
        json.dump(
            {"files_changed": -1, "files": [], "tools_used": {}, "error": f"invalid input JSON: {e}"},
            sys.stdout,
        )
        print()
        return

    if not isinstance(request, dict):
        json.dump(
            {"files_changed": -1, "files": [], "tools_used": {}, "error": "input must be a JSON object"},
            sys.stdout,
        )
        print()
        return

    transcript_path = request.get("transcript_path")
    if not transcript_path or not isinstance(transcript_path, str):
        json.dump(
            {"files_changed": -1, "files": [], "tools_used": {}, "error": "missing transcript_path"},
            sys.stdout,
        )
        print()
        return

    if not os.path.isfile(transcript_path):
        json.dump(
            {"files_changed": -1, "files": [], "tools_used": {}, "error": f"file not found: {transcript_path}"},
            sys.stdout,
        )
        print()
        return

    result = scan_transcript(transcript_path)
    json.dump(result, sys.stdout)
    print()


if __name__ == "__main__":
    main()
  • hooks/eval_checkpoint.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""PostToolUse(Write|Edit) checkpoint — continuous QA during work.

Runs the project's test command at configurable intervals during a session.
Injects feedback via systemMessage when tests fail, without blocking.

Hook protocol (PostToolUse hook, matcher: Write|Edit):
  - exit 0 with no stdout → no-op (silent pass or not yet at interval)
  - exit 0 with {"systemMessage": "..."} → inject test failure feedback
"""

from __future__ import annotations

import json
import os
import re
import shlex
import subprocess
import sys
import time
from pathlib import Path
from typing import Optional


# -- Configuration --

CHECKPOINT_ENABLED = os.getenv("LACP_EVAL_CHECKPOINT_ENABLED", "0") == "1"
CHECKPOINT_INTERVAL = int(os.getenv("LACP_EVAL_CHECKPOINT_INTERVAL", "10"))
TEST_TIMEOUT = int(os.getenv("LACP_EVAL_CHECKPOINT_TIMEOUT", "30"))

_LACP_STATE_DIR = Path.home() / ".lacp" / "hooks" / "state"

# Allowed test runner prefixes (same as stop_quality_gate.py)
_ALLOWED_TEST_RUNNERS = (
    "bun test", "pnpm test", "yarn test", "npm test",
    "make test", "cargo test", "python3 -m pytest",
    "bin/lacp-test",
)

HOOKS_DIR = Path(__file__).parent
sys.path.insert(0, str(HOOKS_DIR))


def _safe_session_id(session_id: str) -> str:
    return re.sub(r"[^a-zA-Z0-9_-]", "_", session_id) if session_id else "default"


def _session_state_dir(session_id: str) -> Path:
    safe_id = _safe_session_id(session_id)
    d = _LACP_STATE_DIR / safe_id
    d.mkdir(parents=True, exist_ok=True, mode=0o700)
    return d


def _get_write_count(session_id: str) -> int:
    counter_file = _session_state_dir(session_id) / "write-count"
    try:
        return int(counter_file.read_text().strip()) if counter_file.exists() else 0
    except (ValueError, OSError):
        return 0


def _set_write_count(session_id: str, count: int) -> None:
    counter_file = _session_state_dir(session_id) / "write-count"
    try:
        counter_file.write_text(str(count))
    except OSError:
        pass


def _get_test_command(session_id: str) -> Optional[str]:
    """Read cached test command from session start contract."""
    try:
        from hook_contracts import read_contract
        contract = read_contract("session_start", session_id)
        if contract and contract.get("test_cmd"):
            cmd = contract["test_cmd"]
            if any(cmd.startswith(prefix) for prefix in _ALLOWED_TEST_RUNNERS):
                return cmd
    except Exception:
        pass
    return None


def _update_checkpoint_contract(session_id: str, result: str, fail_count: int) -> None:
    """Update the eval checkpoint contract for telemetry/stop hook consumption."""
    try:
        from hook_contracts import EvalCheckpoint, write_contract
        checkpoint = EvalCheckpoint(
            write_count=_get_write_count(session_id),
            last_check_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            last_result=result,
            fail_count=fail_count,
        )
        write_contract("eval_checkpoint", checkpoint, session_id)
    except Exception:
        pass


def _get_fail_count(session_id: str) -> int:
    """Read current fail count from checkpoint contract."""
    try:
        from hook_contracts import read_contract
        contract = read_contract("eval_checkpoint", session_id)
        if contract:
            return int(contract.get("fail_count", 0))
    except Exception:
        pass
    return 0


def main() -> None:
    if not CHECKPOINT_ENABLED:
        return

    raw = sys.stdin.read()
    try:
        hook_input = json.loads(raw) if raw.strip() else {}
    except json.JSONDecodeError:
        return

    session_id = hook_input.get("session_id") or ""
    cwd = hook_input.get("cwd") or ""

    # Increment write counter
    count = _get_write_count(session_id) + 1
    _set_write_count(session_id, count)

    # Check if we're at an interval
    if count % CHECKPOINT_INTERVAL != 0:
        return

    # Find test command
    test_cmd = _get_test_command(session_id)
    if not test_cmd:
        return

    # Run tests
    try:
        result = subprocess.run(
            shlex.split(test_cmd),
            shell=False,
            capture_output=True,
            text=True,
            timeout=TEST_TIMEOUT,
            cwd=cwd or None,
        )
    except (subprocess.TimeoutExpired, OSError):
        return  # Fail-open

    fail_count = _get_fail_count(session_id)

    if result.returncode == 0:
        _update_checkpoint_contract(session_id, "pass", fail_count)
        return  # Silent success

    # Tests failed — inject feedback (don't block)
    fail_count += 1
    _update_checkpoint_contract(session_id, "fail", fail_count)

    output = (result.stdout or "") + (result.stderr or "")
    last_lines = "\n".join(output.strip().splitlines()[-8:])

    # Think-mode enhancement: inject thinking prompt alongside failure report
    think_prompt = ""
    if os.getenv("LACP_CONTEXT_MODE", "") in ("think", "tdd", "debugging"):
        think_prompt = (
            "\n\nBefore making changes, think about: "
            "What changed since tests last passed? "
            "Which specific file most likely caused this failure? "
            "What is the minimal fix (not a rewrite)?"
        )

    msg = (
        f"Checkpoint ({count} writes): tests failing (exit {result.returncode}). "
        f"Fix before continuing:\n{last_lines}{think_prompt}"
    )
    print(json.dumps({"systemMessage": msg}))


if __name__ == "__main__":
    main()
  • hooks/extract_memories.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""Per-turn memory extraction — captures durable signals from conversation.

Lightweight Stop hook that scans the last assistant message for memory-worthy
signals and writes them to a staging file. Brain-expand later promotes from
staging into the knowledge graph.

No LLM calls — heuristic pattern matching only. Keeps hook latency <100ms.

Hook protocol (Stop hook):
  - Reads JSON from stdin: {transcript_path, session_id, cwd, ...}
  - Writes signals to staging file: ~/.lacp/memory-staging/pending.jsonl
  - Exits 0 (never blocks stop)
"""
from __future__ import annotations

import json
import os
import re
import sys
from datetime import UTC, datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent / "automation" / "scripts"))

STAGING_DIR = Path.home() / ".lacp" / "memory-staging"
STAGING_FILE = STAGING_DIR / "pending.jsonl"
MAX_STAGING_LINES = 500  # rotate after this many

# Patterns that indicate durable, memory-worthy content
MEMORY_INDICATORS = [
    # Decisions and preferences
    re.compile(r"(?:we decided|the decision is|going with|chose to|prefer)\s+(.{20,})", re.I),
    # Lessons learned
    re.compile(r"(?:lesson learned|takeaway|key insight|important note|remember that)\s*:?\s*(.{20,})", re.I),
    # Architecture/design choices
    re.compile(r"(?:architecture|design choice|trade-?off|we.ll use|stack is)\s*:?\s*(.{20,})", re.I),
    # Process/workflow
    re.compile(r"(?:workflow|process|convention|rule|policy)\s*:?\s*(.{20,})", re.I),
    # User corrections (feedback type)
    re.compile(r"(?:don.t|never|always|stop|instead of)\s+(.{15,})", re.I),
]

# Anti-patterns — skip these even if they match indicators
SKIP_PATTERNS = [
    re.compile(r"```[\s\S]{100,}```"),  # large code blocks
    re.compile(r"^\s*[-*]\s+`[^`]+`\s*$", re.M),  # bullet list of code refs
    re.compile(r"(?:error|traceback|stack trace)", re.I),  # error output
]


def extract_signals(text: str, session_id: str, cwd: str) -> list[dict]:
    """Extract memory-worthy signals from assistant message text."""
    if not text or len(text) < 50:
        return []

    # Skip if text is mostly code
    code_ratio = text.count("```") / max(1, len(text) / 100)
    if code_ratio > 0.3:
        return []

    # Check anti-patterns
    for pattern in SKIP_PATTERNS:
        if pattern.search(text):
            return []

    signals = []
    for pattern in MEMORY_INDICATORS:
        for match in pattern.finditer(text):
            signal_text = match.group(1).strip() if match.lastindex else match.group(0).strip()
            # Clean up
            signal_text = re.sub(r"\s+", " ", signal_text)
            if len(signal_text) < 20 or len(signal_text) > 300:
                continue
            signals.append({
                "text": signal_text[:280],
                "source": "per-turn-extraction",
                "session_id": session_id,
                "cwd": cwd,
                "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
                "day": datetime.now(UTC).strftime("%Y-%m-%d"),
            })

    # Deduplicate within this extraction
    seen = set()
    unique = []
    for s in signals:
        key = s["text"][:50].lower()
        if key not in seen:
            seen.add(key)
            unique.append(s)

    return unique[:5]  # cap at 5 per turn


def append_to_staging(signals: list[dict]) -> int:
    """Append signals to staging JSONL file. Returns count written."""
    if not signals:
        return 0

    STAGING_DIR.mkdir(parents=True, exist_ok=True)

    # Rotate if too large
    if STAGING_FILE.exists():
        line_count = sum(1 for _ in STAGING_FILE.open())
        if line_count > MAX_STAGING_LINES:
            archive = STAGING_DIR / f"staging-{datetime.now(UTC).strftime('%Y%m%d')}.jsonl"
            STAGING_FILE.rename(archive)

    with STAGING_FILE.open("a", encoding="utf-8") as f:
        for signal in signals:
            f.write(json.dumps(signal) + "\n")

    return len(signals)


def _extract_last_assistant(transcript_path: str) -> str:
    """Extract last assistant message from JSONL transcript."""
    if not transcript_path or not os.path.isfile(transcript_path):
        return ""
    last = ""
    try:
        with open(transcript_path, encoding="utf-8", errors="ignore") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    msg = json.loads(line)
                    if msg.get("type") == "assistant":
                        content = msg.get("message", {}).get("content", "")
                        if isinstance(content, list):
                            # Extract text parts
                            text_parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"]
                            content = "\n".join(text_parts)
                        if isinstance(content, str) and len(content) > 50:
                            last = content
                except json.JSONDecodeError:
                    continue
    except OSError:
        pass
    return last


def main() -> int:
    try:
        hook_input = json.loads(sys.stdin.read())
    except (json.JSONDecodeError, EOFError):
        return 0  # silent exit on bad input

    session_id = hook_input.get("session_id", "")
    cwd = hook_input.get("cwd", "")
    transcript_path = hook_input.get("transcript_path", "")

    # Get the last assistant message
    last_message = hook_input.get("last_assistant_message", "")
    if not last_message:
        last_message = _extract_last_assistant(transcript_path)

    if not last_message:
        return 0

    signals = extract_signals(last_message, session_id, cwd)
    count = append_to_staging(signals)

    if count > 0 and os.environ.get("LACP_EXTRACT_DEBUG") == "1":
        print(json.dumps({"extracted": count, "signals": signals}), file=sys.stderr)

    retur
  • hooks/hook_contracts.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""Typed hook state contracts — shared state between hooks within a session.

Provides a simple contract system where hooks can write structured data
that other hooks can read with schema validation. Contracts are stored
per-session and auto-cleaned.

Inspired by herm's tool interface (Definition → ToolDefinition, Execute → (string, error))
and langdag's typed ContentBlock/Node structures.
"""

from __future__ import annotations

import json
import os
import re
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Optional


def _safe_session_id(session_id: str) -> str:
    """Sanitize session_id for use in file paths (L1: CWE-22)."""
    return re.sub(r"[^a-zA-Z0-9_-]", "_", session_id) if session_id else "default"


# Contract storage location
def _contracts_dir(session_id: str) -> Path:
    return Path.home() / ".lacp" / "hooks" / "contracts" / _safe_session_id(session_id)


@dataclass
class SessionStartOutput:
    test_cmd: Optional[str] = None
    git_branch: Optional[str] = None
    context_mode: Optional[str] = None
    started_at: Optional[str] = None
    context_budget_hint: Optional[int] = None


@dataclass
class StopGateInput:
    test_cmd: Optional[str] = None
    session_changes: Optional[list[str]] = None
    transcript_path: Optional[str] = None
    session_started_at: Optional[str] = None
    context_budget_hint: Optional[int] = None
    tool_use_count: Optional[int] = None


def write_contract(name: str, data: object, session_id: str | None = None) -> Path | None:
    """Write a contract to disk. Returns the path written, or None on failure."""
    sid = session_id or os.getenv("CLAUDE_SESSION_ID", "default")
    contracts = _contracts_dir(sid)
    try:
        contracts.mkdir(parents=True, exist_ok=True)
        path = contracts / f"{name}.json"
        payload = asdict(data) if hasattr(data, "__dataclass_fields__") else data
        path.write_text(json.dumps(payload, default=str))
        return path
    except (OSError, TypeError):
        return None


def read_contract(name: str, session_id: str | None = None) -> dict | None:
    """Read a contract from disk. Returns dict or None if missing/invalid."""
    sid = session_id or os.getenv("CLAUDE_SESSION_ID", "default")
    path = _contracts_dir(sid) / f"{name}.json"
    if not path.is_file():
        return None
    try:
        return json.loads(path.read_text())
    except (json.JSONDecodeError, OSError):
        return None


@dataclass
class SprintContract:
    acceptance_criteria: list[str]
    expected_files: list[str]
    expected_tests: list[str]
    agreed_at: str = ""


@dataclass
class EvalCheckpoint:
    write_count: int = 0
    last_check_at: str = ""
    last_result: str = ""  # "pass", "fail", or ""
    fail_count: int = 0


@dataclass
class HandoffArtifact:
    task_summary: str = ""
    files_modified: list[str] = None
    open_issues: list[str] = None
    next_steps: list[str] = None
    test_status: str = "unknown"  # "pass", "fail", "unknown"
    git_branch: str = ""
    git_diff_summary: str = ""
    created_at: str = ""

    def __post_init__(self):
        if self.files_modified is None:
            self.files_modified = []
        if self.open_issues is None:
            self.open_issues = []
        if self.next_steps is None:
            self.next_steps = []


@dataclass
class TaskPlan:
    subtasks: list[dict] = None  # [{name, files, depends_on, status, scope}]
    created_at: str = ""

    def __post_init__(self):
        if self.subtasks is None:
            self.subtasks = []


def cleanup_contracts(session_id: str | None = None) -> int:
    """Remove all contracts for a session. Returns count of files removed."""
    sid = session_id or os.getenv("CLAUDE_SESSION_ID", "default")
    contracts = _contracts_dir(sid)
    if not contracts.is_dir():
        return 0
    count = 0
    try:
        for f in contracts.iterdir():
            f.unlink()
            count += 1
        contracts.rmdir()
    except OSError:
        pass
    return count


def cleanup_stale_contracts(max_age_hours: int = 48) -> int:
    """Remove contract directories older than max_age_hours. Returns count removed."""
    import time as _time
    contracts_root = Path.home() / ".lacp" / "hooks" / "contracts"
    if not contracts_root.is_dir():
        return 0
    cutoff = _time.time() - (max_age_hours * 3600)
    removed = 0
    try:
        for d in contracts_root.iterdir():
            if not d.is_dir():
                continue
            try:
                mtime = max(f.stat().st_mtime for f in d.iterdir()) if any(d.iterdir()) else d.stat().st_mtime
                if mtime < cutoff:
                    for f in d.iterdir():
                        f.unlink()
                    d.rmdir()
                    removed += 1
            except OSError:
                continue
    except OSError:
        pass
    return removed


def cleanup_stale_state(max_age_hours: int = 48) -> int:
    """Remove session state directories older than max_age_hours. Returns count removed."""
    import time as _time
    state_root = Path.home() / ".lacp" / "hooks" / "state"
    if not state_root.is_dir():
        return 0
    cutoff = _time.time() - (max_age_hours * 3600)
    removed = 0
    try:
        for d in state_root.iterdir():
            if not d.is_dir():
                continue
            if d.name == "quality-gate.log":
                continue  # skip the debug log file
            try:
                mtime = max(f.stat().st_mtime for f in d.iterdir()) if any(d.iterdir()) else d.stat().st_mtime
                if mtime < cutoff:
                    for f in d.iterdir():
                        f.unlink()
                    d.rmdir()
                    removed += 1
            except OSError:
                continue
    except OSError:
        pass
    return removed
  • hooks/hook_telemetry.pyGitHub
    Read the script
    #!/usr/bin/env python3
"""Lightweight telemetry logger for Claude Code hooks.

Append-only JSONL log with auto-rotation. Usable as CLI or importable module.
Python 3.11 stdlib only.
"""

import argparse
import fcntl
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

TELEMETRY_DIR = Path.home() / ".local" / "share" / "claude-hooks"
TELEMETRY_FILE = TELEMETRY_DIR / "telemetry.jsonl"
MAX_SIZE = 10 * 1024 * 1024  # 10 MB
MAX_ROTATIONS = 3


def _rotate_if_needed() -> None:
    """Rotate telemetry file if it exceeds MAX_SIZE."""
    try:
        if not TELEMETRY_FILE.exists() or TELEMETRY_FILE.stat().st_size < MAX_SIZE:
            return
    except OSError:
        return

    # Shift .3 -> delete, .2 -> .3, .1 -> .2, current -> .1
    for i in range(MAX_ROTATIONS, 0, -1):
        src = TELEMETRY_FILE.with_suffix(f".{i}")
        if i == MAX_ROTATIONS:
            src.unlink(missing_ok=True)
        else:
            dst = TELEMETRY_FILE.with_suffix(f".{i + 1}")
            if src.exists():
                src.rename(dst)

    TELEMETRY_FILE.rename(TELEMETRY_FILE.with_suffix(".1"))


def log_decision(
    hook: str,
    decision: str,
    reason: str,
    session_id: str = "unknown",
    elapsed_ms: int = 0,
    details: dict | None = None,
) -> None:
    """Append a single telemetry entry (thread-safe)."""
    TELEMETRY_DIR.mkdir(parents=True, exist_ok=True)
    _rotate_if_needed()

    entry = {
        "ts": datetime.now(timezone.utc).astimezone().isoformat(),
        "hook": hook,
        "session_id": session_id,
        "decision": decision,
        "reason": reason,
        "elapsed_ms": elapsed_ms,
        "details": details or {},
    }

    line = json.dumps(entry, separators=(",", ":")) + "\n"

    fd = os.open(str(TELEMETRY_FILE), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)
        os.write(fd, line.encode())
    finally:
        fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)


def main() -> None:
    parser = argparse.ArgumentParser(description="Log a hook telemetry decision")
    parser.add_argument("--hook", required=True, help="Hook name")
    parser.add_argument("--session-id", default="unknown", help="Session ID")
    parser.add_argument(
        "--decision",
        required=True,
        choices=["allow", "block", "skip"],
        help="Decision outcome",
    )
    parser.add_argument("--reason", required=True, help="Human-readable reason")
    parser.add_argument("--elapsed", type=int, default=0, help="Elapsed time in ms")
    parser.add_argument(
        "--detail",
        action="append",
        default=[],
        metavar="KEY=VALUE",
        help="Extra key=value detail (repeatable)",
    )

    args = parser.parse_args()

    details = {}
    for kv in args.detail:
        if "=" in kv:
            k, v = kv.split("=", 1)
            details[k] = v
        else:
            details[kv] = True

    log_decision(
        hook=args.hook,
        session_id=args.session_id,
        decision=args.decision,
        reason=args.reason,
        elapsed_ms=args.elapsed,
        details=details,
    )


if __name__ == "__main__":
    main()
  • hooks/pretool_guard.pyGitHub
    Read the script
    #!/usr/bin/env python3
import hashlib
import json
import os
import re
import shlex
import socket
import subprocess
import sys
import time
from pathlib import Path

BLOCKS = [
    (re.compile(r"\b(?:npm|yarn|pnpm|cargo)\s+publish\b", re.IGNORECASE), "BLOCKED: Publishing to registry requires explicit user approval. Ask the user first."),
    (re.compile(r"\b(?:curl|wget)\b.*\|\s*(?:python3?|node|ruby|perl)\b", re.IGNORECASE), "BLOCKED: Piping network content to an interpreter is unsafe. Download first, review, then run."),
    (re.compile(r"\bchmod\s+(?:-R\s+)?777\b"), "BLOCKED: chmod 777 is overly permissive. Use specific permissions (e.g. 755, 644)."),
    (re.compile(r"\bgit\s+reset\s+--hard\b", re.IGNORECASE), "BLOCKED: git reset --hard is destructive. Ask the user first."),
    (re.compile(r"\bgit\s+clean\s+-f", re.IGNORECASE), "BLOCKED: git clean -f is destructive. Ask the user first."),
    (re.compile(r"\bdocker\s+run\b[^\n\r]*--privileged\b", re.IGNORECASE), "BLOCKED: docker run --privileged is a security risk. Use specific capabilities instead."),
    (re.compile(r":\(\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:"), "BLOCKED: Fork bomb detected."),
    (re.compile(r"\b(?:scp|rsync)\b.*\s/root(?:/|\s|$)", re.IGNORECASE), "BLOCKED: scp/rsync to /root is restricted. Use a non-root target path."),
]

PROTECTED_PATHS = re.compile(r"(\.env($|\.)|config\.toml($|\.)|secrets?|\.claude/settings\.json$|authorized_keys$|\.(pem|key)$|(^|/)\.gnupg(/|$))", re.IGNORECASE)

MUTATING_LOCAL_PATTERNS = [
    re.compile(r"\b(?:python(?:3)?\s+-m\s+venv|uv\s+venv|virtualenv)\b", re.IGNORECASE),
    re.compile(r"\b(?:pip(?:3)?\s+install|poetry\s+install|npm\s+install|pnpm\s+install|yarn\s+install)\b", re.IGNORECASE),
    re.compile(r"\b(?:apt(?:-get)?\s+install|brew\s+install)\b", re.IGNORECASE),
]

HOOKS_DIR = Path.home() / ".claude" / "hooks"
DEFAULT_TTL_SECONDS = 12 * 3600


def _scope_id() -> str:
    # Explicit scope wins.
    explicit = os.getenv("CLAUDE_CONTEXT_SCOPE", "").strip()
    if explicit:
        return explicit

    # Use per-window/per-pane IDs when present.
    for key in ("TMUX_PANE", "WEZTERM_PANE", "ITERM_SESSION_ID", "TERM_SESSION_ID", "WINDOWID"):
        val = os.getenv(key, "").strip()
        if val:
            return f"{key}:{val}"

    # Fallback to current working directory hash.
    cwd = os.getcwd()
    digest = hashlib.sha1(cwd.encode("utf-8")).hexdigest()[:12]
    return f"cwd:{digest}"


def _state_path() -> Path:
    safe = re.sub(r"[^A-Za-z0-9._:-]", "_", _scope_id())
    return HOOKS_DIR / f".remote_context.{safe}.json"


def _read_payload() -> dict:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        return json.loads(raw)
    except Exception:
        return {}


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


def _get_path(payload: dict) -> str:
    p = str(payload.get("tool_input", {}).get("file_path") or "")
    if not p:
        return ""
    try:
        return str(Path(p).expanduser())
    except Exception:
        return p


_PUSH_MAIN_RE = re.compile(r"\bgit\s+push\b[^\n\r]*(?:\bmain\b|\bmaster\b)", re.IGNORECASE)
_REPO_VISIBILITY_CACHE: dict[str, bool] = {}  # remote_url -> is_private


def _is_push_to_main(cmd: str) -> bool:
    return bool(_PUSH_MAIN_RE.search(cmd))


def _is_repo_private() -> bool:
    """Check if current repo is private. Returns True (allow push) if private or unknown."""
    try:
        remote = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=5,
        ).stdout.strip()
    except Exception:
        return True  # Can't determine — allow

    if remote in _REPO_VISIBILITY_CACHE:
        return _REPO_VISIBILITY_CACHE[remote]

    # Check cache file to avoid repeated gh calls
    cache_file = Path.home() / ".claude" / "hooks" / ".repo_visibility_cache.json"
    try:
        cache = json.loads(cache_file.read_text()) if cache_file.exists() else {}
    except Exception:
        cache = {}

    if remote in cache:
        _REPO_VISIBILITY_CACHE[remote] = cache[remote]
        return cache[remote]

    # Query GitHub API via gh
    try:
        result = subprocess.run(
            ["gh", "repo", "view", "--json", "isPrivate", "-q", ".isPrivate"],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode == 0:
            is_private = result.stdout.strip().lower() == "true"
            _REPO_VISIBILITY_CACHE[remote] = is_private
            cache[remote] = is_private
            cache_file.parent.mkdir(parents=True, exist_ok=True)
            cache_file.write_text(json.dumps(cache, indent=2) + "\n")
            return is_private
    except Exception:
        pass

    return True  # Default: allow (assume private if can't check)


def _is_rm_rf(cmd: str) -> bool:
    try:
        argv = shlex.split(cmd)
    except Exception:
        return False
    if not argv or argv[0] != "rm":
        return False

    recursive = False
    force = False
    for token in argv[1:]:
        if token == "--":
            break
        if token.startswith("--"):
            if token == "--recursive":
                recursive = True
            elif token == "--force":
                force = True
            continue
        if not token.startswith("-"):
            continue
        flags = token[1:]
        if "r" in flags or "R" in flags:
            recursive = True
        if "f" in flags:
            force = True

    return recursive and force


def _is_mutating_local_setup(cmd: str) -> bool:
    return any(rx.search(cmd) for rx in MUTATING_LOCAL_PATTERNS)


def _parse_ssh_host(cmd: str) -> str:
    try:
        argv = shlex.split(cmd)
    except Exception:
        return ""
    if not argv or argv[0] != "ssh":
        return ""
    for tok in argv[1:]:
        if tok.startswith("-"):
            continue
        host = tok.split("@", 1)[-1]
        if hos
  • hooks/self_memory_system.pyGitHub
  • hooks/session_orient.shGitHub
  • hooks/session_start.pyGitHub
  • hooks/stop_quality_gate.pyGitHub
  • hooks/stop_quality_gate.shGitHub
  • hooks/test_stop_hook.pyGitHub
  • hooks/thinking_nudge.pyGitHub
  • hooks/write_validate.pyGitHub

All 14 scripts are listed above. The source is inlined for 6 of them, starting with whatever hooks.json actually runs. See all of them in the repo.

Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.

Ships withlacp

Local policy, evidence, and recovery controls for coding agents. LACP wraps Claude, Codex, Hermes, and other CLI agents with deterministic routing, approval gates, execution records, memory controls, and rollback paths.

Get the whole plugin
Stats
PythonBashGitGitHubPlaywright
+3 tags
AnthropicMCPObsidian
296
Stars
0
Forks
Active
Maintenance
Shell
Language
MIT
License
7d ago
Last commit
5mo ago
Created

Repo: 0xNyk/lacp

In this plugin
3 skills1 command
See everything inside