tools/hooks_check.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Verify the Claude Code state-hook wiring, end to end, without a real CLI.
Feature: a hook-speaking CLI is launched with a ``--settings`` block whose
Notification / Stop / UserPromptSubmit hooks POST the session's real state to
the attention endpoint, so the sidebar knows "waiting on you" for certain
instead of guessing from output. This checks the three joints:
1. The settings block is well-formed and only claude (not shell) opts in.
2. A launched claude session's pane really carries ``--settings`` + the
``CLIQUE_*`` env the reporter needs.
3. Running the reporter with that env drives the session's state to ``waiting``
and back to idle on ``clear`` — the authoritative signal outranking the
activity guess.
Its own home, port and tmux socket; nothing touches a panel you are using.
Real Claude accepting the settings and firing the hooks is verified separately
(it needs the CLI and a token); this covers everything up to that.
python3 tools/hooks_check.py
Exit status is 0 on pass, 1 on fail.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from clique.app import HOOK, _hooks_settings
from clique.registry import Registry
PASSWORD = "hooks-check" # noqa: S105 — a throwaway panel on loopback
PORT = 3286
BASE = f"http://127.0.0.1:{PORT}"
HOME = Path("/tmp/clique-hooks-check-home")
SOCKET = "clique-hooks-check"
def _run(cmd, **kw):
return subprocess.run(cmd, capture_output=True, text=True, **kw)
def main() -> int:
# 1. Settings block + registry opt-in (in-process, no launch).
cfg = json.loads(_hooks_settings(sys.executable))["hooks"]
settings_ok = (
{"Notification", "Stop", "UserPromptSubmit"} <= set(cfg)
# Stop is a turn boundary, not a request — it clears rather than nagging.
and "clear" in cfg["Stop"][0]["hooks"][0]["command"]
and "waiting" not in cfg["Stop"][0]["hooks"][0]["command"]
# A genuine wait still reports waiting, via the idle/permission Notification.
and "waiting" in cfg["Notification"][1]["hooks"][0]["command"]
and "clear" in cfg["UserPromptSubmit"][0]["hooks"][0]["command"]
and str(HOOK) in cfg["Notification"][0]["hooks"][0]["command"]
)
reg = Registry(ROOT / "clique" / "config" / "clis.toml").types()
optin_ok = reg["claude"].hooks is True and reg["shell"].hooks is False
print(f" {'ok ' if settings_ok else 'FAIL'} settings block well-formed")
print(f" {'ok ' if optin_ok else 'FAIL'} claude opts in, shell does not")
# 2 + 3. Real panel: launch injection, then the reporter drives state.
shutil.rmtree(HOME, ignore_errors=True)
HOME.mkdir(parents=True)
env = dict(os.environ, CLIQUE_HOME=str(HOME), CLIQUE_TMUX_SOCKET=SOCKET)
_run(
[sys.executable, "-m", "clique", "password"],
input=f"{PASSWORD}\n{PASSWORD}\n",
env=env,
cwd=str(ROOT),
)
mint = _run(
[sys.executable, "-m", "clique", "token", "create", "check"], env=env, cwd=str(ROOT)
).stdout
apitok = next((ln.strip() for ln in mint.splitlines() if ln.strip().startswith("mxp_")), "")
proc = subprocess.Popen(
[
sys.executable,
"-m",
"clique",
"--host",
"127.0.0.1",
"--port",
str(PORT),
"--state",
str(HOME / "state.json"),
],
env=env,
cwd=str(ROOT),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
def api(path, method="GET", body=None):
req = urllib.request.Request(
BASE + path,
data=(json.dumps(body).encode() if body is not None else None),
method=method,
headers={"Authorization": "Bearer " + apitok, "Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=10) as r:
return json.loads(r.read() or "{}")
launch_ok = report_ok = False
try:
for _ in range(80):
try:
urllib.request.urlopen(BASE + "/healthz", timeout=2).read()
break
except (urllib.error.URLError, OSError):
time.sleep(0.25)
cid = api("/api/sessions", "POST", {"cli": "claude", "cwd": "/tmp", "name": "hooked"})["id"]
time.sleep(1.0)
mux = "sm-" + cid.replace("-", "")[:8]
start_cmd = _run(
["tmux", "-L", SOCKET, "list-panes", "-a", "-F", "#{pane_start_command}"]
).stdout
pane_env = _run(["tmux", "-L", SOCKET, "show-environment", "-t", mux]).stdout
launch_ok = (
"--settings" in start_cmd
and "hook.py" in start_cmd
and "CLIQUE_URL=" in pane_env
and "CLIQUE_TOKEN=" in pane_env
and f"CLIQUE_SESSION={cid}" in pane_env
)
print(f" {'ok ' if launch_ok else 'FAIL'} claude pane gets --settings + CLIQUE_* env")
# A shell session (not busy long) + the reporter with its pane env.
sid = api("/api/sessions", "POST", {"cli": "shell", "cwd": "/tmp", "name": "report"})["id"]
time.sleep(1.0)
# The reporter authenticates with this session's OWN token, read from its
# pane environment (there is no shared hook.token any more).
smux = "sm-" + sid.replace("-", "")[:8]
senv = _run(["tmux", "-L", SOCKET, "show-environment", "-t", smux, "CLIQUE_TOKEN"]).stdout
htok = senv.split("=", 1)[1].strip() if "=" in senv else ""
henv = dict(os.environ, CLIQUE_SESSION=sid, CLIQUE_URL=BASE, CLIQUE_TOKEN=htok)
def state():
return api(f"/api/sessions/{sid}/wait?for=waiting,idle,error,working&timeout=1")[
"state"
]
_run([sys.executable, str(HOOK), "waiting"], input="{}", env=henv)
time.sleep(0.3)
after_wait = state()
_run([sys.executable, str(HOOK), "clear"], input="{}", env=henv)
time.sleep(0.3)
after_clear = state()
report_ok = after_wait == "waiting" and after_clear in ("idle", "working")
print(
f" {'ok ' if report_ok else 'FAIL'} reporter drives state: "
f"waiting->{after_wait}, clear->{after_clear}"
)
finally:
proc.terminate()
_run(["tmux", "-L", SOCKET, "kill-server"])
ok = settings_ok and optin_ok and launch_ok and report_ok
print("hooks_check:", "PASS" if ok else "FAIL")
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())