plan-annotate.py - Claude Code hook for interactive plan review via kitty overlay editor

plan-annotate.py

#!/usr/bin/env python3
"""plan-annotate.py - PreToolUse hook for ExitPlanMode.

interactive plan review hook that lets you annotate Claude's plans directly
in your editor before approving them. when Claude calls ExitPlanMode, this
hook intercepts the call, opens the plan in $EDITOR via a kitty terminal
overlay, and waits for you to review/edit. if you make changes, the hook
computes a unified diff and sends it back to Claude as a denial reason,
forcing Claude to revise the plan based on your annotations. if you make
no changes, the normal approval dialog appears.

this creates a feedback loop: annotate → Claude revises → annotate again →
until you're satisfied and close the editor without changes.

annotation style - edit the plan text directly in your editor:
  - add new lines to request additions (e.g., "add error handling here")
  - delete lines to request removal
  - modify lines to request changes (e.g., change "use polling" to "use websockets")
  - add inline comments after existing text (e.g., "- [ ] create handler - use JWT not sessions")
any text change works - the hook diffs original vs edited and Claude sees
exactly what you added, removed, or modified.

hook integration (settings.json):

    "hooks": {
      "PreToolUse": [{
        "matcher": "ExitPlanMode",
        "hooks": [{
          "type": "command",
          "command": "~/.claude/scripts/plan-annotate.py",
          "timeout": 345600
        }]
      }]
    }

hook receives JSON on stdin with the plan content in tool_input.plan field.
returns PreToolUse hook JSON response with permissionDecision:
  - "ask"  → no changes made, proceed to normal confirmation
  - "deny" → changes detected, unified diff sent as denial reason

requirements:
  - kitty terminal with remote control enabled (allow_remote_control yes)
  - $EDITOR set (defaults to micro)

limitations:
  - requires kitty terminal - without it, falls back to "ask" (no annotation)
  - does not work in non-kitty terminals (iTerm2, Terminal.app, etc.)
  - does not work inside tmux (kitty overlay needs direct kitty access)
  - the hook blocks until the editor closes; timeout should be set high
  - plan content comes from Claude's ExitPlanMode call, not from the plan
    file on disk - if you edit the file on disk separately, those changes
    won't be seen by this hook

install:
    mkdir -p ~/.claude/scripts
    cp plan-annotate.py ~/.claude/scripts/plan-annotate.py
    chmod +x ~/.claude/scripts/plan-annotate.py

then add the hook to ~/.claude/settings.json (create if missing).
if the file already has a "hooks" section, merge the PreToolUse entry
into the existing array. if not, add the full block shown above.

after saving settings.json, restart Claude Code for hooks to take effect.

usage:
    plan-annotate.py [--test]
"""

import difflib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path


def read_plan_from_stdin() -> str:
    """read plan content from hook event JSON on stdin."""
    raw = sys.stdin.read()
    if not raw.strip():
        return ""
    try:
        event = json.loads(raw)
        return event.get("tool_input", {}).get("plan", "")
    except json.JSONDecodeError:
        return ""


def make_response(decision: str, reason: str = "") -> str:
    """build PreToolUse hook JSON response."""
    resp: dict = {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": decision,
        }
    }
    if reason:
        resp["hookSpecificOutput"]["permissionDecisionReason"] = reason
    return json.dumps(resp, indent=2)


def get_diff(original: str, edited: str) -> str:
    """get unified diff between original and edited content."""
    orig_lines = original.splitlines(keepends=True)
    edit_lines = edited.splitlines(keepends=True)
    diff = difflib.unified_diff(orig_lines, edit_lines, fromfile="original", tofile="annotated", n=2)
    return "".join(diff)


def open_editor(filepath: Path) -> int:
    """open file in $EDITOR via kitty overlay, blocking until editor closes.
    returns non-zero if kitty is not available (no terminal fallback)."""
    editor = os.environ.get("EDITOR", "micro")

    if not shutil.which("kitty"):
        return 1

    # use a sentinel file to detect when editor closes
    sentinel = Path(tempfile.mktemp(prefix="plan-done-"))
    wrapper = f'{editor} {filepath}; touch {sentinel}'
    subprocess.run(
        ["kitty", "@", "launch", "--type=overlay",
         f"--title=Plan Review: {filepath.name}",
         "sh", "-c", wrapper],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    while not sentinel.exists():
        time.sleep(0.3)
    sentinel.unlink(missing_ok=True)
    return 0


def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description="plan annotation hook for ExitPlanMode")
    parser.add_argument("--test", action="store_true", help="run unit tests")
    args = parser.parse_args()

    if args.test:
        run_tests()
        return

    plan_content = read_plan_from_stdin()
    if not plan_content:
        print(make_response("ask", "no plan content in hook event"))
        return

    # write plan to temp file for editing
    with tempfile.NamedTemporaryFile(mode="w", suffix=".md", prefix="plan-review-", delete=False) as tmp:
        tmp.write(plan_content)
        tmp_path = Path(tmp.name)

    try:
        if open_editor(tmp_path) != 0:
            print(make_response("ask", "kitty not available, skipping plan annotation"))
            return

        edited_content = tmp_path.read_text()
        diff = get_diff(plan_content, edited_content)

        if not diff:
            print(make_response("ask", "plan reviewed, no changes"))
        else:
            feedback = (
                "user reviewed the plan in an editor and made changes. "
                "the diff below shows what the user modified (lines starting with - are original, + are user's version).\n"
                "examine each diff hunk to understand the user's feedback:\n"
                "- added lines (+) are user's annotations, comments, or requested additions\n"
                "- removed lines (-) with replacement (+) show what the user wants changed\n"
                "- removed lines (-) without replacement mean the user wants that removed\n"
                "- context lines (no prefix) show surrounding plan content for reference\n\n"
                f"{diff}\n"
                "adjust the plan to address each annotation, then call ExitPlanMode again."
            )
            print(make_response("deny", feedback))
    finally:
        tmp_path.unlink(missing_ok=True)


def run_tests() -> None:
    """run embedded unit tests."""
    import unittest

    class TestGetDiff(unittest.TestCase):
        def test_no_changes(self) -> None:
            text = "# Plan\n- task 1\n- task 2\n"
            self.assertEqual(get_diff(text, text), "")

        def test_added_line(self) -> None:
            original = "# Plan\n- task 1\n- task 2\n"
            edited = "# Plan\n- task 1\nadd timestamps\n- task 2\n"
            diff = get_diff(original, edited)
            self.assertIn("+add timestamps", diff)
            self.assertIn("task 1", diff)

        def test_removed_line(self) -> None:
            original = "# Plan\n- task 1\n- task 2\n"
            edited = "# Plan\n- task 2\n"
            diff = get_diff(original, edited)
            self.assertIn("-- task 1", diff)

        def test_modified_line(self) -> None:
            original = "# Plan\n- task 1\n"
            edited = "# Plan\n- task 1 (use JWT)\n"
            diff = get_diff(original, edited)
            self.assertIn("-- task 1", diff)
            self.assertIn("+- task 1 (use JWT)", diff)

        def test_multiple_changes(self) -> None:
            original = "# Plan\n\n## A\n- item\n\n## B\n- item\n"
            edited = "# Plan\n\n## A\n- item\nnote about A\n\n## B\n- item\nnote about B\n"
            diff = get_diff(original, edited)
            self.assertIn("+note about A", diff)
            self.assertIn("+note about B", diff)

    class TestReadPlanFromStdin(unittest.TestCase):
        def test_valid_event(self) -> None:
            import io
            event = json.dumps({"tool_input": {"plan": "# My Plan\n- task 1"}})
            old_stdin = sys.stdin
            sys.stdin = io.StringIO(event)
            try:
                self.assertEqual(read_plan_from_stdin(), "# My Plan\n- task 1")
            finally:
                sys.stdin = old_stdin

        def test_empty_stdin(self) -> None:
            import io
            old_stdin = sys.stdin
            sys.stdin = io.StringIO("")
            try:
                self.assertEqual(read_plan_from_stdin(), "")
            finally:
                sys.stdin = old_stdin

        def test_no_plan_field(self) -> None:
            import io
            old_stdin = sys.stdin
            sys.stdin = io.StringIO(json.dumps({"tool_input": {}}))
            try:
                self.assertEqual(read_plan_from_stdin(), "")
            finally:
                sys.stdin = old_stdin

        def test_invalid_json(self) -> None:
            import io
            old_stdin = sys.stdin
            sys.stdin = io.StringIO("not json")
            try:
                self.assertEqual(read_plan_from_stdin(), "")
            finally:
                sys.stdin = old_stdin

    class TestResponses(unittest.TestCase):
        def test_ask_response(self) -> None:
            result = json.loads(make_response("ask", "reviewed"))
            out = result["hookSpecificOutput"]
            self.assertEqual(out["hookEventName"], "PreToolUse")
            self.assertEqual(out["permissionDecision"], "ask")

        def test_deny_response(self) -> None:
            result = json.loads(make_response("deny", "fix this"))
            out = result["hookSpecificOutput"]
            self.assertEqual(out["permissionDecision"], "deny")
            self.assertIn("fix this", out["permissionDecisionReason"])

        def test_special_chars_in_json(self) -> None:
            result = make_response("deny", 'has "quotes" and\nnewlines')
            parsed = json.loads(result)
            self.assertIn("quotes", parsed["hookSpecificOutput"]["permissionDecisionReason"])

    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    for tc in [TestGetDiff, TestReadPlanFromStdin, TestResponses]:
        suite.addTests(loader.loadTestsFromTestCase(tc))
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(suite)
    sys.exit(0 if result.wasSuccessful() else 1)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\r\033[K", end="")
        sys.exit(130)