#!/usr/bin/env python3 """ Claude Code -> local plan feed Stop hook. POSTs Claude's last response (and the user's question that triggered it) to plan_server.rb so every connected browser tab picks it up automatically. Silent if the server isn't running -- never breaks normal Claude Code usage. Configured automatically by install.sh as a Claude Code "Stop" hook. """ import hashlib import os import sys import json import tempfile import urllib.parse import urllib.request # Override with PLAN_FEED_URL to point at your Vercel deployment, e.g. # PLAN_FEED_URL=https://your-app.vercel.app/notify FEED_URL = os.environ.get( "PLAN_FEED_URL", "https://vercel-server-jinyossatron11s-projects.vercel.app/notify", ) def site_name() -> str: """Which site this Claude posts to. Set PLAN_SITE when launching, e.g. PLAN_SITE=acme claude ; falls back to the 'default' site.""" return (os.environ.get("PLAN_SITE") or "").strip() or "default" def user_name() -> str: """Which user this Claude session belongs to. Set PLAN_USER when launching, e.g. PLAN_USER=alice claude ; blank if unset.""" return (os.environ.get("PLAN_USER") or "").strip() def room_name() -> str: """Which feed room (within the site) this Claude posts to. Each terminal is pinned to a room by setting PLAN_ROOM when launching, e.g. PLAN_ROOM=alice claude If unset, fall back to the project folder name so separate projects still land in separate rooms; finally 'general'. """ room = (os.environ.get("PLAN_ROOM") or "").strip() if room: return room try: return os.path.basename(os.getcwd()) or "general" except Exception: return "general" def last_user_message(transcript_path: str) -> str: """Return the last clean user message in the transcript. Stop hook fires before the assistant entry for the current turn is written, so anchoring to the last assistant entry always gives the previous turn. Instead, just find the last user entry that looks like real typed text. """ if not transcript_path: return "" try: with open(transcript_path, encoding="utf-8") as f: entries = [json.loads(l) for l in f if l.strip()] for obj in reversed(entries): if obj.get("type") != "user": continue content = obj.get("message", {}).get("content", "") if isinstance(content, list): parts = [ b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text" ] content = "\n".join(p for p in parts if p) text = (content or "").strip() # skip system injections: XML tags, slash commands, tool outputs if text and not text.startswith("<") and not text.startswith("/"): return text except Exception: pass return "" def site_pw() -> str: """Password of the target site, if it is locked. Set PLAN_SITE_PW.""" return (os.environ.get("PLAN_SITE_PW") or "").strip() def pending_mentions(base, site, room): """Fetch unanswered @claude comments for this room and mark them answered. Marks BEFORE returning so a crashed continuation can never loop forever; returns [] on any failure (the hook must never break Claude Code). """ try: qs = urllib.parse.urlencode({"site": site, "room": room, "pw": site_pw()}) with urllib.request.urlopen(f"{base}/mentions?{qs}", timeout=3) as r: rows = json.loads(r.read()) if not isinstance(rows, list) or not rows: return [] payload = json.dumps({"site": site, "room": room, "password": site_pw(), "ids": [m["id"] for m in rows]}).encode("utf-8") req = urllib.request.Request( f"{base}/mentions/ack", data=payload, headers={"Content-Type": "application/json"}, ) urllib.request.urlopen(req, timeout=3) return rows except Exception: return [] def reply_state_path(site, room): """Where we remember which mention the current continuation is answering, so the NEXT post can be tagged as a reply to it.""" tag = hashlib.md5(f"{site}/{room}".encode("utf-8")).hexdigest()[:12] return os.path.join(tempfile.gettempdir(), f"plan-feed-reply-{tag}.json") def save_reply_state(site, room, mentions): try: with open(reply_state_path(site, room), "w", encoding="utf-8") as f: json.dump(mentions, f, ensure_ascii=False) except Exception: pass def pop_reply_state(site, room): """Return and clear the pending reply info (or None).""" path = reply_state_path(site, room) try: with open(path, encoding="utf-8") as f: mentions = json.load(f) os.remove(path) return mentions if isinstance(mentions, list) and mentions else None except Exception: return None def mention_reason(mentions): lines = [] for m in mentions: ctx = (m.get("question") or m.get("msg") or "").strip() ctx = f' (บนข้อความเรื่อง: "{ctx[:120]}")' if ctx else "" lines.append(f"- {m.get('user') or 'someone'}: {m.get('text', '')}{ctx}") return ( "มีคอมเมนต์ @claude จากหน้า plan-feed ถึงคุณ:\n" + "\n".join(lines) + "\n\nช่วยตอบ/จัดการตามคอมเมนต์ข้างต้น " "(คำตอบของคุณจะถูกโพสต์ขึ้น feed ห้องเดิมโดยอัตโนมัติ)" ) def main(): try: data = json.load(sys.stdin) except json.JSONDecodeError: return text = (data.get("last_assistant_message") or "").strip() if not text: return transcript_path = data.get("transcript_path", "") question = last_user_message(transcript_path) site, room = site_name(), room_name() # If the previous Stop was blocked to answer @claude comments, this # response IS the answer: link it to the original card and show the # comment (not the transcript question) on it. reply_to = 0 replying = pop_reply_state(site, room) if replying: reply_to = int(replying[0].get("message_id") or 0) question = "\n".join( f"{m.get('user') or 'someone'}: {m.get('text', '')}" for m in replying ) payload = json.dumps( {"text": text, "question": question, "reply_to": reply_to, "site": site, "room": room, "user": user_name()} ).encode("utf-8") req = urllib.request.Request( FEED_URL, data=payload, headers={"Content-Type": "application/json"} ) try: urllib.request.urlopen(req, timeout=2) except Exception: pass # feed server unreachable -- ignore # Opt-in (PLAN_ANSWER_MENTIONS=1): block the Stop so Claude answers # @claude comments immediately in this session. Off by default — the # normal path is recall.py injecting comments as context on the next # prompt, which never hijacks the session you're working in. if (os.environ.get("PLAN_ANSWER_MENTIONS") or "") != "1": return base = FEED_URL.rsplit("/notify", 1)[0] mentions = pending_mentions(base, site, room) if mentions: save_reply_state(site, room, mentions) print(json.dumps({"decision": "block", "reason": mention_reason(mentions)}, ensure_ascii=False)) if __name__ == "__main__": main()