File size: 6,707 Bytes
6686473 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | #!/usr/bin/env python3
"""
Shared AI hook logger — works with Claude Code, Gemini CLI, Codex, Cursor, Copilot.
Reads JSON from stdin, normalizes to common format, appends to .ai-log/session.jsonl
"""
import json
import os
import sys
import subprocess
from datetime import datetime, timezone, timedelta
from pathlib import Path
VN_TZ = timezone(timedelta(hours=7))
def git(cmd):
try:
return subprocess.check_output(cmd, shell=True, text=True, stderr=subprocess.DEVNULL).strip()
except Exception:
return ""
def detect_tool(data: dict) -> str:
"""Detect which AI tool sent this hook event.
Priority:
1. --tool=NAME CLI argument (cross-platform: works in cmd.exe, PowerShell, bash)
2. AI_TOOL_NAME env var (legacy, bash-only when set inline)
3. Heuristics from payload shape
"""
for arg in sys.argv[1:]:
if arg.startswith("--tool="):
return arg.split("=", 1)[1].lower()
tool_env = os.environ.get("AI_TOOL_NAME", "").lower()
if tool_env:
return tool_env
# Heuristics
if "transcript_path" in data:
return "codex"
if data.get("hook_event_name", "").startswith(("Before", "After", "Session", "Pre", "Notification")):
return "gemini"
if data.get("hook_event_name", "")[0:1].islower():
# camelCase event names → Cursor or Copilot
if "workspace_roots" in data:
return "cursor"
if "toolName" in data:
return "copilot"
if "hook_event_name" in data:
return "claude"
return "unknown"
def normalize(data: dict, tool: str) -> dict | None:
"""Normalize tool-specific payload to common log entry."""
event = data.get("hook_event_name") or data.get("event", "")
ts = datetime.now(VN_TZ).isoformat()
# Resolve repo from git origin. When cwd is not a git working tree (or
# origin isn't set), skip the event entirely — these entries can't be
# tied back to a team on the server and would just clutter the pending
# queue forever.
origin = git("git remote get-url origin")
if not origin:
return None
repo = origin.rstrip("/").split("/")[-1]
if repo.endswith(".git"):
repo = repo[:-4]
base = {
"ts": ts,
"tool": tool,
"event": event,
"session_id": (
data.get("session_id") or
data.get("conversation_id") or
data.get("generation_id") or ""
),
"model": data.get("model", ""),
"repo": repo,
"branch": git("git rev-parse --abbrev-ref HEAD"),
"commit": git("git rev-parse --short HEAD"),
"student": git("git config user.email"),
}
if tool == "claude":
prompt = ""
# UserPromptSubmit: prompt is at top level
if event == "UserPromptSubmit":
prompt = data.get("prompt", "")[:1000]
# PostToolUse: extract from tool_input
elif isinstance(data.get("tool_input"), dict):
prompt = data["tool_input"].get("prompt") or data["tool_input"].get("content") or ""
base.update({
"prompt": prompt,
"tool_name": data.get("tool_name", ""),
"tool_input": data.get("tool_input") if event != "UserPromptSubmit" else None,
"tool_response": str(data.get("tool_response", ""))[:500],
})
elif tool == "gemini":
if event == "BeforeAgent":
prompt = data.get("prompt", "")[:1000]
base.update({"prompt": prompt})
else:
req = data.get("request", {})
contents = req.get("contents", [])
prompt = ""
for c in reversed(contents):
for part in c.get("parts", []):
if part.get("text"):
prompt = part["text"][:1000]
break
if prompt:
break
resp = data.get("response", {})
answer = ""
try:
answer = resp["candidates"][0]["content"]["parts"][0]["text"][:500]
except Exception:
pass
base.update({"prompt": prompt, "response_summary": answer})
elif tool == "codex":
base.update({
"prompt": data.get("prompt", "")[:1000],
"turn_id": data.get("turn_id", ""),
"transcript_path": data.get("transcript_path", ""),
})
elif tool == "cursor":
base.update({
"prompt": data.get("prompt", "")[:1000],
"files_context": data.get("attachments", []),
})
elif tool == "copilot":
base.update({
"prompt": data.get("prompt", "")[:1000],
"tool_name": data.get("toolName", ""),
"tool_args": data.get("toolArgs"),
})
# Skip only true noise: no prompt AND no tool-specific payload (tool_input,
# response_summary, tool_response, tool_args, files_context). Previously
# this only checked `prompt`, which dropped Claude Bash/Edit events (their
# tool_input has `command` / `file_path`, not `prompt` or `content`) and
# any Gemini/Cursor/Copilot turn that carried context but no plain prompt.
_PAYLOAD_KEYS = ("prompt", "tool_input", "response_summary",
"tool_response", "tool_args", "files_context")
_LIFECYCLE_EVENTS = ("Stop", "stop", "SessionEnd", "sessionEnd", "AfterModel")
has_payload = any(base.get(k) for k in _PAYLOAD_KEYS)
if not has_payload and event not in _LIFECYCLE_EVENTS:
return None
return base
def main():
# Read stdin as UTF-8 explicitly. On Windows, sys.stdin defaults to the
# system code page (e.g. cp1252), which corrupts non-Latin1 prompts
# (Vietnamese, CJK, emoji) into mojibake. The hook payload is always UTF-8.
raw = sys.stdin.buffer.read().decode("utf-8", errors="replace").strip()
if not raw:
sys.exit(0)
try:
data = json.loads(raw)
except json.JSONDecodeError:
sys.exit(0)
tool = detect_tool(data)
entry = normalize(data, tool)
if not entry:
sys.exit(0)
log_dir = Path(os.environ.get("AI_LOG_DIR", ".ai-log"))
log_dir.mkdir(exist_ok=True)
log_file = log_dir / "session.jsonl"
with open(log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
# Output valid JSON (required by some tools like Gemini)
print(json.dumps({"status": "logged"}))
if __name__ == "__main__":
main()
|