refactor: replace inline hook commands with hook_handler module

Inline python -c one-liners with nested escaping (exec, triple-escaped
URLs) broke on Windows and were fragile to quote. Now hooks invoke
`python -m aictl.hook_handler --event X --port Y` — a proper module
that handles stdin/stdout passthrough and server POST.

_is_aictl_hook detects all three hook generations (module, inline, curl)
so install/uninstall can clean up any legacy format.

Author: Zvi

aictl/commands/integrations.py

@@ -138,19 +138,28 @@ def _settings_path(scope: str) -> Path:
     return claude_global_dir() / "settings.json"
 
 
+_AICTL_HOOK_MARKERS = ("/api/hooks", "aictl.hook_handler")
+
+
 def _is_aictl_hook(hook: dict) -> bool:
     """Return True if the hook entry was installed by aictl.
 
-    Handles both the current nested format {"matcher": ..., "hooks": [...]} and
-    the old flat format {"type": "command", "command": "..."} (for migration).
+    Detects three generations of hook format:
+    - Current: ``python -m aictl.hook_handler --event ...``
+    - Previous: inline ``python -c "... /api/hooks ..."``
+    - Legacy flat: ``curl ... /api/hooks ...``
     """
     if not isinstance(hook, dict):
         return False
-    # Current format: {"matcher": ..., "hooks": [{"type": "command", "command": "..."}]}
+    # Current nested format: {"matcher": ..., "hooks": [{"type": "command", "command": "..."}]}
     if "hooks" in hook and isinstance(hook["hooks"], list):
-        return any("/api/hooks" in str(h.get("command", "")) for h in hook["hooks"])
+        return any(
+            any(m in str(h.get("command", "")) for m in _AICTL_HOOK_MARKERS)
+            for h in hook["hooks"]
+        )
     # Old flat format (pre-fix): {"type": "command", "command": "..."}
-    return "/api/hooks" in str(hook.get("command", ""))
+    cmd = str(hook.get("command", ""))
+    return any(m in cmd for m in _AICTL_HOOK_MARKERS)
 
 
 def _python_cmd() -> str:
@@ -168,9 +177,12 @@ def _python_cmd() -> str:
 def _build_hook_config(port: int, events: list[str] | None, event_map: dict[str, str] | None = None, matcher: str = "") -> dict:
     """Build the hooks configuration dict for AI tools.
 
-    Each hook reads the rich JSON payload from stdin, merges in
-    environment variables, and POSTs everything to aictl.
-    Uses sys.executable instead of 'python3' for cross-platform compatibility.
+    Each hook invokes ``python -m aictl.hook_handler`` which reads the
+    rich JSON payload from stdin, merges environment variables, POSTs
+    everything to aictl, and passes the payload through to stdout.
+
+    Using a module invocation instead of an inline ``-c`` one-liner
+    avoids shell-escaping issues on Windows and complex quoting.
 
     Optional event_map can translate internal HOOK_EVENTS names to tool-specific names.
     matcher: "" for Claude (default), "*" for Gemini.
@@ -179,24 +191,9 @@ def _build_hook_config(port: int, events: list[str] | None, event_map: dict[str,
     hooks: dict[str, list[dict]] = {}
     python = _python_cmd()
 
-    # The hook command: read stdin (Node-based tool's rich JSON), merge env vars, POST to aictl.
-    # We must print the final JSON to stdout for tool continuity.
     for event in target_events:
         tool_event_name = event_map.get(event, event) if event_map else event
-        # Single-line Python command using semicolons for maximum compatibility.
-        # No literal newlines or complex escaping.
-        cmd = (
-            f"{python} -c \""
-            f"import sys,json,os,urllib.request as u; "
-            f"d=json.load(sys.stdin) if not sys.stdin.isatty() else {{}}; "
-            f"d['event']='{event}'; "
-            f"d.setdefault('session_id',os.environ.get('SESSION_ID','')); "
-            f"d.setdefault('cwd',os.environ.get('CWD','')); "
-            f"exec('try: u.urlopen(u.Request(\\\"http://localhost:{port}/api/hooks\\\", "
-            f"json.dumps(d).encode(), {{\\\"Content-Type\\\":\\\"application/json\\\"}}), timeout=2)\\n"
-            f"except: pass'); "  # noqa: S110 — generated hook script must not crash if aictl server is unreachable
-            f"print(json.dumps(d))\""
-        )
+        cmd = f"{python} -m aictl.hook_handler --event {event} --port {port}"
         hooks[tool_event_name] = [{"matcher": matcher, "hooks": [{"type": "command", "command": cmd}]}]
     return hooks
 

aictl/hook_handler.py

@@ -0,0 +1,65 @@
+# aictl - Cross-platform AI Tool Context Control + Dashboard
+# Copyright (c) 2026 Zvi Schneider. MIT License.
+"""Hook handler for AI coding tool integration.
+
+Invoked by AI tools (Claude Code, Gemini CLI) as a hook command.
+Reads JSON payload from stdin, enriches it with event metadata and
+environment variables, POSTs it to the aictl server, and passes the
+payload through to stdout.
+
+Usage:
+    python -m aictl.hook_handler --event SessionStart --port 8484
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import urllib.request
+
+
+def main() -> None:
+    # Minimal arg parsing — no argparse import to keep startup fast.
+    event = ""
+    port = 8484
+    args = sys.argv[1:]
+    i = 0
+    while i < len(args):
+        if args[i] == "--event" and i + 1 < len(args):
+            event = args[i + 1]
+            i += 2
+        elif args[i] == "--port" and i + 1 < len(args):
+            port = int(args[i + 1])
+            i += 2
+        else:
+            i += 1
+
+    # Read payload from stdin (AI tools pipe rich JSON here).
+    if not sys.stdin.isatty():
+        data = json.load(sys.stdin)
+    else:
+        data = {}
+
+    # Enrich with event name and environment context.
+    data["event"] = event
+    data.setdefault("session_id", os.environ.get("SESSION_ID", ""))
+    data.setdefault("cwd", os.environ.get("CWD", ""))
+
+    # POST to aictl server (best-effort — never crash if server is down).
+    try:
+        req = urllib.request.Request(
+            f"http://localhost:{port}/api/hooks",
+            json.dumps(data).encode(),
+            {"Content-Type": "application/json"},
+        )
+        urllib.request.urlopen(req, timeout=2)  # noqa: S310 — localhost only
+    except Exception:  # noqa: BLE001, S110 — must not crash if aictl unreachable
+        pass
+
+    # Pass through to stdout for tool continuity.
+    print(json.dumps(data))
+
+
+if __name__ == "__main__":
+    main()

test/test_hooks.py

@@ -45,44 +45,45 @@ def test_nested_format(self):
     def test_custom_port(self):
         config = _build_hook_config(9999, None)
         cmd = config["SessionStart"][0]["hooks"][0]["command"]
-        assert "localhost:9999" in cmd
+        assert "--port 9999" in cmd
 
     def test_custom_event_subset(self):
         config = _build_hook_config(8484, ["SessionStart", "SessionEnd"])
         assert len(config) == 2
         assert "SessionStart" in config
         assert "SessionEnd" in config
 
-    def test_reads_stdin(self):
+    def test_invokes_hook_handler_module(self):
         config = _build_hook_config(8484, ["PostToolUse"])
         cmd = config["PostToolUse"][0]["hooks"][0]["command"]
-        assert "sys.stdin" in cmd
-
-    def test_posts_to_api_hooks(self):
-        config = _build_hook_config(8484, ["PostToolUse"])
-        cmd = config["PostToolUse"][0]["hooks"][0]["command"]
-        assert "/api/hooks" in cmd
+        assert "aictl.hook_handler" in cmd
 
     def test_sets_event_name(self):
         config = _build_hook_config(8484, ["PreToolUse"])
         cmd = config["PreToolUse"][0]["hooks"][0]["command"]
-        assert "'PreToolUse'" in cmd
+        assert "--event PreToolUse" in cmd
 
-    def test_merges_env_vars(self):
+    def test_no_inline_python(self):
+        """Command must invoke the module, not an inline -c one-liner."""
         config = _build_hook_config(8484, ["SessionStart"])
         cmd = config["SessionStart"][0]["hooks"][0]["command"]
-        assert "SESSION_ID" in cmd
-        assert "CWD" in cmd
+        assert " -c " not in cmd
+        assert "-m aictl.hook_handler" in cmd
 
 
 class TestIsAictlHook:
-    def test_detects_nested_format(self):
-        """Current nested format {"matcher", "hooks": [...]} is detected."""
+    def test_detects_module_format(self):
+        """Current module-based format is detected."""
+        hook = {"matcher": "", "hooks": [{"type": "command", "command": "python -m aictl.hook_handler --event SessionStart --port 8484"}]}
+        assert _is_aictl_hook(hook)
+
+    def test_detects_old_inline_format(self):
+        """Previous inline -c format with /api/hooks is detected for migration."""
         hook = {"matcher": "", "hooks": [{"type": "command", "command": "python -c '...http://localhost:8484/api/hooks...'"}]}
         assert _is_aictl_hook(hook)
 
     def test_detects_old_flat_format_for_migration(self):
-        """Old flat format (pre-fix) is detected so reinstall can clean it up."""
+        """Legacy flat format (pre-fix) is detected so reinstall can clean it up."""
         hook = {"type": "command", "command": "curl -s http://localhost:8484/api/hooks ..."}
         assert _is_aictl_hook(hook)
 
@@ -237,7 +238,7 @@ def test_force_installs_alongside_user_hooks(self, tmp_settings):
             else:
                 all_cmds.append(h.get("command", ""))
         assert any("my-linter.sh" in c for c in all_cmds)
-        assert any("/api/hooks" in c for c in all_cmds)
+        assert any("aictl.hook_handler" in c for c in all_cmds)
 
     def test_no_conflict_when_only_aictl_hooks_exist(self, tmp_settings):
         """Re-installing when only old aictl hooks exist should succeed without --force."""
@@ -332,17 +333,67 @@ def test_hook_events_match_validator(self):
         )
 
 
+class TestHookHandler:
+    """Tests for the aictl.hook_handler module."""
+
+    def test_enriches_event_and_env(self, monkeypatch):
+        import io
+        from aictl.hook_handler import main
+        monkeypatch.setattr("sys.argv", ["hook_handler", "--event", "PreToolUse", "--port", "9999"])
+        monkeypatch.setattr("sys.stdin", io.StringIO('{"tool": "Bash"}'))
+        monkeypatch.setenv("SESSION_ID", "test-session")
+        monkeypatch.setenv("CWD", "/tmp")
+        captured = io.StringIO()
+        monkeypatch.setattr("sys.stdout", captured)
+        # urlopen will fail (no server) but should not raise
+        main()
+        result = json.loads(captured.getvalue())
+        assert result["event"] == "PreToolUse"
+        assert result["tool"] == "Bash"
+        assert result["session_id"] == "test-session"
+        assert result["cwd"] == "/tmp"
+
+    def test_empty_stdin(self, monkeypatch):
+        """When stdin is a tty (no pipe), handler uses empty dict."""
+        import io
+        from aictl.hook_handler import main
+        monkeypatch.setattr("sys.argv", ["hook_handler", "--event", "SessionStart", "--port", "8484"])
+
+        class FakeTTY(io.StringIO):
+            def isatty(self):
+                return True
+
+        monkeypatch.setattr("sys.stdin", FakeTTY())
+        captured = io.StringIO()
+        monkeypatch.setattr("sys.stdout", captured)
+        main()
+        result = json.loads(captured.getvalue())
+        assert result["event"] == "SessionStart"
+
+    def test_does_not_overwrite_existing_session_id(self, monkeypatch):
+        """If payload already has session_id, don't overwrite with env var."""
+        import io
+        from aictl.hook_handler import main
+        monkeypatch.setattr("sys.argv", ["hook_handler", "--event", "Stop", "--port", "8484"])
+        monkeypatch.setattr("sys.stdin", io.StringIO('{"session_id": "from-payload"}'))
+        monkeypatch.setenv("SESSION_ID", "from-env")
+        captured = io.StringIO()
+        monkeypatch.setattr("sys.stdout", captured)
+        main()
+        result = json.loads(captured.getvalue())
+        assert result["session_id"] == "from-payload"
+
+
 class TestPythonCmd:
     """Hook command uses sys.executable, not the hardcoded 'python3' string."""
 
     def test_does_not_use_bare_python3(self):
-        """Hook command must not start with the bare 'python3' invocation.
+        """Hook command must not start with a bare 'python3' invocation.
 
         The interpreter is sys.executable (a full path, possibly containing
-        'python3' as part of the path, e.g. /usr/bin/python3.11). The old
-        implementation used the bare literal 'python3 -c', which breaks on
-        Windows where 'python3' is not available. We check that the command
-        starts with a quoted full path, not the literal word 'python3'.
+        'python3' as part of the path, e.g. /usr/bin/python3.11). We check
+        that the command starts with a quoted full path, not the literal
+        word 'python3', since 'python3' is not available on Windows.
         """
         import sys
         config = _build_hook_config(8484, ["SessionStart"])