#!/usr/bin/env python3 """discover_hook.py — the discover harness's tool-loop enforcement. Registered in .claude/settings.json as: discover_hook.py pre | post | subagent-stop Reads the hook event JSON on stdin. Active only while .discover/run.json exists above the cwd. Exit 2 with a message on stderr blocks a PreToolUse call, or returns the message to the model on PostToolUse and SubagentStop. Every block and every feedback is appended to .discover/telemetry.jsonl. overlay no screenshot, zoom or GIF on a tab whose DOM changed since its last overlay attestation login no typing or form input on a page whose URL looks like sign-in, sign-up or auth seal no reads of .pii-quarantine/, _improve/.sealed/, critic answer keys, or (prior_policy sealed) prior corpora lint a written surface record or lane summary is linted; findings come back to the writer dispatch every Agent dispatch is logged for lane_status.py lane a lane agent may not stop before its _summary.md carries this run's run_id """ import json, os, re, subprocess, sys, time CHROME = "mcp__claude-in-chrome__" SHOT_ACTIONS = {"screenshot", "zoom"} NEUTRAL_ACTIONS = {"screenshot", "zoom", "wait"} LOGIN_URL = re.compile(r"(log-?in|sign-?in|sign-?up|/auth|oauth|/sso|password|/verify|accounts\.)", re.I) LANES = {"structure": "structure", "dimensional-analyzer": "dimensions", "schema": "schema", "exterior": "stack"} def root_of(cwd): p = os.path.abspath(cwd or os.getcwd()) while p != os.path.dirname(p): if os.path.exists(os.path.join(p, ".discover", "run.json")): return p p = os.path.dirname(p) return None def load(path, default): try: return json.load(open(path)) except (OSError, ValueError): return default def tele(root, rule, detail): try: with open(os.path.join(root, ".discover", "telemetry.jsonl"), "a") as f: f.write(json.dumps({"rule": "hook." + rule, "detail": detail[:300], "ts": time.strftime("%Y-%m-%dT%H:%M:%S")}) + "\n") except OSError: pass def block(root, rule, msg): tele(root, rule, msg) sys.stderr.write(f"[discover:{rule}] {msg}\n") sys.exit(2) def tabs_path(root): return os.path.join(root, ".discover", "state", "tabs.json") def sealed(root, cfg): s = [".pii-quarantine", "_improve/.sealed", ".discover-critic/key.json"] if cfg.get("prior_policy") == "sealed": s += cfg.get("prior_paths", []) return s def touches(text, needles): return next((n for n in needles if n and n in text), None) def pre(ev, root, cfg): name, ti = ev.get("tool_name", ""), ev.get("tool_input", {}) or {} # seal if name in ("Read", "Grep", "Glob", "Bash", "NotebookRead"): text = json.dumps(ti) hit = touches(text, sealed(root, cfg)) if hit and not (name == "Bash" and re.search(r"(checks|score|sweep|freeze_pack|stage)\.py", ti.get("command", ""))): block(root, "seal", f"'{hit}' is sealed for this run; scripts read it, agents do not") if not name.startswith(CHROME): return tool = name[len(CHROME):] tabs = load(tabs_path(root), {}) calls = [(tool, ti)] if tool != "browser_batch" else [(a.get("name", a.get("tool", "")).replace(CHROME, ""), a.get("input", a.get("arguments", a))) for a in (ti.get("actions") or ti.get("calls") or [])] fresh = {k: v.get("overlay", False) for k, v in tabs.items()} for t, inp in calls: tab = str(inp.get("tabId", inp.get("tab_id", ""))) url = tabs.get(tab, {}).get("url", "") if t == "javascript_tool" and re.search(r"DISCOVER_OVERLAY_OK|__dsc\.capture\(", str(inp.get("text", inp.get("code", "")))): fresh[tab] = True continue if t in ("form_input",) or (t == "computer" and inp.get("action") == "type"): if LOGIN_URL.search(url): block(root, "login", f"tab {tab} is on a sign-in or auth page ({url}); the user logs in, never the harness") if t == "computer": act = inp.get("action", "") if act in SHOT_ACTIONS and inp.get("save_to_disk", True) is not False: if not fresh.get(tab): block(root, "overlay", f"tab {tab} changed since its last overlay; run window.__dsc.capture() (surface-census scripts/capture.js) on it, then take the screenshot") elif act not in NEUTRAL_ACTIONS: fresh[tab] = False elif t in ("gif_creator", "upload_image"): if not fresh.get(tab): block(root, "overlay", f"tab {tab} changed since its last overlay; run window.__dsc.capture() before recording") elif t in ("navigate", "tabs_create_mcp", "form_input", "file_upload"): fresh[tab] = False def post(ev, root, cfg): name, ti = ev.get("tool_name", ""), ev.get("tool_input", {}) or {} resp = ev.get("tool_response") rtext = json.dumps(resp) if not isinstance(resp, str) else resp if name.startswith(CHROME): tool = name[len(CHROME):] tabs = load(tabs_path(root), {}) os.makedirs(os.path.dirname(tabs_path(root)), exist_ok=True) tab = str(ti.get("tabId", "")) st = tabs.setdefault(tab, {}) if tool == "navigate": st.update(overlay=False, url=ti.get("url", st.get("url", ""))) elif tool == "tabs_create_mcp": m = re.search(r'"?tabId"?\s*[:=]\s*(\d+)', rtext) if m: tabs[m.group(1)] = {"overlay": False, "url": ti.get("url", "")} elif tool == "javascript_tool": m = re.search(r"DISCOVER_OVERLAY_OK[^\"']*?url=(\S+)", rtext) if m: st.update(overlay=True, path=m.group(1).strip("\\\"'")) elif tool == "computer" and ti.get("action") not in NEUTRAL_ACTIONS: st["overlay"] = False elif tool in ("form_input", "file_upload"): st["overlay"] = False elif tool == "browser_batch": for k in tabs: tabs[k]["overlay"] = tabs[k].get("overlay", False) and "DISCOVER_OVERLAY_OK" in rtext[-4000:] m = re.search(r"https?://[^\s\"'\\]+", rtext) if tool in ("tabs_context_mcp", "navigate") else None if m and tool == "navigate": st["url"] = m.group(0) json.dump(tabs, open(tabs_path(root), "w")) return if name in ("Agent", "Task"): agent = ti.get("subagent_type", "") with open(os.path.join(root, ".discover", "dispatch.jsonl"), "a") as f: f.write(json.dumps({"agent": agent, "run_id": cfg.get("run_id"), "model": ti.get("model"), "t": time.time()}) + "\n") return if name in ("Write", "Edit", "MultiEdit"): path = ti.get("file_path", "") checks = os.path.join(root, ".claude", "skills", "evidence", "scripts", "checks.py") cmd = None if re.search(r"/model/surfaces/\d[^/]*\.md$", path): cmd = "lint-record" elif re.search(r"/evidence/[^/]+/_summary\.md$", path): cmd = "lint-summary" if cmd and os.path.exists(checks): r = subprocess.run([sys.executable, checks, cmd, path], capture_output=True, text=True, timeout=60) errs = [l for l in r.stdout.splitlines() if l.startswith("[error]")] if errs: tele(root, "lint", f"{os.path.basename(path)}: {len(errs)} error(s)") sys.stderr.write("[discover:lint] " + os.path.basename(path) + "\n" + "\n".join(errs[:12]) + "\nFix these now, or record each as a defect with its reason.\n") sys.exit(2) def subagent_stop(ev, root, cfg): agent = ev.get("agent_type") or ev.get("subagent_type") or "" if agent not in LANES or ev.get("stop_hook_active"): return summ = os.path.join(cfg.get("run_dir", ""), "evidence", LANES[agent], "_summary.md") if agent == "dimensional-analyzer" and not os.path.exists(os.path.dirname(summ)): # a corpus from before the rename summ = os.path.join(cfg.get("run_dir", ""), "evidence", "claims", "_summary.md") text = open(summ).read() if os.path.exists(summ) else "" if not re.search(r"^run_id:\s*" + re.escape(str(cfg.get("run_id"))) + r"\s*$", text, re.M): block(root, "lane", f"write {summ} with run_id: {cfg.get('run_id')} and an honest status before stopping") def main(): try: ev = json.load(sys.stdin) except ValueError: return 0 aid = ev.get("agent_id") if aid and os.path.exists(os.path.join(os.environ.get("DISCOVER_IMPROVE_SANDBOX", "/tmp/discover-improve"), "bindings", f"{aid}.json")): return 0 # an improvement agent bound to its box: its own guard runs this box's hooks there root = root_of(ev.get("cwd")) if not root: return 0 cfg = load(os.path.join(root, ".discover", "run.json"), {}) if cfg.get("closed"): return 0 mode = sys.argv[1] if len(sys.argv) > 1 else "" {"pre": pre, "post": post, "subagent-stop": subagent_stop}.get(mode, lambda *a: None)(ev, root, cfg) return 0 if __name__ == "__main__": sys.exit(main())