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
  
  SERVER = os.environ.get("AGENTGLASS_SERVER", "http://127.0.0.1:4000").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
  
  SERVER = os.environ.get("AGENTGLASS_SERVER", "http://127.0.0.1:4000").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:
      try:
          return json.loads(path.read_text()) if path.exists() else {}
      except Exception:
          return {}
  
  
  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_MARK}\n"
      f"[otel]\n"
      f'exporter = {{ otlp-http = {{ endpoint = "{SERVER}/v1/logs", protocol = "binary" }} }}\n'
  )
  
  
  def codex_installed() -> bool:
      return CODEX_CONFIG.parent.exists() or shutil.which("codex") is not None
  
  
  def wire_codex(undo: bool) -> bool:
      if not codex_installed():
          return False
      text = CODEX_CONFIG.read_text() if CODEX_CONFIG.exists() else ""
      if undo:
          if CODEX_BLOCK in text:
              _backup(CODEX_CONFIG)
              CODEX_CONFIG.write_text(text.replace(CODEX_BLOCK, ""))
              print(f"[agentglass] unwired Codex CLI ({CODEX_CONFIG})")
          return True
      if f"{SERVER}/v1/logs" in text:
          print("[agentglass] Codex CLI already connected — nothing to do.")
          return True
      if "[otel]" in text:
          # Respect a config the user already wrote; just tell them where to point.
          print(f"[agentglass] Codex CLI has its own [otel] config — leaving it. Point its "
                f"otlp-http endpoint at {SERVER}/v1/logs to stream here.")
          return True
      _backup(CODEX_CONFIG)
      CODEX_CONFIG.pa

• 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
  
  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 60)
  """
  import argparse
  import json
  import os
  import sys
  import time
  import urllib.error
  import urllib.parse
  import urllib.request
  import uuid
  
  DEFAULT_SERVER = os.environ.get("AGENTGLASS_SERVER", "http://127.0.0.1:4000")
  
  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 _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 sending one it never asked for changes nothing.
      """
      from_env = os.environ.get("AGENTGLASS_TOKEN", "").strip()
      if from_env:
          return from_env
      config_home = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
      try:
          with open(os.path.join(config_home, "agentglass", "token"), encoding="utf-8") as fh:
              return fh.read().strip()
      except OSError:
          return ""
  
  
  TIMEOUT = int(os.environ.get("AGENTGLASS_GATE_TIMEOUT", "60"))
  # Default is fail-open: if agentglass is unreachable, allow (never block agents
  # by accident). Set this to invert it — an unreachable control plane DENIES the
  # tool call. Opt-in, because with agentglass down every gated call is blocked.
  FAIL_CLOSED = os.environ.get("AGENTGLASS_GATE_FAILCLOSED") == "1"
  
  
  def allow_silently():
      # No output + exit 0 → Claude Code proceeds as normal (default behaviour).
      sys.exit(0)
  
  
  def emit(decision: str, reason: str):
      # Explicit PreToolUse decision. "deny" blocks the tool; "allow" approves it.
      print(json.dumps({
          "hookSpecificOutput": {
              "hookEventName": "PreToolUse",
              "permissionDecision": decision,
              "permissionDecisionReason": reason,
          }
      }))
      sys.exit(0)
  
  
  def main():
      # agentglass's own internal `claude` calls bypass the gate (allow silently).
      if os.environ.get("AGENTGLASS_INTERNAL"):
          allow_silently()
      ap = argparse.ArgumentParser()
      ap.add_argument("--source-app", default=os.path.basename(os.getcwd()))
      ap.add_argument("--server", default=DEFAULT_SERVER)
      args = ap.parse_args()
      server = _agentglass_local_only(getattr(args, "server", None) or DEFAULT_SERVER)
  
      try:
          raw = sys.stdin.read()
          payload = json.loads(raw) if raw.strip() else {}
      except json.JSONDecodeError:
          allow_silently()
  
      # The id is ours, not the server's. That is what makes a dropped connection
      # recoverable: without it there is no name for the request we were waiting
      # on, and a restart mid-wait can only be read as "no answer".
      gate_id = str(uuid.uuid4()

• 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
  
  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 None
      try:
          parts = shlex.split(cmd)
      except ValueError:
          return None
      return parts[2] if len(parts) >= 3 else None
  
  
  def install_statusline(cfg):
      """Take the statusLine slot, chaining whatever was there.
  
      Unlike hooks, statusLine is a single slot rather than a list, so there is no
      way to add to it without owning it. Owning it is only acceptable because the
      previous command is called on every render with the same stdin and owns the
      output — see hooks/statusline.sh, where every path out still runs the chain.
  
      Skipped on Windows: the forwarder is POSIX sh, and a status line that fails
      is one the user sees on every single render. The usage endpoint poll keeps
      working there, just without the free feed.
      """
      if os.name == "nt":
          return
      if not os.path.exists(STATUSLINE):
          return
      current = cfg.get("statusLine")
      cur_cmd = current.get("command") if isinstance(current, dict) else None
      # Already ours: re-point the script path, keep what it wraps.
      chai

• 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 http://127.0.0.1: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)
  
  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;
    }
  }
  
  export function resolveServer(env = process.env) {
    return preferIpv4(normalizeServer(
      env.AGENTGLASS_SERVER || env.AGENTGLASS_URL || "http://127.0.0.1:4000",
    ));
  }
  
  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;
    }
  }
  
  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) usage.cache_creation_input_tokens = t.cache.write;
    if (usage.input_tokens + usage.output_tokens === 0) return undefined;
    return usage;
  }
  
  export const AgentGlassPlugin = async ({ directory, worktree }) => {
    if (process.env.AGENTGLASS_INTERNAL) return {};
  
    const sourceApp = sourceAppFromDir(worktree || directory);
    const childSessions = new Set();
    const pendingPreTool = new Map();
    const completedMessages = new Map();
    const completingMessages = new Set();
    const clearSession = (id) => {
      childSessions.delete(id);
      for (const [callID, pre] of pendingPreTool) if (pre.sessionID === id) pendingPreTool.delete(callID);
      for (const [messageID, sessionID] of completedMessages) if (sessionID === id) completedMessages.delete(messageID);
    };
  
    return {
      "chat.message": async ({ sessionID, agent, model }) => {
        if (sessionID && childSessions.has(sessionID)) return;
  
        const payload = { project_path: worktree || directory };
        if (directory) payload.cwd = directory;
  
        const body = {
          source_app: sourceApp,
          session_id: sessionID || "unknown",
          hook_event_type: "UserPromptSubmit",
          payload,
          model_name: model ? `${model.providerID}/${model.modelID}` : undefined,
        };
        await postEvent(body);
      },
  
      "tool.execute.before": async ({ tool, sessionID, callID }, output) => {
        if (sessionID && childSessions.has(sessionID)) return;
  
        pendingPreTool.set(callID, { timestamp: Date.now(), sessionID });
  
        const payload = {
          tool_name: tool,
          tool_use_id: callID,
          tool_input: output?.args,
        };
        if (worktree || directory) payload.project_path = worktree || directory;
  
        await postEvent({
          source_app: sourceApp,
          session_id: sessionID || "unknown",
          hook_event_type: "PreToolUse",
          payload,
        });
      },
  
      "tool.execute.after": async (
        { tool, sessionID, callID, args },
        output
      ) => {
        if (sessionID && childSessions.has(sessionID)) return;
  
        const pre = pendingPreTool.get(callID);
        pendingPreTool.delete(callID);
  
        const isError = output?.output == null && output?.metadata?.error != null;
        const payload = {
          tool_name: tool,
          tool_use_id: callID,
          tool_input: args,
        };
        if (output?.output != null) {
          payload.tool_response = { content: limitToolOutput(output.

• 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
  
  SERVER = os.environ.get("AGENTGLASS_SERVER", "http://127.0.0.1:4000")
  
  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.