Hooks

Where it lives

• hooks/connect_opencode.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Auto-connect opencode to agentglass via its plugin system.
  
  agentglass exposes an HTTP /ingest endpoint that accepts normalised events
  from any provider.  This script deploys the opencode plugin (a dependency-free
  JS file that hooks into opencode's event bus and POSTs events to /ingest) into
  the user's opencode plugin directory.
  
    python3 hooks/connect_opencode.py               # deploy the plugin
    python3 hooks/connect_opencode.py --undo        # remove it again
    python3 hooks/connect_opencode.py --postinstall # lifecycle mode (honours AGENTGLASS_NO_OPENCODE)
  
  Idempotent, backs up before overwriting, and never fails the install.
  """
  import argparse
  import os
  import shutil
  import sys
  import time
  from pathlib import Path
  
  def _default_server(host):
      """http://<host>:<port> of the running server. The port is the one it wrote
      at bind (next to the token file), because the app moves to another port when
      4000 is taken and the AGENTGLASS_PORT an agent pane inherits can be stale;
      then AGENTGLASS_PORT, then 4000. No probe: a hook cannot pay a round trip."""
      base = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
      port = ""
      try:
          with open(os.path.join(base, "agentglass", "port"), encoding="utf-8") as f:
              port = f.read().strip()
      except OSError:
          pass
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = os.environ.get("AGENTGLASS_PORT", "").strip()
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = "4000"
      return f"http://{host}:{port}"
  
  
  SERVER = (os.environ.get("AGENTGLASS_SERVER") or _default_server("127.0.0.1")).rstrip("/")
  
  PLUGIN_FILENAME = "agentglass.js"
  PLUGIN_SRC = Path(__file__).resolve().parent / "opencode-plugin.js"
  PLUGIN_MARKER = "// agentglass opencode plugin"
  
  
  def _agentglass_local_only(url):
      from urllib.parse import urlparse
      if os.environ.get("AGENTGLASS_ALLOW_REMOTE") == "1":
          return True
      u = urlparse(url or "")
      if u.scheme not in ("http", "https") or (u.hostname or "") not in ("localhost", "127.0.0.1", "::1"):
          sys.stderr.write("[agentglass] refusing non-local server %r\n" % url)
          return False
      return True
  
  
  def _opencode_config_dir():
      xdg = os.environ.get("XDG_CONFIG_HOME")
      base = Path(xdg) if xdg else Path.home() / ".config"
      return base / "opencode"
  
  
  def _backup(path: Path) -> None:
      if path.exists():
          bak = path.with_name(path.name + f".bak.agentglass.{time.strftime('%Y%m%d-%H%M%S')}")
          shutil.copy2(path, bak)
          print(f"[agentglass] backup -> {bak}")
  
  
  def _is_agentglass_plugin(path: Path) -> bool:
      return path.read_text().startswith(PLUGIN_MARKER)
  
  
  def opencode_installed() -> bool:
      cfg = _opencode_config_dir()
      return cfg.exists() or shutil.which("opencode") is not None
  
  
  def wire_plugin(undo: bool) -> bool:
      cfg = _opencode_config_dir()
      dest = cfg / "plugins" / PLUGIN_FILENAME
  
      if undo:
          if dest.exists():
              if not _is_agentglass_plugin(dest):
                  print(f"[agentglass] leaving unrelated opencode plugin at {dest}")
                  return True
              _backup(dest)
              dest.unlink()
              print(f"[agentglass] removed opencode plugin ({dest})")
          else:
              print("[agentglass] opencode plugin not found — nothing to undo.")
          return True
  
      if not PLUGIN_SRC.exists():
          print(f"[agentglass] plugin source not found: {PLUGIN_SRC}")
          return False
  
      cfg.mkdir(parents=True, exist_ok=True)
      (cfg / "plugins").mkdir(parents=True, exist_ok=True)
  
      if dest.exists():
          if not _is_agentglass_plugin(dest):
              print(f"[agentglass] refusing to replace unrelated opencode plugin at {dest}")
              return False
          existing = dest.read_text()
          new = PLUGIN_SRC.read_text()
          if existing == new:
              print("[agentglass] opencode plugin already installed — nothing to do.")
              return True
          _backup(dest)
  
      shutil.copy2(PLUGIN_SRC, dest)
      print(f"[agentglass] deployed opencode plugin -> {dest}")
      print("[agentglass]   start a new `opencode` session for the plugin to activate.")
      return True
  
  
  def check_server() -> None:
      import urllib.request
      try:
          req = urllib.request.Request(f"{SERVER}/health", method="GET")
          with urllib.request.urlopen(req, timeout=2) as resp:
              if resp.status == 200:
                  print(f"[agentglass] server is reachable at {SERVER}")
              else:
                  print(f"[agentglass] server responded {resp.status} at {SERVER}")
      except Exception:
          print(f"[agentglass] server not reachable at {SERVER} — is agentglass running?")
  
  
  def main() -> int:
      ap = argparse.ArgumentParser(description="Connect opencode to agentglass via plugin.")
      ap.add_argument("--undo", action="store_true", help="remove the agentglass opencode plugin")
      ap.add_argument("--postinstall", action="store_true", help="lifecycle mode: honour AGENTGLASS_NO_OPENCODE, never fail")
      args = ap.parse_args()
      if not args.undo and not _agentglass_local_only(SERVER):
          return 0 if args.postinstall else 1
  
      if args.postinstall and os.environ.get("AGENTGLASS_NO_OPENCODE"):
          print("[agentglass] AGENTGLASS_NO_OPENCODE set — skipping opencode auto-connect.")
          return 0
  
      if not opencode_installed():
          if not args.postinstall:
              print("[agentglass] opencode not detected (looked for ~/.config/opencode/ or `opencode` in PATH).")
          return 0
  
      try:
          if not wire_plugin(args.undo):
              return 0 if args.postinstall else 1
      except Exception as e:
          print(f"[agentglass] opencode auto-connect skipped ({e}).")
          return 0 if args.postinstall else 1
  
      if not args.undo:
          check_server()
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• hooks/connect_otel.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Auto-connect other AI-agent CLIs to agentglass via OpenTelemetry.
  
  agentglass exposes an OTLP/HTTP trace receiver (JSON + protobuf) at
  `<server>/v1/traces`. Any tool that emits OpenTelemetry GenAI (`gen_ai.*`) trace
  spans can stream in — this script wires the ones that do, on install, the same
  way `install_hooks.py` wires Claude Code. It's idempotent, backs up before it
  writes, and never fails the install.
  
    python3 hooks/connect_otel.py               # detect + wire installed agent CLIs
    python3 hooks/connect_otel.py --undo        # unwire them again
    python3 hooks/connect_otel.py --postinstall # lifecycle mode (honors AGENTGLASS_NO_OTEL)
  
  Currently wired automatically:
    * Gemini CLI       → ~/.gemini/settings.json   (OTLP traces → /v1/traces)
    * OpenAI Codex CLI → ~/.codex/config.toml       (OTLP logs   → /v1/logs)
  """
  import argparse
  import json
  import os
  import shutil
  import sys
  import time
  from pathlib import Path
  
  def _default_server(host):
      """http://<host>:<port> of the running server. The port is the one it wrote
      at bind (next to the token file), because the app moves to another port when
      4000 is taken and the AGENTGLASS_PORT an agent pane inherits can be stale;
      then AGENTGLASS_PORT, then 4000. No probe: a hook cannot pay a round trip."""
      base = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
      port = ""
      try:
          with open(os.path.join(base, "agentglass", "port"), encoding="utf-8") as f:
              port = f.read().strip()
      except OSError:
          pass
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = os.environ.get("AGENTGLASS_PORT", "").strip()
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = "4000"
      return f"http://{host}:{port}"
  
  
  SERVER = (os.environ.get("AGENTGLASS_SERVER") or _default_server("127.0.0.1")).rstrip("/")
  
  def _agentglass_local_only(url):
      """Refuse to send transcript/telemetry anywhere but this machine.
      AGENTGLASS_SERVER is attacker-influenceable (a repo-local settings.json can
      set it), and the payloads carry full session content. Opt out explicitly
      with AGENTGLASS_ALLOW_REMOTE=1 if you really run the server elsewhere."""
      import os
      from urllib.parse import urlparse, urlunparse
      # Exactly "1": a truthy test would let AGENTGLASS_ALLOW_REMOTE=0 — which
      # reads as "off" to every person who writes it — switch the guard off
      # instead of on. So would "false", "no" and "off".
      if os.environ.get("AGENTGLASS_ALLOW_REMOTE") == "1":
          return url
      u = urlparse(url or "")
      if u.scheme not in ("http", "https") or (u.hostname or "") not in ("localhost", "127.0.0.1", "::1"):
          import sys
          sys.stderr.write("[agentglass] refusing non-local server %r\n" % url)
          sys.exit(0)
      # `localhost` is still allowed above — it is this machine, which is the only
      # thing the guard is about. It is rewritten because of what it costs: the
      # server binds IPv4-only, and on a host whose resolver answers ::1 first
      # every event pays a refused IPv6 connect before falling back. That is
      # microseconds on most machines and seconds on some. Rewriting here rather
      # than only in the default means an install that already wrote `localhost`
      # into its settings.json gets the fix without re-running setup.
      if u.hostname == "localhost":
          netloc = "127.0.0.1" + (":%d" % u.port if u.port else "")
          url = urlunparse(u._replace(netloc=netloc))
      return url
  
  
  
  def _backup(path: Path) -> None:
      if path.exists():
          bak = path.with_name(path.name + f".bak.agentglass.{time.strftime('%Y%m%d-%H%M%S')}")
          shutil.copy2(path, bak)
          print(f"[agentglass] backup → {bak}")
  
  
  def _load(path: Path) -> dict:
      if not path.exists():
          return {}
      try:
          return json.loads(path.read_text())
      except Exception:
          # Returning {} here made the caller write the telemetry block over a
          # person's own settings (a comment in the file is enough to land here).
          raise ValueError(f"{path} is not valid JSON, leaving it as it is")
  
  
  def _write(path: Path, data: dict) -> None:
      path.parent.mkdir(parents=True, exist_ok=True)
      path.write_text(json.dumps(data, indent=2) + "\n")
  
  
  # --- Gemini CLI ------------------------------------------------------------
  GEMINI_SETTINGS = Path.home() / ".gemini" / "settings.json"
  
  
  def gemini_installed() -> bool:
      return GEMINI_SETTINGS.parent.exists() or shutil.which("gemini") is not None
  
  
  def wire_gemini(undo: bool) -> bool:
      if not gemini_installed():
          return False
      cfg = _load(GEMINI_SETTINGS)
      tel = cfg.get("telemetry") if isinstance(cfg.get("telemetry"), dict) else {}
      if undo:
          # Only unwire if it still points at us; leave a user's own setup alone.
          if tel.get("otlpEndpoint") == SERVER:
              _backup(GEMINI_SETTINGS)
              for k in ("enabled", "traces", "otlpEndpoint", "otlpProtocol"):
                  tel.pop(k, None)
              cfg["telemetry"] = tel
              _write(GEMINI_SETTINGS, cfg)
              print(f"[agentglass] unwired Gemini CLI ({GEMINI_SETTINGS})")
          return True
      if tel.get("otlpEndpoint") == SERVER and tel.get("traces") and tel.get("enabled"):
          print("[agentglass] Gemini CLI already connected — nothing to do.")
          return True
      _backup(GEMINI_SETTINGS)
      tel.update({"enabled": True, "traces": True, "otlpEndpoint": SERVER, "otlpProtocol": "http"})
      cfg["telemetry"] = tel
      _write(GEMINI_SETTINGS, cfg)
      print(f"[agentglass] connected Gemini CLI → {SERVER}/v1/traces ({GEMINI_SETTINGS})")
      print("[agentglass]   start a new `gemini` session for it to take effect.")
      return True
  
  
  # --- OpenAI Codex CLI (OTLP logs) ------------------------------------------
  CODEX_CONFIG = Path.home() / ".codex" / "config.toml"
  CODEX_MARK = "# agentglass — OTLP logs → agentglass"
  CODEX_BLOCK = (
      f"\n{CODEX

• hooks/gate_event.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """agentglass control-plane gate (OPT-IN).
  
  A PreToolUse hook that holds a tool call until you approve or deny it from the
  agentglass dashboard. Point a project's PreToolUse hook at this to gate its
  tools remotely.
  
      python3 hooks/gate_event.py --source-app my-project
  
  Safety by design — it NEVER blocks your agents by accident:
    * if agentglass is unreachable or errors → allow (exit 0, no output)
    * if no one decides within the timeout → the server auto-allows
    * only sessions wired to this hook are gated; everything else is untouched
  
  How long it waits is per hook entry, because patience is not one number: a
  `Bash` matcher gating `rm -rf` is worth standing up for, a file write probably
  isn't. Each matcher in settings.json carries its own command line, so `--timeout`
  on that line is the per-matcher setting.
  
      { "matcher": "Bash", "hooks": [{ "type": "command", "command":
        "python3 hooks/gate_event.py --source-app my-project --timeout 900" }] }
  
  The server clamps what it is asked for, and its ceiling is the larger of 300s and
  its own AGENTGLASS_GATE_TIMEOUT — so a matcher wanting longer than five minutes
  needs that raised on the server too, or it is quietly held for five.
  
  Durable across a server restart: the hook picks the request id, so if the
  connection drops mid-wait (agentglass restarted, a crash, a proxy hanging up)
  it re-attaches to that same request instead of giving up and falling into the
  timeout branch. It only gives up once its own deadline has passed.
  
  Deny/allow are returned to Claude Code via the PreToolUse permissionDecision.
  
  Env:
      AGENTGLASS_SERVER   server base url (default http://127.0.0.1:4000)
      AGENTGLASS_GATE_TIMEOUT  seconds to wait for a human (default 300)
      AGENTGLASS_GATE_FAILCLOSED  "1" → an unreachable agentglass DENIES the call
          instead of allowing it. Off by default; with it on, agentglass being
          down blocks every gated call.
  """
  import argparse
  import json
  import os
  import sys
  import time
  import urllib.error
  import urllib.parse
  import urllib.request
  import uuid
  
  def _default_server(host):
      """http://<host>:<port> of the running server. The port is the one it wrote
      at bind (next to the token file), because the app moves to another port when
      4000 is taken and the AGENTGLASS_PORT an agent pane inherits can be stale;
      then AGENTGLASS_PORT, then 4000. No probe: a hook cannot pay a round trip."""
      base = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
      port = ""
      try:
          with open(os.path.join(base, "agentglass", "port"), encoding="utf-8") as f:
              port = f.read().strip()
      except OSError:
          pass
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = os.environ.get("AGENTGLASS_PORT", "").strip()
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = "4000"
      return f"http://{host}:{port}"
  
  
  DEFAULT_SERVER = (os.environ.get("AGENTGLASS_SERVER") or _default_server("127.0.0.1"))
  
  def _agentglass_local_only(url):
      """Refuse to send transcript/telemetry anywhere but this machine.
      AGENTGLASS_SERVER is attacker-influenceable (a repo-local settings.json can
      set it), and the payloads carry full session content. Opt out explicitly
      with AGENTGLASS_ALLOW_REMOTE=1 if you really run the server elsewhere."""
      import os
      from urllib.parse import urlparse, urlunparse
      # Exactly "1": a truthy test would let AGENTGLASS_ALLOW_REMOTE=0 — which
      # reads as "off" to every person who writes it — switch the guard off
      # instead of on. So would "false", "no" and "off".
      if os.environ.get("AGENTGLASS_ALLOW_REMOTE") == "1":
          return url
      u = urlparse(url or "")
      if u.scheme not in ("http", "https") or (u.hostname or "") not in ("localhost", "127.0.0.1", "::1"):
          import sys
          sys.stderr.write("[agentglass] refusing non-local server %r\n" % url)
          # Fail-closed means an answer that cannot be had is a deny. This address
          # can come from a repo's own settings.json, so allowing here would let
          # a repo switch the flag off.
          if FAIL_CLOSED:
              emit("deny", "agentglass is configured fail-closed and its address is not this machine, so this call was blocked without a human seeing it. Report it rather than working around it.")
          sys.exit(0)
      # `localhost` is still allowed above — it is this machine, which is the only
      # thing the guard is about. It is rewritten because of what it costs: the
      # server binds IPv4-only, and on a host whose resolver answers ::1 first
      # every event pays a refused IPv6 connect before falling back. That is
      # microseconds on most machines and seconds on some. Rewriting here rather
      # than only in the default means an install that already wrote `localhost`
      # into its settings.json gets the fix without re-running setup.
      if u.hostname == "localhost":
          netloc = "127.0.0.1" + (":%d" % u.port if u.port else "")
          url = urlunparse(u._replace(netloc=netloc))
      return url
  
  def _shared_secret():
      """The token the server is running with, or "" when it has none.
  
      The environment first, then the file — the same order and the same file as
      hooks/statusline.sh, for the same reason. A hook inherits the environment of
      whatever launched Claude Code, which for a desktop icon or a terminal opened
      outside agentglass is nothing at all. Since the desktop app runs its sidecar
      with a token even on loopback (electron/main.js), reading only the
      environment would leave every gate post unauthenticated: /gate answers 401,
      the HTTPError branch below cancels the retry loop, and the call lands in
      fail-open. The gate would stop holding, silently, with nothing on screen to
      say it had.
  
      0600 and owned by this user, so being able to read it is the check. An
      unreadable file is not an error — a server started without a token wants no
      header at all, and

• hooks/install_hooks.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """agentglass hook installer.
  
  Wires the agentglass event forwarder into your Claude Code settings so every
  session streams to the dashboard — no hand-copying required. Zero third-party
  deps (stdlib only).
  
      python3 hooks/install_hooks.py               # install into ~/.claude/settings.json (global)
      python3 hooks/install_hooks.py --uninstall   # remove the agentglass hooks
      python3 hooks/install_hooks.py --project .   # install into <project>/.claude/settings.json instead
      python3 hooks/install_hooks.py --postinstall # lifecycle mode used by `bun install`
  
  Notes:
    * Idempotent — re-running re-points the send_event.py path in place and never
      duplicates entries or disturbs your other hooks (magia, guards, etc.).
    * It also takes the `statusLine` slot, CHAINING whatever was there: Claude Code
      pipes `rate_limits` to that command on every turn, which is the plan usage
      the dashboard would otherwise pay a rate-limited endpoint for. Your own
      status line is passed the same stdin, owns the output, and comes back
      untouched on --uninstall. POSIX only; skipped on Windows.
    * The target settings file is backed up (`*.bak.agentglass.<timestamp>`) before
      any change, and only when there is actually a change to make.
    * `--source-app` is intentionally omitted so each project auto-labels in the
      dashboard by its own working-directory name (send_event.py defaults to the
      cwd basename).
    * `--postinstall` respects `AGENTGLASS_NO_HOOKS=1` (skips) and never fails the
      install, so `bun install` stays green even without Python or write access.
  """
  import argparse
  import json
  import os
  import re
  import shlex
  import shutil
  import sys
  import time
  
  HOOKS_DIR = os.path.dirname(os.path.abspath(__file__))
  SEND_EVENT = os.path.join(HOOKS_DIR, "send_event.py")
  MARKER = "send_event.py"  # substring that identifies a hook command as ours
  
  GATE = os.path.join(HOOKS_DIR, "gate_event.py")
  GATE_MARKER = "gate_event.py"
  # The gate is a SEPARATE switch, and never a side effect of installing the
  # forwarder.
  #
  # The forwarder is telemetry, and the rule beside `hook_command` is that
  # telemetry must never be able to stop a tool call — that is what its trailing
  # `|| exit 0` buys. The gate is the opposite on purpose: it holds a tool call
  # until a person decides, and an outward one (a push, a pull request, a
  # comment, a review, a merge, a ticket, a message in a channel) is held closed.
  #
  # So it has its own flag here and its own button in the app, and no trailing
  # `|| exit 0`: swallowing the exit status is exactly what would turn a gate back
  # into telemetry. `gate_event.py` already allows on every failure it controls —
  # an unreachable server, a timeout, an answer it cannot read — so the honest
  # failure mode lives inside the script instead of being bolted on outside it.
  # The one thing outside it is the script not existing, which `gate_command`
  # guards with a file test.
  
  STATUSLINE = os.path.join(HOOKS_DIR, "statusline.sh")
  # Our status line, recognised by SHAPE rather than by a filename substring.
  #
  # This was `"statusline.sh" in cmd`, and the collision it caused is the kind
  # that eats somebody's config in silence: a status line of their own called
  # `mi-statusline.sh`, `custom-statusline.sh` or the widely used
  # `ccstatusline.sh` all CONTAIN "statusline.sh". The installer read them as
  # ours, tried to unwrap a command that was never wrapped, found no third
  # argument, and dropped it. Nothing failed and nothing was said; their status
  # line was simply gone after installing.
  #
  # The path is not usable as the marker either — the repo moves, and an install
  # from an older checkout has to stay recognisable. So: our command always
  # begins `sh "…/hooks/statusline.sh"`, and that shape is what is matched.
  SL_RE = re.compile(r'^sh\s+"(?:.*/)?hooks/statusline\.sh"')
  
  
  def _is_ours_statusline(cmd):
      return bool(cmd) and SL_RE.match(cmd.strip()) is not None
  
  # event -> (matcher or None, attach transcript for token/cost)
  EVENTS = {
      "SessionStart":     (None, False),
      "UserPromptSubmit": (None, False),
      "PreToolUse":       ("*",  False),
      "PostToolUse":      ("*",  False),
      "Notification":     (None, False),
      "SubagentStop":     (None, True),
      "Stop":             (None, True),
      "PreCompact":       (None, False),
      "SessionEnd":       (None, True),
  }
  
  
  def settings_path(project):
      base = os.path.join(project, ".claude") if project else os.path.expanduser("~/.claude")
      return os.path.join(base, "settings.json")
  
  
  def _is_ours(entry):
      return any(MARKER in h.get("command", "") for h in entry.get("hooks", []))
  
  
  def load(path):
      if not os.path.exists(path):
          return {}
      with open(path, "r", encoding="utf-8") as f:
          raw = f.read().strip()
      return json.loads(raw) if raw else {}
  
  
  def _hook_python():
      """Interpreter name written into Claude Code hook commands.
  
      On Windows most installs expose `py` (launcher) and/or `python`, not
      `python3`. Prefer those at install time. Deliberately avoid
      `sys.executable` so a later-deleted venv does not break every hook.
      """
      if os.name != "nt":
          return "python3"
      for candidate in ("py", "python"):
          if shutil.which(candidate):
              return candidate
      return "py"
  
  
  def _statusline_command(chained):
      """Our forwarder, told what it is wrapping.
  
      The wrapped command travels as a quoted argument rather than being stashed
      somewhere out of sight: it is then visible in the same settings.json the
      user is reading, and uninstall has everything it needs to put things back
      without consulting any state of ours.
      """
      cmd = 'sh "%s"' % STATUSLINE
      return cmd + " " + shlex.quote(chained) if chained else cmd
  
  
  def _chained_from(cmd):
      """What our wrapper was told to call, so re-installing re-points the script
      path without forgetting the status line it wraps."""
      if not _is_ours_statusline(cmd):
          return N

• hooks/opencode-plugin.jsGitHub ↗
  Read the script

  // agentglass opencode plugin
  // Forwards opencode session events to the agentglass /ingest endpoint.
  //
  // Install:  python3 hooks/connect_opencode.py
  // Manual:   cp hooks/opencode-plugin.js ~/.config/opencode/plugins/agentglass.js
  //
  // Env:
  //   AGENTGLASS_SERVER        server base url (default: the port the running server wrote, else AGENTGLASS_PORT, else 4000)
  //   AGENTGLASS_URL           deprecated alias for AGENTGLASS_SERVER
  //   AGENTGLASS_ALLOW_REMOTE  set to 1 to allow a non-local server
  //   AGENTGLASS_INTERNAL      set to skip (prevents re-ingestion loops)
  
  import { readFileSync } from "node:fs";
  import { homedir } from "node:os";
  import { join } from "node:path";
  
  export function normalizeServer(url) {
    return url.replace(/\/+$/, "");
  }
  
  /**
   * Rewrite a `localhost` host to `127.0.0.1`, leaving everything else alone.
   *
   * `localhost` stays allowed by isAllowedServer — it is this machine, which is
   * all that guard is about. It is rewritten because the server binds IPv4-only,
   * so on a host whose resolver answers ::1 first, every event pays a refused
   * connect before falling back. Same reasoning, and the same rewrite, as the
   * Python hooks' `_agentglass_local_only`.
   */
  export function preferIpv4(url) {
    try {
      const u = new URL(url);
      if (u.hostname !== "localhost") return url;
      u.hostname = "127.0.0.1";
      return u.toString().replace(/\/+$/, "");
    } catch {
      return url;
    }
  }
  
  /**
   * The port the running server wrote at bind (next to the token file), then
   * AGENTGLASS_PORT, then 4000. The app moves to another port when 4000 is taken
   * and the AGENTGLASS_PORT an agent pane inherits can be stale; no probe, since
   * an event must not pay a round trip.
   */
  export function localPort(env = process.env) {
    const ok = (v) => /^\d+$/.test(v) && Number(v) > 0 && Number(v) < 65536;
    const base = env.XDG_CONFIG_HOME || join(env.HOME || homedir(), ".config");
    let port = "";
    try { port = readFileSync(join(base, "agentglass", "port"), "utf8").trim(); } catch { /* none written */ }
    if (!ok(port)) port = String(env.AGENTGLASS_PORT || "").trim();
    return ok(port) ? port : "4000";
  }
  
  export function resolveServer(env = process.env) {
    return preferIpv4(normalizeServer(
      env.AGENTGLASS_SERVER || env.AGENTGLASS_URL || `http://127.0.0.1:${localPort(env)}`,
    ));
  }
  
  const SERVER = resolveServer();
  const INGEST_PATH = "/ingest";
  
  const DEFAULT_SESSION_RE = /^(New session - |Child session - )\d{4}-\d{2}-\d{2}T/;
  const MAX_TOOL_OUTPUT = 256 * 1024;
  
  export function limitToolOutput(value) {
    if (typeof value !== "string" || value.length <= MAX_TOOL_OUTPUT) return value;
    return `${value.slice(0, MAX_TOOL_OUTPUT)}\n[output truncated by agentglass]`;
  }
  
  export function isAllowedServer(url, allowRemote = false) {
    if (allowRemote) return true;
    try {
      const parsed = new URL(url);
      return (
        (parsed.protocol === "http:" || parsed.protocol === "https:") &&
        ["localhost", "127.0.0.1", "[::1]"].includes(parsed.hostname)
      );
    } catch {
      return false;
    }
  }
  
  function sessionNameFromInfo(info) {
    if (!info?.title || DEFAULT_SESSION_RE.test(info.title)) return undefined;
    return info.title;
  }
  
  async function postEvent(body) {
    if (!isAllowedServer(SERVER, process.env.AGENTGLASS_ALLOW_REMOTE === "1")) {
      return false;
    }
    try {
      const res = await fetch(`${SERVER}${INGEST_PATH}`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(3000),
      });
      return res.ok;
    } catch {
      return false;
    }
  }
  
  /**
   * The agent rules' verdict for one call (GET /agent-rules/check), or null.
   * OpenCode's tool names are lower case (`bash`, `edit`); the rules speak Claude
   * Code's (`Bash`, `Edit`). A plugin cannot open a prompt, so an "ask" is refused
   * with the reason, which is the stricter of the two things it could do. An
   * unreachable server passes, said once on stderr.
   */
  async function rulesVerdict(tool, args, dir) {
    if (!dir || !dir.startsWith("/") || !isAllowedServer(SERVER, process.env.AGENTGLASS_ALLOW_REMOTE === "1")) return null;
    const name = String(tool || "");
    const claudeName = name ? name[0].toUpperCase() + name.slice(1) : "";
    if (!/^[A-Za-z][A-Za-z0-9_-]{0,99}$/.test(claudeName)) return null;
    const q = new URLSearchParams({ root: dir, tool: claudeName, input: JSON.stringify(args ?? {}).slice(0, 8000) });
    // The seat's own credential names it to the server, so a rule scoped to the
    // seat applies on this CLI too; a worker runs without one.
    const token = (process.env.AGENTGLASS_TOKEN || "").trim();
    try {
      const res = await fetch(`${SERVER}/agent-rules/check?${q}`, {
        signal: AbortSignal.timeout(2000),
        ...(token ? { headers: { Authorization: `Bearer ${token}` } } : {}),
      });
      if (!res.ok) return null;
      const out = await res.json();
      return out.decision === "never" || out.decision === "ask" ? out : null;
    } catch (e) {
      console.error(`[agentglass] rules check skipped, server unreachable (${e?.message ?? e}); this call was not checked`);
      return null;
    }
  }
  
  function sourceAppFromDir(dir) {
    if (!dir) return "opencode";
    const parts = dir.replace(/\\/g, "/").split("/");
    return parts[parts.length - 1] || "opencode";
  }
  
  function sessionIDFromProps(properties) {
    return typeof properties?.sessionID === "string" && properties.sessionID
      ? properties.sessionID
      : undefined;
  }
  
  function modelLabel(info) {
    if (!info) return undefined;
    const pid = info.model?.providerID || info.providerID;
    const mid = info.model?.modelID || info.modelID;
    if (pid && mid) return `${pid}/${mid}`;
    if (mid) return mid;
    return undefined;
  }
  
  function usageFromAssistant(info) {
    const t = info?.tokens;
    if (!t) return undefined;
    const usage = {
      input_tokens: t.input || 0,
      output_tokens: t.output || 0,
    };
    if (t.cache?.read) usage.cache_read_input_tokens = t.cache.read;
    if (t.cache?.write)

• hooks/seed_demo.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Seed the dashboard with a burst of realistic demo events (no Claude needed).
  
      python3 hooks/seed_demo.py            # steady stream for ~30s
      python3 hooks/seed_demo.py --once     # one batch and exit
  """
  import argparse
  import json
  import os
  import random
  import time
  import urllib.request
  
  def _default_server(host):
      """http://<host>:<port> of the running server. The port is the one it wrote
      at bind (next to the token file), because the app moves to another port when
      4000 is taken and the AGENTGLASS_PORT an agent pane inherits can be stale;
      then AGENTGLASS_PORT, then 4000. No probe: a hook cannot pay a round trip."""
      base = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
      port = ""
      try:
          with open(os.path.join(base, "agentglass", "port"), encoding="utf-8") as f:
              port = f.read().strip()
      except OSError:
          pass
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = os.environ.get("AGENTGLASS_PORT", "").strip()
      if not (port.isdigit() and 0 < int(port) < 65536):
          port = "4000"
      return f"http://{host}:{port}"
  
  
  SERVER = (os.environ.get("AGENTGLASS_SERVER") or _default_server("127.0.0.1"))
  
  def _agentglass_local_only(url):
      """Refuse to send transcript/telemetry anywhere but this machine.
      AGENTGLASS_SERVER is attacker-influenceable (a repo-local settings.json can
      set it), and the payloads carry full session content. Opt out explicitly
      with AGENTGLASS_ALLOW_REMOTE=1 if you really run the server elsewhere."""
      import os
      from urllib.parse import urlparse, urlunparse
      # Exactly "1": a truthy test would let AGENTGLASS_ALLOW_REMOTE=0 — which
      # reads as "off" to every person who writes it — switch the guard off
      # instead of on. So would "false", "no" and "off".
      if os.environ.get("AGENTGLASS_ALLOW_REMOTE") == "1":
          return url
      u = urlparse(url or "")
      if u.scheme not in ("http", "https") or (u.hostname or "") not in ("localhost", "127.0.0.1", "::1"):
          import sys
          sys.stderr.write("[agentglass] refusing non-local server %r\n" % url)
          sys.exit(0)
      # `localhost` is still allowed above — it is this machine, which is the only
      # thing the guard is about. It is rewritten because of what it costs: the
      # server binds IPv4-only, and on a host whose resolver answers ::1 first
      # every event pays a refused IPv6 connect before falling back. That is
      # microseconds on most machines and seconds on some. Rewriting here rather
      # than only in the default means an install that already wrote `localhost`
      # into its settings.json gets the fix without re-running setup.
      if u.hostname == "localhost":
          netloc = "127.0.0.1" + (":%d" % u.port if u.port else "")
          url = urlunparse(u._replace(netloc=netloc))
      return url
  
  APPS = ["api-refactor", "docs-agent", "test-writer", "migration"]
  MODELS = ["claude-opus-5", "claude-sonnet-5", "claude-haiku-4-5"]
  TOOLS = ["Bash", "Read", "Edit", "Grep", "Write", "WebFetch", "Task"]
  _seed = random.Random(7)
  
  
  def post(body):
      data = json.dumps(body).encode()
      req = urllib.request.Request(SERVER.rstrip("/") + "/ingest", data=data,
                                   headers={"Content-Type": "application/json"}, method="POST")
      try:
          urllib.request.urlopen(req, timeout=2).read()
      except Exception as e:
          print("post failed:", e)
  
  
  def session():
      app = _seed.choice(APPS)
      sid = f"{app}-{_seed.randint(1000,9999)}"
      model = _seed.choice(MODELS)
      post({"source_app": app, "session_id": sid, "hook_event_type": "SessionStart", "model_name": model})
      post({"source_app": app, "session_id": sid, "hook_event_type": "UserPromptSubmit", "model_name": model,
            "payload": {"prompt": "do the thing"}})
      for _ in range(_seed.randint(3, 8)):
          tool = _seed.choice(TOOLS)
          tid = f"t{_seed.randint(0,99999)}"
          post({"source_app": app, "session_id": sid, "hook_event_type": "PreToolUse", "model_name": model,
                "payload": {"tool_name": tool, "tool_use_id": tid}})
          time.sleep(_seed.uniform(0.05, 0.4))
          err = _seed.random() < 0.12
          post({"source_app": app, "session_id": sid,
                "hook_event_type": "PostToolUseFailure" if err else "PostToolUse", "model_name": model,
                "payload": {"tool_name": tool, "tool_use_id": tid, **({"error": "command failed"} if err else {})}})
      # final Stop with cumulative token usage
      itok, otok = _seed.randint(3000, 40000), _seed.randint(1000, 15000)
      post({"source_app": app, "session_id": sid, "hook_event_type": "Stop", "model_name": model,
            "chat": [{"message": {"model": model, "usage": {
                "input_tokens": itok, "output_tokens": otok,
                "cache_read_input_tokens": _seed.randint(0, 60000)}}}]})
  
  
  def main():
      ap = argparse.ArgumentParser()
      ap.add_argument("--once", action="store_true")
      ap.add_argument("--seconds", type=int, default=30)
      args = ap.parse_args()
      global SERVER
      SERVER = _agentglass_local_only(SERVER)
      print(f"seeding → {SERVER}")
      if args.once:
          session()
          return
      end = time.time() + args.seconds
      while time.time() < end:
          session()
          time.sleep(_seed.uniform(0.3, 1.2))
      print("done")
  
  
  if __name__ == "__main__":
      main()
  

• hooks/send_event.pyGitHub ↗
• hooks/statusline.shGitHub ↗

All 8 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.