| """moth hooks / moth hook / moth unhook : see and change which coding agents tell moth about their turns. |
| |
| moth hooks [--project DIR] harnesses found on this machine, and whether moth is hooked into each |
| moth hook <harness ...> | --all [flags] install moth's hook / plugin (asks first; --yes skips, --dry-run shows) |
| moth unhook <harness ...> | --all [flags] remove exactly what `moth hook` adds, nothing else |
| |
| Harnesses: claude-code, codex, hermes, pi, opencode, dsh (the adapters in moth/integrations/, README there). |
| Rules every change follows: |
| - a config file it edits is backed up first, next to it: <file>.bak-moth-<timestamp>; |
| - JSON / YAML / TOML configs are merged: other hooks, plugins and keys stay as they were; |
| - installing twice changes nothing; a copied adapter (Pi, OpenCode, Hermes) that differs from this install's copy |
| (say after `moth update`) is replaced, the old copy backed up (Hermes: in ~/.moth/backups/, so Hermes doesn't load |
| it as a second plugin); |
| - unhook removes only moth's entries, and deletes a copied file only if it is moth's. |
| Stdlib only (no torch, no PyYAML): safe to run anywhere. Paths are read from the environment on every call |
| (HOME, XDG_CONFIG_HOME, CLAUDE_CONFIG_DIR, CODEX_HOME, HERMES_HOME, PI_CODING_AGENT_DIR). |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import filecmp |
| import json |
| import os |
| import re |
| import shlex |
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| import time |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| INTEGRATIONS = os.path.join(HERE, "integrations") |
| HARNESSES = ("claude-code", "codex", "hermes", "pi", "opencode", "dsh") |
| ALIASES = {"claude": "claude-code", "claude_code": "claude-code", "claudecode": "claude-code", "codex-cli": "codex", |
| "hermes-agent": "hermes", "open-code": "opencode", "deepseek": "dsh"} |
| BINARIES = {"claude-code": "claude", "codex": "codex", "hermes": "hermes", "pi": "pi", "opencode": "opencode", |
| "dsh": "dsh"} |
| SIGNATURE = {"pi": "// moth extension for Pi", "opencode": "// moth plugin for OpenCode"} |
| IGNORE = shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store") |
|
|
|
|
| def prog() -> str: |
| return "moth" if os.path.basename(sys.argv[0]) in ("moth", "moth.exe") else "python -m moth" |
|
|
|
|
| |
| def home() -> str: |
| return os.path.expanduser("~") |
|
|
|
|
| def xdg_config() -> str: |
| return os.environ.get("XDG_CONFIG_HOME") or os.path.join(home(), ".config") |
|
|
|
|
| def claude_dir() -> str: |
| return os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join(home(), ".claude") |
|
|
|
|
| def claude_settings(project: str | None = None) -> str: |
| if project: |
| return os.path.join(os.path.abspath(project), ".claude", "settings.json") |
| return os.path.join(claude_dir(), "settings.json") |
|
|
|
|
| def codex_dir() -> str: |
| return os.environ.get("CODEX_HOME") or os.path.join(home(), ".codex") |
|
|
|
|
| def hermes_dir() -> str: |
| return os.environ.get("HERMES_HOME") or os.path.join(home(), ".hermes") |
|
|
|
|
| def pi_dir() -> str: |
| return os.environ.get("PI_CODING_AGENT_DIR") or os.path.join(home(), ".pi", "agent") |
|
|
|
|
| def opencode_dir() -> str: |
| return os.path.join(xdg_config(), "opencode") |
|
|
|
|
| def dsh_dir() -> str: |
| return os.path.join(home(), ".dsh") |
|
|
|
|
| def config_dir(name: str) -> str: |
| return {"claude-code": claude_dir(), "codex": codex_dir(), "hermes": hermes_dir(), |
| "pi": os.path.dirname(pi_dir()), "opencode": opencode_dir(), "dsh": dsh_dir()}[name] |
|
|
|
|
| def copy_target(name: str) -> str: |
| return {"hermes": os.path.join(hermes_dir(), "plugins", "moth"), |
| "pi": os.path.join(pi_dir(), "extensions", "moth.ts"), |
| "opencode": os.path.join(opencode_dir(), "plugins", "moth.js")}[name] |
|
|
|
|
| def copy_source(name: str) -> str: |
| return {"hermes": os.path.join(INTEGRATIONS, "hermes", "moth"), "pi": os.path.join(INTEGRATIONS, "pi", "moth.ts"), |
| "opencode": os.path.join(INTEGRATIONS, "opencode", "moth.js")}[name] |
|
|
|
|
| def script(name: str) -> str: |
| return os.path.join(INTEGRATIONS, name) |
|
|
|
|
| _WIN = os.name == "nt" |
|
|
|
|
| def command(*parts: str) -> str: |
| return subprocess.list2cmdline(parts) if os.name == "nt" else shlex.join(parts) |
|
|
|
|
| def hook_command(*parts: str) -> str: |
| """A hook's command string. On Windows with forward slashes: Claude Code can run hook commands through Git Bash, |
| which eats unquoted backslashes (C:\\Python\\python.exe -> C:Pythonpython.exe); C:/Python/python.exe works in bash, |
| cmd and PowerShell alike, and so does Python with a forward-slash script path.""" |
| if _WIN: |
| return subprocess.list2cmdline([p.replace("\\", "/") for p in parts]) |
| return shlex.join(parts) |
|
|
|
|
| def ps_call(cmd: str) -> str: |
| """Codex on Windows runs a hook command through PowerShell: `powershell -NoProfile -Command <cmd>`, the session's |
| shell (codex-rs core/src/session/mod.rs build_hooks_config, core/src/shell.rs derive_exec_args, rust-v0.163.0). |
| There a line that starts with a quoted path is a string, not a call ("Unexpected token"), so an interpreter path |
| with a space (C:/Program Files/..., a venv under C:/Users/First Last) needs PowerShell's call operator `&` before |
| it. Seen on windows-latest (tests/e2e): without it Codex ran nothing. Not for Claude Code, whose Git Bash rejects &.""" |
| return "& " + cmd if cmd.startswith('"') else cmd |
|
|
|
|
| HOOK_RUN = "hook-run" |
|
|
|
|
| def appimage() -> str | None: |
| """$APPIMAGE when this is moth's own AppImage (the frozen executable runs from inside its mount, $APPDIR); None |
| otherwise. Every child of any AppImage inherits APPIMAGE (an editor shipped as one, and its terminal), so it alone |
| doesn't say this moth is one.""" |
| image, appdir = os.environ.get("APPIMAGE"), os.environ.get("APPDIR") |
| if not (getattr(sys, "frozen", False) and image and appdir): |
| return None |
| exe, root = os.path.realpath(sys.executable), os.path.realpath(appdir) |
| return image if exe.startswith(root.rstrip(os.sep) + os.sep) else None |
|
|
|
|
| def frozen_exe() -> str | None: |
| """The installers' bundled executable a hook runs instead of a Python (PyInstaller sets sys.frozen); None for a |
| Python install. moth-hook(.exe) next to the app's executable: a small stdlib-only runner that starts in a fraction |
| of the big executable's time (a hook runs on every agent turn), console on Windows (PowerShell, Codex's hook shell, |
| doesn't wait for a windowed exe). AppImage: the .AppImage file ($APPIMAGE), whose AppRun hands `hook-run` to |
| moth-hook, not its mount under /tmp/.mount_*, which moves on every launch.""" |
| if not getattr(sys, "frozen", False): |
| return None |
| if appimage(): |
| return appimage() |
| exe = sys.executable |
| runner = os.path.join(os.path.dirname(exe), "moth-hook.exe" if _WIN else "moth-hook") |
| return runner if os.path.isfile(runner) else exe |
|
|
|
|
| def hook_script(name: str) -> str: |
| """The integrations/<name> path a hook command names: this install's file; in an AppImage the bare relative |
| integrations/<name> (the bundled copy, wherever the AppImage is mounted this time).""" |
| if appimage(): |
| return "integrations/" + name |
| return script(name) |
|
|
|
|
| def hook_argv(name: str, py: str | None = None) -> list[str]: |
| """What a hook runs for integrations/<name>: [python, script], or [moth executable, hook-run, script] in the |
| installers' app (py given: that Python, always).""" |
| exe = None if py else frozen_exe() |
| return [exe, HOOK_RUN, hook_script(name)] if exe else [py or sys.executable, script(name)] |
|
|
|
|
| def harness_command(name: str, py: str | None = None) -> str: |
| """The hook command for integrations/<name> as `moth hook` writes it, per harness: Codex's gets ps_call on Windows.""" |
| cmd = hook_command(*hook_argv(name, py)) |
| return ps_call(cmd) if _WIN and name == "codex.py" else cmd |
|
|
|
|
| def _uncalled(h: dict, name: str) -> bool: |
| """Windows: a Codex hook whose command starts with a quoted path and no `&`, which PowerShell won't run.""" |
| return _WIN and name == "codex.py" and str(h.get("command", "")).lstrip().startswith('"') |
|
|
|
|
| def _backslashed(h: dict) -> bool: |
| """Windows: a moth hook written with backslash paths (older moth), which Git Bash would mangle.""" |
| return _WIN and "\\" in str(h.get("command", "")) |
|
|
|
|
| def short(path: str) -> str: |
| h = home() |
| return "~" + path[len(h):] if path == h or path.startswith(h + os.sep) else path |
|
|
|
|
| |
| def _read(path: str) -> str | None: |
| try: |
| with open(path, encoding="utf-8") as f: |
| return f.read() |
| except FileNotFoundError: |
| return None |
|
|
|
|
| def _stamp() -> str: |
| return time.strftime("%Y%m%d-%H%M%S") |
|
|
|
|
| def backup_path(path: str, where: str | None = None) -> str: |
| """<file>.bak-moth-<timestamp> (or in `where`), never overwriting an earlier backup.""" |
| base = os.path.join(where, os.path.basename(path)) if where else path |
| out, n = f"{base}.bak-moth-{_stamp()}", 1 |
| while os.path.lexists(out): |
| out, n = f"{base}.bak-moth-{_stamp()}.{n}", n + 1 |
| return out |
|
|
|
|
| def write_atomic(path: str, text: str) -> None: |
| path = os.path.realpath(path) |
| folder = os.path.dirname(os.path.abspath(path)) |
| os.makedirs(folder, exist_ok=True) |
| mode = os.stat(path).st_mode & 0o7777 if os.path.exists(path) else None |
| fd, tmp = tempfile.mkstemp(prefix=".moth-", suffix=".tmp", dir=folder) |
| try: |
| with os.fdopen(fd, "w", encoding="utf-8", newline="") as f: |
| f.write(text) |
| f.flush() |
| os.fsync(f.fileno()) |
| if mode is not None: |
| os.chmod(tmp, mode) |
| os.replace(tmp, path) |
| except BaseException: |
| try: |
| os.unlink(tmp) |
| except OSError: |
| pass |
| raise |
|
|
|
|
| def same_tree(a: str, b: str) -> bool: |
| """Two folders with the same files (ignoring __pycache__ / .pyc), byte for byte.""" |
| def files(root): |
| out = set() |
| for d, dirs, fs in os.walk(root): |
| dirs[:] = [x for x in dirs if x != "__pycache__"] |
| out |= {os.path.relpath(os.path.join(d, f), root) for f in fs if not f.endswith(".pyc") and f != ".DS_Store"} |
| return out |
| fa, fb = files(a), files(b) |
| return fa == fb and all(filecmp.cmp(os.path.join(a, f), os.path.join(b, f), shallow=False) for f in fa) |
|
|
|
|
| def same_file(a: str, b: str) -> bool: |
| return os.path.isfile(a) and os.path.isfile(b) and filecmp.cmp(a, b, shallow=False) |
|
|
|
|
| |
| class Action: |
| """kind: write (path, text) | copy (src -> path, a file or a folder) | remove (path, a file or a folder) | |
| run (cmd). backup: back the old path up first (backup_dir: where, for folders a harness would load).""" |
|
|
| def __init__(self, kind: str, desc: str, path: str | None = None, text: str | None = None, |
| src: str | None = None, cmd: list[str] | None = None, backup: bool = True, |
| backup_dir: str | None = None, backup_also: tuple[str, ...] = (), fallback: list[str] | None = None): |
| self.kind, self.desc, self.path, self.text, self.src, self.cmd = kind, desc, path, text, src, cmd |
| self.fallback = fallback |
| self.backup, self.backup_dir, self.backup_also = backup, backup_dir, backup_also |
|
|
| def apply(self) -> list[str]: |
| """Do it; returns the backups made.""" |
| made = [] |
| for p in ((self.path,) if self.backup and self.path else ()) + self.backup_also: |
| if p and os.path.lexists(p): |
| if self.backup_dir: |
| os.makedirs(self.backup_dir, exist_ok=True) |
| dst = backup_path(p, self.backup_dir) |
| (shutil.copytree(p, dst, ignore=IGNORE, symlinks=True) if os.path.isdir(p) and not os.path.islink(p) |
| else shutil.copy2(p, dst)) |
| made.append(dst) |
| if self.kind == "write": |
| write_atomic(self.path, self.text) |
| elif self.kind == "copy": |
| if os.path.isdir(self.src): |
| tmp = self.path + ".moth-new" |
| shutil.rmtree(tmp, ignore_errors=True) |
| os.makedirs(os.path.dirname(self.path), exist_ok=True) |
| shutil.copytree(self.src, tmp, ignore=IGNORE) |
| if os.path.lexists(self.path): |
| shutil.rmtree(self.path) if os.path.isdir(self.path) and not os.path.islink(self.path) \ |
| else os.remove(self.path) |
| os.replace(tmp, self.path) |
| else: |
| with open(self.src, encoding="utf-8") as f: |
| write_atomic(self.path, f.read()) |
| elif self.kind == "remove": |
| if os.path.isdir(self.path) and not os.path.islink(self.path): |
| shutil.rmtree(self.path) |
| elif os.path.lexists(self.path): |
| os.remove(self.path) |
| elif self.kind == "run": |
| r = subprocess.run(self.cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, |
| text=True, timeout=120) |
| if r.returncode != 0 and self.fallback and "unrecognized arguments" in r.stdout: |
| r = subprocess.run(self.fallback, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, text=True, timeout=120) |
| if r.returncode != 0: |
| raise RuntimeError(f"{command(*self.cmd)} failed (exit {r.returncode}): {r.stdout.strip()[-300:]}") |
| return made |
|
|
|
|
| class Plan: |
| def __init__(self, harness: str): |
| self.harness = harness |
| self.actions: list[Action] = [] |
| self.notes: list[str] = [] |
| self.nothing: str | None = None |
| self.error: str | None = None |
|
|
|
|
| |
| |
| |
| |
| HOOK_TIMEOUT = 3 |
|
|
|
|
| def hook_entry(cmd: str) -> dict: |
| """moth's command hook, as `moth hook` writes it and `python -m moth` prints it.""" |
| return {"type": "command", "command": cmd, "timeout": HOOK_TIMEOUT} |
|
|
|
|
| def moth_script_in(cmd, name: str) -> str | None: |
| """The path of moth's integrations/<name> that a hook command runs (any interpreter, MOTH_URL=... prefixes and |
| quoting allowed), else None. A copy of moth elsewhere counts too: it's still moth's hook.""" |
| if not isinstance(cmd, str): |
| return None |
| try: |
| parts = shlex.split(cmd, posix=os.name != "nt") |
| except ValueError: |
| parts = cmd.split() |
| for p in parts: |
| p = p.strip("\"'") |
| norm = p.replace("\\", "/") |
| if norm.endswith("/integrations/" + name) or norm == "integrations/" + name: |
| return p |
| return None |
|
|
|
|
| def _load_json(path: str) -> tuple[dict | None, str | None]: |
| text = _read(path) |
| if text is None or not text.strip(): |
| return {}, None |
| try: |
| data = json.loads(text) |
| except ValueError as e: |
| return None, f"{path} isn't valid JSON ({e}); fix it by hand first, moth won't touch it" |
| if not isinstance(data, dict): |
| return None, f"{path} isn't a JSON object; moth won't touch it" |
| return data, None |
|
|
|
|
| def _dump_json(data: dict, old: str | None) -> str: |
| indent = 2 |
| m = re.search(r"\n( +)\S", old or "") |
| if m: |
| indent = len(m.group(1)) |
| return json.dumps(data, indent=indent, ensure_ascii=False) + "\n" |
|
|
|
|
| def _stop_entries(data: dict, name: str) -> list[tuple[dict, dict, str]]: |
| """(group, handler, script path) for each of moth's handlers under hooks.Stop.""" |
| out = [] |
| hooks = data.get("hooks") |
| stop = hooks.get("Stop") if isinstance(hooks, dict) else None |
| for group in stop if isinstance(stop, list) else (): |
| for h in (group.get("hooks") or ()) if isinstance(group, dict) else (): |
| if isinstance(h, dict): |
| p = moth_script_in(h.get("command"), name) |
| if p: |
| out.append((group, h, p)) |
| return out |
|
|
|
|
| def _where(p: str, name: str) -> str: |
| """'this' (this install's script), 'other' (another moth that exists) or 'missing'.""" |
| if os.path.realpath(os.path.expanduser(p)) == os.path.realpath(script(name)): |
| return "this" |
| if p.replace("\\", "/") == "integrations/" + name and hook_script(name) == "integrations/" + name: |
| return "this" |
| return "other" if os.path.isfile(os.path.expanduser(p)) else "missing" |
|
|
|
|
| def json_hook_status(path: str, name: str) -> tuple[str, str]: |
| data, err = _load_json(path) |
| if err: |
| return "error", err |
| entries = _stop_entries(data, name) |
| if not entries: |
| return "absent", f"no moth hook in {short(path)}" |
| kinds = [_where(p, name) for _, _, p in entries] |
| if "this" in kinds: |
| if any(h.get("async") for _, h, _ in entries): |
| return "outdated", (f"Stop hook in {short(path)} runs in the background (\"async\": true), which one-shot " |
| f"runs (claude -p, codex exec) kill before it posts") |
| if any(_backslashed(h) for _, h, _ in entries): |
| return "outdated", (f"Stop hook in {short(path)} has backslash paths, which Git Bash mangles; " |
| f"`{prog()} hook` rewrites them with forward slashes") |
| if any(_uncalled(h, name) for _, h, _ in entries): |
| return "outdated", (f"Stop hook in {short(path)} starts with a quoted path, which PowerShell (Codex's hook " |
| f"shell on Windows) doesn't run; `{prog()} hook` adds its call operator &") |
| extra = f" ({len(entries)} moth hooks: duplicates)" if len(entries) > 1 else "" |
| return "installed", f"Stop hook in {short(path)}{extra}" |
| if "other" in kinds: |
| return "other", f"Stop hook in {short(path)} runs another moth copy: {entries[0][2]}" |
| return "broken", f"Stop hook in {short(path)} points to a missing file: {entries[0][2]}" |
|
|
|
|
| def plan_json_hook(plan: Plan, path: str, name: str, what: str) -> None: |
| data, err = _load_json(path) |
| if err: |
| plan.error = err |
| return |
| old = _read(path) |
| want = harness_command(name) |
| entries = _stop_entries(data, name) |
| here = any(_where(p, name) == "this" for _, _, p in entries) |
| background = [h for _, h, _ in entries if h.get("async")] |
| slashes = [h for _, h, _ in entries if _backslashed(h)] |
| uncalled = [h for _, h, _ in entries if _uncalled(h, name)] |
| if here and not background and not slashes and not uncalled: |
| plan.nothing = f"already installed ({what} in {short(path)})" |
| return |
| if entries: |
| changes = [] |
| if not here: |
| for _, h, p in entries: |
| h["command"] = want |
| changes.append(f"{entries[0][2]} -> {script(name)}") |
| elif slashes: |
| for h in slashes: |
| h["command"] = h["command"].replace("\\", "/") |
| changes.append("forward slashes in its paths (Git Bash mangles backslashes)") |
| if here and uncalled: |
| for h in uncalled: |
| h["command"] = ps_call(h["command"].strip()) |
| changes.append("PowerShell's call operator & before its quoted interpreter path (Codex runs hooks " |
| "through PowerShell on Windows)") |
| for h in background: |
| del h["async"] |
| h["timeout"] = HOOK_TIMEOUT |
| if background: |
| changes.append("run it in the foreground (a background hook is killed by claude -p / codex exec before " |
| "it posts)") |
| plan.actions.append(Action("write", f"update moth's {what} in {short(path)}: " + "; ".join(changes), path, |
| _dump_json(data, old))) |
| return |
| hooks = data.setdefault("hooks", {}) |
| if not isinstance(hooks, dict): |
| plan.error = f"{path}: \"hooks\" isn't an object; moth won't touch it" |
| return |
| stop = hooks.setdefault("Stop", []) |
| if not isinstance(stop, list): |
| plan.error = f"{path}: \"hooks.Stop\" isn't a list; moth won't touch it" |
| return |
| stop.append({"hooks": [hook_entry(want)]}) |
| verb = "add" if old is not None else "create" |
| plan.actions.append(Action("write", f"{verb} {short(path)} with moth's {what}: {want}" if verb == "create" else |
| f"add moth's {what} to {short(path)}: {want}", path, _dump_json(data, old))) |
|
|
|
|
| def plan_json_unhook(plan: Plan, path: str, name: str, what: str) -> None: |
| data, err = _load_json(path) |
| if err: |
| plan.error = err |
| return |
| old = _read(path) |
| entries = _stop_entries(data, name) |
| if not entries: |
| return |
| stop = data["hooks"]["Stop"] |
| ids = {id(h) for _, h, _ in entries} |
| for group in stop: |
| if isinstance(group, dict) and isinstance(group.get("hooks"), list): |
| group["hooks"] = [h for h in group["hooks"] if id(h) not in ids] |
| data["hooks"]["Stop"] = [g for g in stop if not (isinstance(g, dict) and g.get("hooks") == [] |
| and any(id(g) == id(eg) for eg, _, _ in entries))] |
| if not data["hooks"]["Stop"]: |
| del data["hooks"]["Stop"] |
| if not data["hooks"]: |
| del data["hooks"] |
| plan.actions.append(Action("write", f"remove moth's {what} from {short(path)} " |
| f"({len(entries)} entr{'y' if len(entries) == 1 else 'ies'})", |
| path, _dump_json(data, old))) |
|
|
|
|
| |
| NOTIFY = re.compile(r"^notify\s*=\s*\[(?P<body>[^\]]*)\][^\n]*\n?", re.M) |
|
|
|
|
| def _toml_top(text: str) -> str: |
| """The part of a TOML file before its first [table] header (where a top-level key like notify lives).""" |
| m = re.search(r"^\s*\[", text, re.M) |
| return text[:m.start()] if m else text |
|
|
|
|
| def codex_notify(text: str | None) -> tuple[re.Match | None, list[str]]: |
| if not text: |
| return None, [] |
| m = NOTIFY.search(_toml_top(text)) |
| if not m: |
| return None, [] |
| items = [_toml_basic(s[1:-1]) if s.startswith('"') else s[1:-1] |
| for s in re.findall(r'"(?:[^"\\]|\\.)*"|\'[^\']*\'', m.group("body"))] |
| return m, items |
|
|
|
|
| _TOML_ESC = {"b": "\b", "t": "\t", "n": "\n", "f": "\f", "r": "\r", "e": "\x1b", '"': '"', "\\": "\\"} |
|
|
|
|
| def _toml_basic(body: str) -> str: |
| """The value of a TOML basic string ("..." without the quotes). Not json.loads: TOML also has \\e and \\UXXXXXXXX, |
| and a hand-written invalid escape (a raw Windows path like "C:\\Users\\me\\x.py") must not crash `moth hooks`: then |
| the text is taken as it is, which is what the person meant.""" |
| out, i = [], 0 |
| while i < len(body): |
| c = body[i] |
| if c != "\\": |
| out.append(c) |
| i += 1 |
| continue |
| k = body[i + 1:i + 2] |
| if k in _TOML_ESC: |
| out.append(_TOML_ESC[k]) |
| i += 2 |
| elif k in ("u", "U"): |
| n = 4 if k == "u" else 8 |
| hexd = body[i + 2:i + 2 + n] |
| try: |
| if len(hexd) != n: |
| raise ValueError |
| out.append(chr(int(hexd, 16))) |
| except ValueError: |
| return body |
| i += 2 + n |
| else: |
| return body |
| return "".join(out) |
|
|
|
|
| def codex_config() -> str: |
| return os.path.join(codex_dir(), "config.toml") |
|
|
|
|
| def codex_hooks() -> str: |
| return os.path.join(codex_dir(), "hooks.json") |
|
|
|
|
| def notify_is_moth(items: list[str]) -> str | None: |
| for it in items: |
| if moth_script_in(command(it), "codex.py") or it.replace("\\", "/").endswith("/integrations/codex.py"): |
| return it |
| return None |
|
|
|
|
| |
| def yaml_enabled_plugins(text: str | None) -> list[str]: |
| """plugins.enabled of a Hermes config.yaml (block or flow list). PyYAML if it's there, else a small reader.""" |
| if not text: |
| return [] |
| try: |
| import yaml |
| d = yaml.safe_load(text) |
| en = (d or {}).get("plugins", {}).get("enabled") if isinstance(d, dict) else None |
| return [str(x) for x in en] if isinstance(en, list) else [] |
| except ImportError: |
| pass |
| except Exception: |
| return [] |
| lines = text.splitlines() |
| out, in_plugins, en_indent = [], False, None |
| for line in lines: |
| if not line.strip() or line.lstrip().startswith("#"): |
| continue |
| indent = len(line) - len(line.lstrip()) |
| if indent == 0: |
| in_plugins = line.split("#")[0].strip() == "plugins:" |
| en_indent = None |
| continue |
| if not in_plugins: |
| continue |
| if en_indent is None: |
| m = re.match(r"\s*enabled:\s*(.*)$", line) |
| if m: |
| rest = m.group(1).split("#")[0].strip() |
| if rest.startswith("["): |
| return [x.strip().strip("'\"") for x in rest.strip("[]").split(",") if x.strip()] |
| en_indent = indent |
| continue |
| m = re.match(r"\s*-\s*(.+)$", line) |
| if indent >= en_indent and m: |
| out.append(m.group(1).split(" #")[0].strip().strip("'\"")) |
| elif indent <= en_indent: |
| break |
| return out |
|
|
|
|
| def hermes_config() -> str: |
| return os.path.join(hermes_dir(), "config.yaml") |
|
|
|
|
| def hermes_is_moth(folder: str) -> bool: |
| y = _read(os.path.join(folder, "plugin.yaml")) or "" |
| return bool(re.search(r"^name:\s*['\"]?moth['\"]?\s*$", y, re.M)) and \ |
| os.path.isfile(os.path.join(folder, "__init__.py")) |
|
|
|
|
| |
| def file_is_moth(path: str, name: str) -> bool: |
| if same_file(path, copy_source(name)): |
| return True |
| try: |
| with open(path, encoding="utf-8") as f: |
| return f.readline().startswith(SIGNATURE[name]) |
| except (OSError, UnicodeDecodeError): |
| return False |
|
|
|
|
| |
| DSH_LINE = re.compile(r"^[ \t]*-[ \t]*insert:[ \t]*\[[ \t]*\{[ \t]*id:[ \t]*['\"]?moth['\"]?[ \t]*,[ \t]*name:[ \t]*" |
| r"(?P<name>'(?:[^']|'')*'|\"(?:[^\"\\]|\\.)*\"|[^}\s]+)[ \t]*\}[ \t]*\][ \t]*(?:#[^\n]*)?\n?", |
| re.M) |
|
|
|
|
| def dsh_profiles() -> list[str]: |
| base = os.path.join(dsh_dir(), "profiles") |
| return sorted(d for d in os.listdir(base) if os.path.isdir(os.path.join(base, d))) if os.path.isdir(base) else [] |
|
|
|
|
| def dsh_patch(profile: str) -> str: |
| return os.path.join(dsh_dir(), "profiles", profile, "cordis.patch.yml") |
|
|
|
|
| def dsh_line() -> str: |
| return "- insert: [{ id: moth, name: '" + script(os.path.join("dsh", "moth.mjs")).replace("'", "''") + "' }]\n" |
|
|
|
|
| def _dsh_name(m: re.Match) -> str: |
| n = m.group("name") |
| if n.startswith("'"): |
| return n[1:-1].replace("''", "'") |
| if n.startswith('"'): |
| try: |
| return json.loads(n) |
| except ValueError: |
| return n[1:-1] |
| return n |
|
|
|
|
| |
| def detected(name: str) -> tuple[bool, str]: |
| why = [] |
| b = shutil.which(BINARIES[name]) |
| if b: |
| why.append(f"`{BINARIES[name]}` on PATH") |
| if os.path.isdir(config_dir(name)): |
| why.append(short(config_dir(name))) |
| return bool(why), ", ".join(why) |
|
|
|
|
| def status(name: str, project: str | None = None) -> dict: |
| """{"name", "detected", "why", "state", "detail"}; state: installed | outdated | other | broken | partial | |
| absent | error. outdated / other / broken / partial are fixed by `moth hook <name>`.""" |
| found, why = detected(name) |
| st = {"name": name, "detected": found, "why": why} |
| if name == "claude-code": |
| st["state"], st["detail"] = json_hook_status(claude_settings(project), "claude_code.py") |
| elif name == "codex": |
| s, d = json_hook_status(codex_hooks(), "codex.py") |
| _, items = codex_notify(_read(codex_config())) |
| legacy = notify_is_moth(items) |
| if legacy and s != "absent" and s != "error": |
| s, d = "error", f"BOTH variants installed (hooks.json and legacy notify in {short(codex_config())}): " \ |
| f"every reply posts twice; `{prog()} unhook codex` then `{prog()} hook codex`" |
| elif legacy: |
| s = "installed" if _where(legacy, "codex.py") == "this" else _where(legacy, "codex.py").replace( |
| "missing", "broken") |
| d = f"legacy notify in {short(codex_config())}" |
| elif s == "installed": |
| d += " (Codex runs it only once approved: /hooks in codex)" |
| st["state"], st["detail"] = s, d |
| elif name in ("pi", "opencode"): |
| dst = copy_target(name) |
| if not os.path.exists(dst): |
| st["state"], st["detail"] = "absent", f"no {short(dst)}" |
| elif same_file(dst, copy_source(name)): |
| st["state"], st["detail"] = "installed", short(dst) |
| elif file_is_moth(dst, name): |
| st["state"], st["detail"] = "outdated", f"{short(dst)} differs from this install's copy" |
| else: |
| st["state"], st["detail"] = "error", f"{short(dst)} exists but isn't moth's adapter" |
| elif name == "hermes": |
| dst = copy_target(name) |
| on = "moth" in yaml_enabled_plugins(_read(hermes_config())) |
| if not os.path.exists(dst): |
| st["state"], st["detail"] = ("broken", f"enabled in config.yaml but {short(dst)} is missing") if on else \ |
| ("absent", f"no {short(dst)}") |
| elif not hermes_is_moth(dst): |
| st["state"], st["detail"] = "error", f"{short(dst)} exists but isn't moth's plugin" |
| elif not same_tree(dst, copy_source(name)): |
| st["state"], st["detail"] = "outdated", f"{short(dst)} differs from this install's copy" + \ |
| ("" if on else ", and not enabled") |
| elif not on: |
| st["state"], st["detail"] = "partial", f"{short(dst)} copied but not enabled (hermes plugins enable moth)" |
| else: |
| st["state"], st["detail"] = "installed", f"{short(dst)}, enabled" |
| elif name == "dsh": |
| found_in = [] |
| for p in dsh_profiles(): |
| for m in DSH_LINE.finditer(_read(dsh_patch(p)) or ""): |
| found_in.append((p, _dsh_name(m))) |
| if not found_in: |
| st["state"], st["detail"] = "absent", "no moth entry in any ~/.dsh profile" |
| else: |
| kinds = [_where(n, os.path.join("dsh", "moth.mjs")) for _, n in found_in] |
| st["state"] = "installed" if all(k == "this" for k in kinds) else \ |
| "other" if "missing" not in kinds else "broken" |
| st["detail"] = "profile " + ", ".join(sorted({p for p, _ in found_in})) |
| return st |
|
|
|
|
| MARK = {"installed": "✓", "outdated": "!", "other": "!", "broken": "✗", "partial": "!", "error": "✗", "absent": "·"} |
| WORD = {"installed": "installed", "outdated": "outdated", "other": "another moth", "broken": "broken", |
| "partial": "half-installed", "error": "problem", "absent": "not installed"} |
|
|
|
|
| def status_lines(project: str | None = None, only_detected: bool = False) -> list[tuple[str, str]]: |
| """(mark, text) per harness, for `moth hooks` and `moth doctor`.""" |
| out = [] |
| for name in HARNESSES: |
| st = status(name, project) |
| if not st["detected"] and st["state"] == "absent": |
| if not only_detected: |
| out.append(("-", f"{name:<12} not found on this machine")) |
| continue |
| hint = "" |
| if st["state"] in ("outdated", "other", "broken", "partial", "absent"): |
| hint = f" -> {prog()} hook {name}" |
| where = f" [{st['why']}]" if st["why"] else " [not detected, but configured]" |
| out.append((MARK[st["state"]], f"{name:<12} {WORD[st['state']]}: {st['detail']}{hint}{where}")) |
| return out |
|
|
|
|
| def ascii_safe(mark: str) -> str: |
| enc = (getattr(sys.stdout, "encoding", None) or "ascii") |
| try: |
| mark.encode(enc) |
| return mark |
| except (UnicodeEncodeError, LookupError): |
| return {"✓": "ok", "✗": "XX", "!": "!!", "·": "..", "-": "--"}.get(mark, mark) |
|
|
|
|
| def show(project: str | None = None) -> int: |
| print("coding agents moth can watch (✓ hooked up):".replace("✓", ascii_safe("✓"))) |
| for mark, text in status_lines(project): |
| print(f" {ascii_safe(mark)} {text}") |
| print(f"\n{prog()} hook <name> installs one ({prog()} hook --all: every harness found), {prog()} unhook <name> " |
| "removes it.\nThe Claude Code band above the prompt is a separate mod: claude --plugin-dir " |
| f"{command(os.path.join(INTEGRATIONS, 'claude_code_mod'))}") |
| return 0 |
|
|
|
|
| |
| def plan_hook(name: str, project: str | None = None, profile: str | None = None, legacy_notify: bool = False) -> Plan: |
| plan = Plan(name) |
| if name == "claude-code": |
| plan_json_hook(plan, claude_settings(project), "claude_code.py", "Stop hook") |
| if plan.actions: |
| plan.notes.append("Claude Code picks the hook up in new sessions (or after /hooks). The band above the " |
| f"prompt is separate: claude --plugin-dir " |
| f"{command(os.path.join(INTEGRATIONS, 'claude_code_mod'))}") |
| elif name == "codex": |
| text = _read(codex_config()) |
| m, items = codex_notify(text) |
| legacy = notify_is_moth(items) |
| hs, _ = json_hook_status(codex_hooks(), "codex.py") |
| if legacy and hs not in ("absent", "error"): |
| plan.error = (f"both Codex variants are installed (hooks.json and notify in {short(codex_config())}), so " |
| f"every reply posts twice: run `{prog()} unhook codex`, then `{prog()} hook codex`") |
| elif legacy_notify: |
| if hs not in ("absent", "error"): |
| plan.error = (f"moth's hooks.json hook is installed; moth never installs both variants. To switch: " |
| f"`{prog()} unhook codex`, then `{prog()} hook codex --legacy-notify`") |
| elif legacy and _where(legacy, "codex.py") == "this": |
| plan.nothing = f"already installed (legacy notify in {short(codex_config())})" |
| elif m and not legacy: |
| plan.error = (f"{short(codex_config())} already has a notify program ({' '.join(items)}); Codex has " |
| f"one notify slot, so moth won't replace it. Use the hooks.json variant (without " |
| f"--legacy-notify) instead") |
| else: |
| want = f"notify = {json.dumps(hook_argv('codex.py'), ensure_ascii=False)}\n" |
| if m: |
| new = text[:m.start()] + want + text[m.end():] |
| desc = f"update moth's legacy notify in {short(codex_config())}" |
| else: |
| lead = re.match(r"(?:[ \t]*(?:#[^\n]*)?\n)*", text or "").end() |
| new = (text or "")[:lead] + want + (text or "")[lead:] |
| desc = f"add legacy notify to {short(codex_config())}: {want.strip()}" |
| plan.actions.append(Action("write", desc, codex_config(), new)) |
| elif legacy: |
| plan.nothing = (f"already installed as legacy notify in {short(codex_config())}; moth never installs both " |
| f"variants (to switch to hooks.json: `{prog()} unhook codex`, then `{prog()} hook codex`)") |
| if _where(legacy, "codex.py") != "this": |
| plan.nothing = None |
| want = f"notify = {json.dumps(hook_argv('codex.py'), ensure_ascii=False)}\n" |
| plan.actions.append(Action("write", f"update moth's legacy notify in {short(codex_config())}", |
| codex_config(), text[:m.start()] + want + text[m.end():])) |
| else: |
| plan_json_hook(plan, codex_hooks(), "codex.py", "Stop hook") |
| if plan.actions and not legacy_notify and not legacy: |
| plan.notes.append("Codex runs a new or changed hook only after you approve it once: start `codex` and " |
| "accept it in the hook review (or later with /hooks). Until then it is skipped silently.") |
| elif name in ("pi", "opencode"): |
| src, dst = copy_source(name), copy_target(name) |
| if not os.path.exists(dst): |
| plan.actions.append(Action("copy", f"copy {short(src)} -> {short(dst)}", dst, src=src, backup=False)) |
| elif same_file(dst, src): |
| plan.nothing = f"already installed ({short(dst)})" |
| elif not file_is_moth(dst, name): |
| plan.error = f"{short(dst)} exists and isn't moth's adapter; moth won't overwrite it" |
| else: |
| plan.actions.append(Action("copy", f"update {short(dst)} (differed from this install's copy)", dst, |
| src=src)) |
| if plan.actions: |
| plan.notes.append(f"{'Pi' if name == 'pi' else 'OpenCode'} loads it at its next start.") |
| elif name == "hermes": |
| src, dst = copy_source(name), copy_target(name) |
| backups = os.path.join(home(), ".moth", "backups") |
| if not os.path.exists(dst): |
| plan.actions.append(Action("copy", f"copy {short(src)}/ -> {short(dst)}/", dst, src=src, backup=False)) |
| elif not hermes_is_moth(dst): |
| plan.error = f"{short(dst)} exists and isn't moth's plugin; moth won't overwrite it" |
| return plan |
| elif not same_tree(dst, src): |
| plan.actions.append(Action("copy", f"update {short(dst)}/ (differed from this install's copy; old one " |
| f"kept in {short(backups)})", dst, src=src, backup_dir=backups)) |
| if "moth" not in yaml_enabled_plugins(_read(hermes_config())): |
| hermes = shutil.which("hermes") |
| if hermes: |
| plan.actions.append(Action("run", f"run `hermes plugins enable moth` (adds moth to plugins.enabled in " |
| f"{short(hermes_config())})", |
| |
| cmd=[hermes, "plugins", "enable", "--no-allow-tool-override", "moth"], |
| fallback=[hermes, "plugins", "enable", "moth"], |
| backup=False, backup_also=(hermes_config(),))) |
| else: |
| plan.notes.append(f"`hermes` isn't on PATH: enable the plugin with `hermes plugins enable moth`, or add " |
| f"moth under plugins.enabled in {short(hermes_config())}") |
| if not plan.actions and not plan.notes: |
| plan.nothing = f"already installed ({short(dst)}, enabled)" |
| elif name == "dsh": |
| profiles = dsh_profiles() |
| if not profile: |
| if len(profiles) == 1: |
| profile = profiles[0] |
| else: |
| plan.error = ("pick a dsh profile with --profile NAME (" + ( |
| f"found: {', '.join(profiles)}" if profiles else f"none in {short(os.path.join(dsh_dir(), 'profiles'))}") |
| + ")") |
| return plan |
| path = dsh_patch(profile) |
| text = _read(path) |
| want = dsh_line() |
| ms = list(DSH_LINE.finditer(text or "")) |
| mjs = os.path.join("dsh", "moth.mjs") |
| if any(_where(_dsh_name(m), mjs) == "this" for m in ms): |
| plan.nothing = f"already installed ({short(path)})" |
| elif ms: |
| new = text |
| for m in reversed(ms): |
| new = new[:m.start()] + want + new[m.end():] |
| plan.actions.append(Action("write", f"update moth's entry in {short(path)} to {script(mjs)}", path, new)) |
| else: |
| body = text or "" |
| if body.strip() in ("", "[]"): |
| body = "" |
| elif not body.endswith("\n"): |
| body += "\n" |
| plan.actions.append(Action("write", f"{'add moth to' if text is not None else 'create'} {short(path)}: " |
| f"{want.strip()}", path, body + want)) |
| return plan |
|
|
|
|
| def plan_unhook(name: str, project: str | None = None, profile: str | None = None) -> Plan: |
| plan = Plan(name) |
| if name == "claude-code": |
| plan_json_unhook(plan, claude_settings(project), "claude_code.py", "Stop hook") |
| elif name == "codex": |
| plan_json_unhook(plan, codex_hooks(), "codex.py", "Stop hook") |
| text = _read(codex_config()) |
| m, items = codex_notify(text) |
| if m and notify_is_moth(items): |
| plan.actions.append(Action("write", f"remove moth's legacy notify from {short(codex_config())}", |
| codex_config(), text[:m.start()] + text[m.end():])) |
| elif name in ("pi", "opencode"): |
| dst = copy_target(name) |
| if os.path.exists(dst): |
| if not file_is_moth(dst, name): |
| plan.error = f"{short(dst)} isn't moth's adapter; moth won't delete it" |
| else: |
| plan.actions.append(Action("remove", f"delete {short(dst)}", dst, |
| backup=not same_file(dst, copy_source(name)))) |
| elif name == "hermes": |
| dst = copy_target(name) |
| if "moth" in yaml_enabled_plugins(_read(hermes_config())): |
| hermes = shutil.which("hermes") |
| if hermes: |
| plan.actions.append(Action("run", f"run `hermes plugins disable moth` ({short(hermes_config())})", |
| cmd=[hermes, "plugins", "disable", "moth"], backup=False, |
| backup_also=(hermes_config(),))) |
| else: |
| plan.notes.append(f"`hermes` isn't on PATH: remove moth from plugins.enabled in " |
| f"{short(hermes_config())} by hand") |
| if os.path.exists(dst): |
| if not hermes_is_moth(dst): |
| plan.error = f"{short(dst)} isn't moth's plugin; moth won't delete it" |
| else: |
| plan.actions.append(Action("remove", f"delete {short(dst)}/", dst, |
| backup=not same_tree(dst, copy_source(name)), |
| backup_dir=os.path.join(home(), ".moth", "backups"))) |
| elif name == "dsh": |
| for p in ([profile] if profile else dsh_profiles()): |
| path = dsh_patch(p) |
| text = _read(path) |
| ms = list(DSH_LINE.finditer(text or "")) |
| if ms: |
| new = text |
| for m in reversed(ms): |
| new = new[:m.start()] + new[m.end():] |
| plan.actions.append(Action("write", f"remove moth's entry from {short(path)}", path, new)) |
| if not plan.actions and not plan.error and not plan.notes: |
| plan.nothing = "not installed, nothing to remove" |
| return plan |
|
|
|
|
| |
| def _confirm(question: str) -> bool: |
| try: |
| return input(question + " [y/N] ").strip().lower() in ("y", "yes") |
| except EOFError: |
| return False |
|
|
|
|
| def run_plans(plans: list[Plan], yes: bool, dry_run: bool, verb: str) -> int: |
| rc = 0 |
| todo = [p for p in plans if p.actions and not p.error] |
| for p in plans: |
| if p.error: |
| print(f"{p.harness}: {p.error}") |
| rc = 1 |
| elif not p.actions: |
| if p.nothing: |
| print(f"{p.harness}: {p.nothing}") |
| for n in p.notes: |
| print(f"{p.harness}: note: {n}") |
| if not todo: |
| return rc |
| print(("would " if dry_run else "moth will ") + verb + ":") |
| for p in todo: |
| for a in p.actions: |
| print(f" {p.harness}: {a.desc}") |
| if dry_run: |
| print("(dry run: nothing changed)") |
| return rc |
| if not yes and not _confirm("go ahead?"): |
| print("cancelled, nothing changed") |
| return 1 |
| for p in todo: |
| for a in p.actions: |
| try: |
| made = a.apply() |
| except Exception as e: |
| print(f" {p.harness}: FAILED: {a.desc}: {e}") |
| rc = 1 |
| break |
| print(f" {p.harness}: done: {a.desc}") |
| for b in made: |
| print(f" backup: {b}") |
| for n in p.notes: |
| print(f" {p.harness}: note: {n}") |
| return rc |
|
|
|
|
| def _names(raw: list[str], everything: bool, pick) -> list[str] | str: |
| if everything: |
| return [n for n in HARNESSES if pick(n)] |
| out = [] |
| for r in raw: |
| n = ALIASES.get(r.lower(), r.lower()) |
| if n not in HARNESSES: |
| return f"unknown harness {r!r}; one of: {', '.join(HARNESSES)}" |
| if n not in out: |
| out.append(n) |
| return out |
|
|
|
|
| def _parser(cmd: str) -> argparse.ArgumentParser: |
| what = {"hooks": "list the coding agents found here and whether moth is hooked into each", |
| "hook": "install moth's hook / plugin into coding agents (backs up and merges configs)", |
| "unhook": "remove moth's hook / plugin from coding agents (only moth's entries)"}[cmd] |
| ap = argparse.ArgumentParser(prog=f"{prog()} {cmd}", description=what, |
| epilog=f"harnesses: {', '.join(HARNESSES)}") |
| if cmd != "hooks": |
| ap.add_argument("harness", nargs="*", help="one or more harness names") |
| ap.add_argument("--all", action="store_true", |
| help="every harness found on this machine" if cmd == "hook" else "every harness moth is in") |
| ap.add_argument("--yes", "-y", action="store_true", help="don't ask before changing files") |
| ap.add_argument("--dry-run", action="store_true", help="only show what would change") |
| ap.add_argument("--profile", help="dsh profile (default: the only one in ~/.dsh/profiles)") |
| if cmd == "hook": |
| ap.add_argument("--legacy-notify", action="store_true", |
| help="Codex: use the legacy notify line in config.toml instead of hooks.json") |
| ap.add_argument("--project", metavar="DIR", help="Claude Code: DIR/.claude/settings.json instead of ~/.claude") |
| return ap |
|
|
|
|
| def main(cmd: str, argv=None) -> int: |
| a = _parser(cmd).parse_args(argv) |
| if cmd == "hooks": |
| return show(a.project) |
| if a.all and a.harness: |
| print("give harness names or --all, not both", file=sys.stderr) |
| return 2 |
| if cmd == "hook": |
| names = _names(a.harness, a.all, lambda n: detected(n)[0]) |
| else: |
| names = _names(a.harness, a.all, lambda n: status(n, a.project)["state"] != "absent") |
| if isinstance(names, str): |
| print(names, file=sys.stderr) |
| return 2 |
| if not names: |
| if a.all: |
| print("no coding agents found on this machine" if cmd == "hook" else "moth isn't hooked into anything") |
| return 0 |
| _parser(cmd).print_usage(sys.stderr) |
| return 2 |
| if cmd == "hook": |
| plans = [plan_hook(n, a.project, a.profile, a.legacy_notify) for n in names] |
| return run_plans(plans, a.yes, a.dry_run, "install") |
| plans = [plan_unhook(n, a.project, a.profile) for n in names] |
| return run_plans(plans, a.yes, a.dry_run, "remove") |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "hooks", sys.argv[2:])) |
|
|
