Agent Orchestration
Hook
Hooks
What claude-command-center runs automatically, and when. A hook is a command Claude Code fires at a fixed moment, without you asking for it.
Install
> /plugin marketplace add amirfish1/claude-command-center > /plugin install ccc-dashboard@claude-command-center
Ships with claude-command-center. Installing the plugin gets these hooks.
Where it lives
- hooks/_notify.pyGitHub
Read the script
"""Tiny stdlib helper for firing macOS notifications from CCC hooks. Used by stop.py (Claude finished a turn — needs input) and notification.py (Claude is asking for permission). Stays a separate file so the hooks themselves remain trivial. Disabled gracefully when: - CCC_NOTIFY=0 in the environment (opt-out) - osascript isn't on PATH (non-macOS) - the user's display is locked / the call fails — Popen is fire-and-forget so a hook never blocks on notification delivery The Stop hook fires on every turn a session takes, not just when it goes truly idle — a session doing many quick turns in a row (e.g. an automated worker draining a queue) would otherwise bang out a banner+sound per turn. notify() is debounced per (session, title) pair via CCC_NOTIFY_COOLDOWN_S (default 30s) so bursts collapse to one sound. """ import json import os import shutil import subprocess import time import urllib.request LIVE_STATE_DIR = os.path.expanduser("~/.claude/command-center/live-state") def _enabled(): return os.environ.get("CCC_NOTIFY", "1") != "0" def _cooldown_s(): try: return float(os.environ.get("CCC_NOTIFY_COOLDOWN_S", "30")) except ValueError: return 30.0 def _debounced(session_id, title): """True if a notification with this (session, title) fired too recently.""" cooldown = _cooldown_s() if cooldown <= 0 or not session_id: return False try: os.makedirs(LIVE_STATE_DIR, exist_ok=True) key = "".join(c if c.isalnum() else "_" for c in f"{session_id}_{title}") marker = os.path.join(LIVE_STATE_DIR, f"_notify_{key}.ts") now = time.time() if os.path.exists(marker) and (now - os.path.getmtime(marker)) < cooldown: return True with open(marker, "w") as f: f.write(str(now)) except OSError: return False return False _TEMP_CWD_PREFIXES = ("/private/var/folders/", "/var/folders/", "/tmp/", "/private/tmp/") def is_throwaway_cwd(cwd): """True for sessions started in an OS temp dir. Eval harnesses and other scripts run hundreds of headless `claude -p` runs there; none wait on a human, so a banner per run is noise (one flood hit ~100 banners/minute).""" cwd = (cwd or "").rstrip("/") + "/" return cwd.startswith(_TEMP_CWD_PREFIXES) def _esc(s): """Escape characters that would break out of an AppleScript string literal. Keep it simple: backslash, double-quote, and trim length so a 5KB tool error doesn't end up in the notification banner.""" s = s or "" return s.replace("\\", "\\\\").replace('"', '\\"')[:240] def notify(title, message, subtitle="", session_id=""): if not _enabled(): return if _debounced(session_id or subtitle, title): return osascript = shutil.which("osascript") if not osascript: return script_parts = [f'display notification "{_esc(message)}"', f'with title "{_esc(title)}"'] if subtitle: script_parts.append(f'subtitle "{_esc(subtitle)}"') script = " ".join(script_parts) try: subprocess.Popen( [osascript, "-e", script], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) except OSError: pass PORT_FILE = os.path.expanduser("~/.claude/command-center/port.txt") def notify_session(kind, title, message, session_id): """Route a session banner through the CCC server (POST /api/notify). The server shows it in the open dashboard/app, where clicking it lands on the session (/?session=<id>), and falls back to an osascript banner only when no dashboard is listening. Server down or unreachable: the old direct osascript banner, so a turn end is never silent.""" if not _enabled() or not session_id: return try: with open(PORT_FILE) as f: base = f.read().strip() if not base: raise OSError("no port") req = urllib.request.Request( f"{base}/api/notify", data=json.dumps({ "kind": kind, "title": title, "body": message, "url": f"/?session={session_id}", "session_id": session_id, "os": "auto", }).encode(), headers={"Content-Type": "application/json"}, method="POST", ) urllib.request.urlopen(req, timeout=1.5).read() except Exception: notify("Claude Command Center", message, subtitle=title, session_id=session_id) - hooks/_reorient_shared.pyGitHub
Read the script
#!/usr/bin/env python3 """Shared re-orientation-after-compaction logic (MEMO-FIX-16). Both `post-compact.py` (Claude Code's PostCompact hook) and `post-compact-codex.py` (Codex's PostCompact hook) print the same style <=600-char block after their engine compacts a session: the WatchTower ticket the session claimed (if any), the last few things actually asked for, and a pointer to `ccc recall` / `ccc shipped`. This module holds the engine-agnostic half of that logic (bounded transcript reads, the "is this a real user ask" filter, block formatting); each engine script supplies its own transcript-record shape. """ import json import os import re # Matches WT-48 as well as multi-segment refs like MEMO-FIX-10. TICKET_REF_RE = re.compile(r"\b([A-Z][A-Z0-9]{0,20}(?:-[A-Z][A-Z0-9]{0,20}){0,3}-\d{1,8})\b") # Bounds the I/O regardless of total transcript size: HEAD catches the # session's opening dispatch message (where a WatchTower ref usually first # appears), TAIL catches the most recent asks and the freshest wt JSON result. HEAD_BYTES = 320_000 TAIL_BYTES = 400_000 MAX_BLOCK_CHARS = 600 # User-role turns that were injected by tooling, not typed by the user: queue # notifications, peer-session messages, background-task events. Engine- # agnostic — CCC formats these the same way regardless of which engine the # session is running. INJECTED_PREFIXES = ( "[watchtower]", "Another Claude session sent a message", "[SYSTEM NOTIFICATION", ) # MEMO-FIX-lineage: "Continue in a new session" / usage-limit auto-resume # stamps this line into the successor's own first user turn (see # usage_limit.py's _usage_limit_retrieval_prompt and ccc_server/ship_graph.py's # CONTINUATION_ORIGIN_RE, which this mirrors). Since it always lands within # HEAD_BYTES of a continuation's transcript, the same bounded head scan that # already looks for a ticket ref picks it up for free. CONTINUATION_ORIGIN_RE = re.compile(r"Origin session id: ([A-Za-z0-9][A-Za-z0-9_.-]{7,127})") def read_chunk(path, size, from_end): with open(path, "rb") as f: if from_end: total = os.fstat(f.fileno()).st_size if total > size: f.seek(total - size) f.readline() # drop the partial line the seek landed inside data = f.read() else: data = f.read(size) return data.decode("utf-8", errors="replace") def records(text): for line in text.splitlines(): line = line.strip() if not line: continue try: yield json.loads(line) except Exception: continue def consider_ask(text, state): """Record one candidate user-ask string into `state["asks"]`, skipping empty/tag-wrapped/injected text, and update `state["ticket_ref"]` if the text names a WatchTower ticket. Also updates `state["continued_from"]` when the text carries an "Origin session id:" marker -- that marker is synthetic (auto-resume tooling wrote it, not the user), so it's excluded from `asks` the same way an injected queue notification would be.""" t = re.sub(r"\s+", " ", text or "").strip() if not t: return m_origin = CONTINUATION_ORIGIN_RE.search(t) if m_origin: state["continued_from"] = m_origin.group(1).strip() return if t.startswith("<") or t.startswith(INJECTED_PREFIXES): return state["asks"].append(t) m = TICKET_REF_RE.search(t) if m: state["ticket_ref"] = m.group(1) def scan_transcript(path, scan_into): """Bounded head+tail scan of `path`, calling `scan_into(text, state)` on each chunk in file order (so tail matches overwrite head matches). Returns the resulting state dict.""" state = {"asks": [], "ticket_ref": "", "ticket_title": "", "continued_from": ""} size = os.path.getsize(path) if size <= HEAD_BYTES + TAIL_BYTES: # Small enough to read once — a separate head+tail read would cover # the same bytes twice. scan_into(read_chunk(path, size, from_end=False), state) else: # size > HEAD_BYTES + TAIL_BYTES here, so the two windows never # overlap. Head first (opening dispatch message, usually where a # WatchTower ref first appears), then tail (recent asks, freshest wt # result) — tail matches overwrite head matches. scan_into(read_chunk(path, HEAD_BYTES, from_end=False), state) scan_into(read_chunk(path, TAIL_BYTES, from_end=True), state) return state def truncate(text, limit): text = text.strip() if len(text) <= limit: return text return text[: limit - 1].rstrip() + "…" def build_block(asks, ticket_ref, ticket_title, continued_from=""): lines = ["Re-orientation after compaction:"] if continued_from: lines.append(f"Continued from: {continued_from}") if ticket_ref: head = ticket_ref if ticket_title: head += f" — {truncate(ticket_title, 60)}" lines.append(f"Ticket: {head}") if asks: lines.append("Last asks:") for a in asks: lines.append(f"- {truncate(a, 90)}") lines.append("Tools: `ccc recall <query>` / `ccc shipped <topic>`.") block = "\n".join(lines) if len(block) > MAX_BLOCK_CHARS: block = block[: MAX_BLOCK_CHARS - 1].rstrip() + "…" return block - hooks/notification.pyGitHub
Read the script
#!/usr/bin/env python3 """Notification hook — writes a needs-approval marker when Claude Code is asking the user for permission (or otherwise needs attention). The previous Needs-Approval signal in the dashboard was heuristic: a combination of `pending_tool` + assistant-as-last-event + age threshold, which routinely confused "tool fired but not yet returned" with "Claude is blocked on a permission prompt". Claude Code emits a distinct `Notification` hook event when it specifically wants the user to look, so wiring that up gives a precise, signal-driven badge instead. Pairs with post-tool-use.py, which clears the marker once a tool returns (the user has either approved or the prompt is no longer relevant). Stdlib only, fails silently like the other hooks. """ import json import os import sys import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) try: from _notify import notify_session, is_throwaway_cwd except ImportError: def notify_session(*_a, **_k): pass def is_throwaway_cwd(_cwd): return False LIVE_STATE_DIR = os.path.expanduser("~/.claude/command-center/live-state") def main(): try: raw = sys.stdin.read() data = json.loads(raw) session_id = data.get("session_id", "") if not session_id: return os.makedirs(LIVE_STATE_DIR, exist_ok=True) message = data.get("message", "") marker = { "session_id": session_id, "message": message, "type": data.get("type", ""), "started_at": time.time(), } path = os.path.join(LIVE_STATE_DIR, f"{session_id}_needs_approval.json") tmp = path + ".tmp" with open(tmp, "w") as f: json.dump(marker, f) os.replace(tmp, path) # macOS banner so the user knows Claude is blocked on a permission # prompt even when CCC isn't focused. Falls through silently on # non-macOS systems or when CCC_NOTIFY=0. if not is_throwaway_cwd(data.get("cwd")): notify_session( "needs_input", "Claude needs your approval", message or "Permission requested", session_id, ) except Exception: pass if __name__ == "__main__": main() - hooks/permission-request.pyGitHub
Read the script
#!/usr/bin/env python3 """PermissionRequest hook: lets a cloud agent answer ONE specific prompt. Claude Code runs this when a permission dialog would open. The hook: 1. exits at once (no output = normal prompt) unless the CCC cloud relay is running and asked for approvals (fresh `enabled` file); 2. hashes the exact tool call it was given (same canonical JSON as ccc_server/relay_context.canonical_input); 3. parks for a short window, watching for a decision file the relay writes after the cloud approved/denied that exact hash; 4. returns allow/deny only when the decision's hash equals its own hash, it has not expired (120 s) and it can be claimed atomically (single use); 5. stops parking as soon as the prompt is answered some other way (the transcript gets the tool_result for this call) or the session exits. Anything else (no decision, wrong hash, expired, error) prints nothing, so the normal prompt decides. Never prompts. Stdlib only. Keep the file-name scheme and canonical form in lockstep with ccc_server/approval_hook.py. """ import hashlib import json import os import re import sys import time ENABLED_FRESH_S = 90 POLL_S = 0.2 HEARTBEAT_S = 2.0 RESOLVED_CHECK_S = 1.0 TAIL_BYTES = 512 * 1024 SKIP_TOOLS = ("AskUserQuestion", "ExitPlanMode") _SID_RE = re.compile(r"[A-Za-z0-9_\-]{1,100}") def base_dir(): return os.environ.get("CCC_APPROVAL_HOOK_DIR") or os.path.join( os.path.expanduser("~"), ".claude", "command-center", "cloud-approvals") def max_wait(): try: return max(0.0, float(os.environ.get("CCC_APPROVAL_HOOK_WAIT", "900"))) except ValueError: return 900.0 def canonical_input(tool, tin): return json.dumps({"tool": str(tool or ""), "input": tin}, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def input_sha256(tool, tin): return hashlib.sha256(canonical_input(tool, tin).encode("utf-8")).hexdigest() def enabled(base, now): try: return now - os.stat(os.path.join(base, "enabled")).st_mtime <= ENABLED_FRESH_S except OSError: return False def _unlink(path): try: os.unlink(path) except OSError: pass def take_decision(path, want_sha, now): """Claim the decision file atomically. Returns "allow"/"deny" or None. A file whose hash differs from `want_sha` is left alone (it is not ours to consume); an expired or unreadable one is removed.""" try: with open(path) as f: doc = json.load(f) except FileNotFoundError: return None except (OSError, ValueError): _unlink(path) return None if not isinstance(doc, dict): _unlink(path) return None try: expires = float(doc.get("expires")) except (TypeError, ValueError): expires = 0.0 if expires <= now: _unlink(path) return None if doc.get("sha256") != want_sha or doc.get("decision") not in ("allow", "deny"): return None claim = "%s.claim.%d" % (path, os.getpid()) try: os.rename(path, claim) # atomic: exactly one claimant wins except OSError: return None _unlink(claim) return doc["decision"] def emit(decision): if decision == "allow": body = {"behavior": "allow"} else: body = {"behavior": "deny", "message": "Denied by an approved cloud agent via Claude Command Center."} print(json.dumps({"hookSpecificOutput": { "hookEventName": "PermissionRequest", "decision": body}})) class PromptWatch: """Tells when the prompt this hook parked on was answered elsewhere. Reads the session transcript incrementally. The pending call is the last tool_use with this exact name+input that has no tool_result yet; once a tool_result for its id is appended, the prompt is resolved. No transcript or no match means "never resolved" (the deadline still bounds the wait).""" def __init__(self, path, tool, tin): self.path, self.tool, self.tin = path, tool, tin self.pending = None self.rest = b"" self.ppid = os.getppid() try: size = os.path.getsize(path) if path else 0 except OSError: size = 0 self.start = size self.offset = max(0, size - TAIL_BYTES) self.skip_first = self.offset > 0 self._scan(initial=True) def _scan(self, initial=False): if not self.path: return False try: with open(self.path, "rb") as f: f.seek(self.offset) chunk = f.read() except OSError: return False self.offset += len(chunk) lines = (self.rest + chunk).split(b"\n") self.rest = lines.pop() if self.skip_first and lines: lines.pop(0) # partial line at the tail cut self.skip_first = False for raw in lines: if self._line(raw) and not initial: return True return False def _line(self, raw): try: content = (json.loads(raw).get("message") or {}).get("content") except (ValueError, AttributeError): return False if not isinstance(content, list): return False for b in content: if not isinstance(b, dict): continue if b.get("type") == "tool_use" and b.get("name") == self.tool \ and b.get("input") == self.tin: self.pending = b.get("id") elif b.get("type") == "tool_result" and self.pending \ and b.get("tool_use_id") == self.pending: self.pending = None # an earlier identical call; or ours return True return False def resolved(self): if os.getppid() != self.ppid: return True # the session exited return self._scan() def - hooks/post-compact-codex.pyGitHub
Read the script
#!/usr/bin/env python3 """Codex PostCompact hook — the Codex equivalent of post-compact.py (MEMO-FIX-16). Codex's own hook subsystem (codex-rs `hooks` crate, `protocol::HookEventName`) defines a native `PostCompact` event, matching Claude Code's hook of the same name field-for-field closely enough to share the same JSON contract: stdin is a JSON object with (at least) `session_id` and a nullable `transcript_path` pointing at the rollout `.jsonl` Codex just finished compacting. This is a direct native trigger, not a fallback — no SessionStart/"resume source" heuristic or manual rollout-marker scan was needed, because PostCompact already exists and fires exactly where Claude's does: right after a compaction completes, before the next turn. One thing PostCompact hooks do NOT need is `pre-compact.py`'s "compacting…" marker file — that convention exists purely for Claude Code's live dashboard badge. CCC tracks Codex's own compaction lifecycle separately (see `compaction_recovery` state in ccc_server/codex.py), so there is nothing for this hook to clear. Reuses hooks/_reorient_shared.py for the bounded transcript read, the "is this a real ask" filter, and the <=600-char block formatting — only the Codex rollout record shapes below are new. Local only, no subprocess, no LLM call, must never fail the turn. """ import json import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from _reorient_shared import ( # noqa: E402 MAX_BLOCK_CHARS, consider_ask, scan_transcript, build_block, records as _reorient_records, ) def _user_message_text(payload): """Classic Codex rollout shape: event_msg/user_message with a flat `message` string.""" return str(payload.get("message") or "") def _item_completed_user_text(payload): """Newer (multi-agent-capable) Codex rollout shape: event_msg/item_completed wrapping a `UserMessage` item with a list of {"type": "text", "text": ...} content blocks.""" item = payload.get("item") if isinstance(payload.get("item"), dict) else {} if item.get("type") != "UserMessage": return None parts = [ c.get("text", "") for c in (item.get("content") or []) if isinstance(c, dict) and c.get("type") == "text" and isinstance(c.get("text"), str) ] return "\n\n".join(p.strip() for p in parts if p.strip()).strip() def _tool_output_texts(payload): """function_call_output / custom_tool_call_output `output` field: a plain string, or a list of strings/{"text": ...} blocks (screenshots etc. are skipped — only text parts can carry a wt ticket ref/title JSON).""" output = payload.get("output") if isinstance(output, list): parts = [] for item in output: if isinstance(item, str): parts.append(item) elif isinstance(item, dict) and isinstance(item.get("text"), str): parts.append(item["text"]) yield "\n".join(parts) elif isinstance(output, str): yield output def _scan_into(text, state): """Codex equivalent of the Claude hook's `_scan_into`: update `state` (asks list, ticket_ref, ticket_title) from one chunk of rollout JSONL text, in file order.""" for rec in _reorient_records(text): rec_type = rec.get("type") payload = rec.get("payload") if isinstance(rec.get("payload"), dict) else {} ptype = payload.get("type") if rec_type == "event_msg" and ptype == "user_message": consider_ask(_user_message_text(payload), state) elif rec_type == "event_msg" and ptype == "item_completed": text_ = _item_completed_user_text(payload) if text_ is not None: consider_ask(text_, state) elif rec_type == "response_item" and ptype in ( "function_call_output", "custom_tool_call_output", ): for result_text in _tool_output_texts(payload): try: obj = json.loads(result_text) except Exception: continue if isinstance(obj, dict) and obj.get("ref") and obj.get("title"): state["ticket_ref"] = str(obj["ref"]) state["ticket_title"] = str(obj["title"]) def main(): try: raw = sys.stdin.read() data = json.loads(raw) transcript_path = data.get("transcript_path") or "" if not transcript_path or not os.path.isfile(transcript_path): return state = scan_transcript(transcript_path, _scan_into) asks = state["asks"][-3:] ticket_ref = state["ticket_ref"] ticket_title = state["ticket_title"] continued_from = state["continued_from"] if not asks and not ticket_ref and not continued_from: return print(build_block(asks, ticket_ref, ticket_title, continued_from)) except Exception: pass if __name__ == "__main__": main() - hooks/post-compact.pyGitHub
Read the script
#!/usr/bin/env python3 """PostCompact hook — clears the compacting marker pre-compact.py wrote, then prints a short re-orientation block. Compaction throws away the transcript that carried a session's implicit context. This gives a quick anchor back: the WatchTower ticket the session claimed (if any), the last few things actually asked for, and a pointer to `ccc recall` / `ccc shipped` for anything deeper. Local only, no subprocess, no LLM call — everything comes from re-reading the transcript Claude Code already wrote to disk. Bounded to a head+tail read of the transcript (not the whole file) so this stays well under Claude Code's hook timeout even for a huge pre-compaction transcript. Every step is wrapped so a malformed transcript, a missing file, or an unexpected payload shape just skips that piece of the block — hooks must never fail the turn. """ import json import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from _reorient_shared import ( # noqa: E402 TICKET_REF_RE, HEAD_BYTES, TAIL_BYTES, MAX_BLOCK_CHARS, INJECTED_PREFIXES, read_chunk as _read_chunk, records as _records, consider_ask, scan_transcript, truncate as _truncate, build_block as _build_block, ) LIVE_STATE_DIR = os.path.expanduser("~/.claude/command-center/live-state") def _clear_compacting_marker(session_id): path = os.path.join(LIVE_STATE_DIR, f"{session_id}_compacting.json") try: os.unlink(path) except FileNotFoundError: pass def _text_blocks(content): if isinstance(content, str): return [content] if isinstance(content, list): return [c.get("text", "") for c in content if isinstance(c, dict) and c.get("type") == "text"] return [] def _has_tool_result(content): return isinstance(content, list) and any( isinstance(c, dict) and c.get("type") == "tool_result" for c in content ) def _tool_result_texts(content): if not isinstance(content, list): return for c in content: if not isinstance(c, dict) or c.get("type") != "tool_result": continue cc = c.get("content") if isinstance(cc, str): yield cc elif isinstance(cc, list): for b in cc: if isinstance(b, dict) and b.get("type") == "text": yield b.get("text", "") def _scan_into(text, state): """Update `state` (asks list, ticket_ref, ticket_title) from one chunk of transcript text, in file order. Called on the head chunk then the tail chunk, so a later match — closer to "now" — always overwrites an earlier one; asks accumulate and get trimmed to the last 3 at the end.""" for rec in _records(text): if rec.get("type") != "user": continue msg = rec.get("message", {}) if msg.get("role") != "user": continue content = msg.get("content") if not _has_tool_result(content): for t in _text_blocks(content): consider_ask(t, state) for result_text in _tool_result_texts(content): try: obj = json.loads(result_text) except Exception: continue if isinstance(obj, dict) and obj.get("ref") and obj.get("title"): state["ticket_ref"] = str(obj["ref"]) state["ticket_title"] = str(obj["title"]) def main(): try: raw = sys.stdin.read() data = json.loads(raw) session_id = data.get("session_id", "") if session_id: _clear_compacting_marker(session_id) transcript_path = data.get("transcript_path", "") if not transcript_path or not os.path.isfile(transcript_path): return state = scan_transcript(transcript_path, _scan_into) asks = state["asks"][-3:] ticket_ref = state["ticket_ref"] ticket_title = state["ticket_title"] continued_from = state["continued_from"] if not asks and not ticket_ref and not continued_from: return print(_build_block(asks, ticket_ref, ticket_title, continued_from)) except Exception: pass if __name__ == "__main__": main() - hooks/post-tool-use.pyGitHub
- hooks/pre-compact.pyGitHub
- hooks/pre-tool-use.pyGitHub
- hooks/session-start.pyGitHub
- hooks/shared-clone-guard.pyGitHub
- hooks/stop.pyGitHub
All 12 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 withclaude-command-center
Your coding agents outgrew your terminal. CCC puts every session on one local board and tells you which one needs you. Not affiliated with Anthropic. Start the next while Claude builds the first. A real share card from one of the maintainer's machines.
Get the whole plugin
Stats
198
Stars
26
Forks
Active
Maintenance
Python
Language
4h ago
Last commit
6mo ago
Created
5h ago
Added
In this plugin
See everything inside