Skip to content

WEB-4507: Capture Claude Code skill & agent invocations from hooks #99

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anonpran wants to merge 3 commits into
base: main
Choose a base branch
from
Open

WEB-4507: Capture Claude Code skill & agent invocations from hooks #99

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9b3f6f1
WEB-4507: capture Claude Code skill & agent invocations from hooks
anonpran Jun 3, 2026
f335091
WEB-4507: address Greptile review (trigger namespacing + dedup-able id)
anonpran Jun 3, 2026
6467cff
WEB-4507: trim verbose comments (no functional change)
anonpran Jun 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 115 additions & 5 deletions claude-code/hooks/unbound.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import sys
import json
import os
import glob
import subprocess
from pathlib import Path
from datetime import datetime, timezone
Expand Down Expand Up @@ -401,6 +402,21 @@ def parse_transcript_file(transcript_path: str, user_prompt_timestamp: Optional[
'content': text_content,
'timestamp': entry_timestamp
})
# Skill/agent invocations (transcript is the reliable source, not PostToolUse).
elif (isinstance(content_item, dict)
and content_item.get('type') == 'tool_use'
and content_item.get('name') in ('Skill', 'Task', 'Agent')
and content_item.get('id')): # need an id to dedup
conversation_data['tool_uses'].append({
'type': 'PostToolUse',
'tool_name': content_item.get('name'),
'tool_input': content_item.get('input', {}),
'tool_response': {},
'timestamp': entry_timestamp,
'tool_use_id': content_item.get('id'),
'cwd': entry.get('cwd'),
'git_branch': entry.get('gitBranch'),
})

# Model is captured unconditionally so it survives even on usage-less assistant entries.
turn_model = turn_model or message.get('model')
Expand Down Expand Up @@ -846,7 +862,7 @@ def process_user_prompt_submit(event: Dict, api_key: str) -> Dict:
return transform_response_for_claude_prompt(api_response)


def build_llm_exchange(events: List[Dict], stop_assistant_message: Optional[str] = None, transcript_assistant_messages: Optional[List[str]] = None, model: Optional[str] = None, usage: Optional[Dict] = None) -> Optional[Dict]:
def build_llm_exchange(events: List[Dict], stop_assistant_message: Optional[str] = None, transcript_assistant_messages: Optional[List[str]] = None, model: Optional[str] = None, usage: Optional[Dict] = None, transcript_tool_uses: Optional[List[Dict]] = None) -> Optional[Dict]:
messages = []
assistant_tool_uses = []

Expand All @@ -868,23 +884,57 @@ def build_llm_exchange(events: List[Dict], stop_assistant_message: Optional[str]
prompt = event.get('prompt')
if prompt:
user_prompt = prompt

elif hook_event_name == 'PostToolUse':
tool_name = event.get('tool_name')
# Skills/agents come from the transcript (stable id); skip the id-less PostToolUse duplicate.
if tool_name in ('Skill', 'Task', 'Agent'):
continue
tool_input = event.get('tool_input', {})
tool_response = event.get('tool_response', {})

if 'content' in tool_response and 'content' in tool_input:
if tool_response['content'] == tool_input['content']:
tool_response = {k: v for k, v in tool_response.items() if k != 'content'}

assistant_tool_uses.append({
'type': 'PostToolUse',
'tool_name': tool_name,
'tool_input': tool_input,
'tool_response': tool_response
})


# Transcript-sourced skill/agent invocations: tag trigger (user vs agent) + is_subagent.
for tu in (transcript_tool_uses or []):
if tu.get('is_subagent'):
trigger = 'agent'
else:
skill_name = (tu.get('tool_input') or {}).get('skill') or ''
up = (user_prompt or '').strip()
trigger = 'agent'
if skill_name and up.startswith('/'):
typed_cmd = up[1:].split(None, 1)[0]
# Match full or bare-vs-namespaced name (not a pure suffix match,
# so ns1:deploy vs ns2:deploy isn't mislabelled user-typed).
if typed_cmd and (
typed_cmd == skill_name
or typed_cmd.split(':')[-1] == skill_name
or typed_cmd == skill_name.split(':')[-1]
):
trigger = 'user'
assistant_tool_uses.append({
'type': tu.get('type', 'PostToolUse'),
'tool_name': tu.get('tool_name'),
'tool_input': tu.get('tool_input', {}),
'tool_response': tu.get('tool_response', {}),
'tool_use_id': tu.get('tool_use_id'),
'timestamp': tu.get('timestamp'),
'cwd': tu.get('cwd'),
'git_branch': tu.get('git_branch'),
'trigger': trigger,
'is_subagent': bool(tu.get('is_subagent')),
})

if user_prompt:
messages.append({'role': 'user', 'content': user_prompt})

Expand Down Expand Up @@ -985,6 +1035,59 @@ def cleanup_old_logs():
save_logs(logs[-AUDIT_LOG_TOTAL_LIMIT:])


def collect_subagent_skill_tool_uses(transcript_path: str, user_prompt_timestamp: Optional[str] = None) -> List[Dict]:
"""Collect Skill/agent tool_use blocks from subagent transcripts.

Subagents write to <session>/subagents/agent-*.jsonl (separate from the parent
transcript), so their skills/agents are missed otherwise. Flagged is_subagent.
"""
tool_uses: List[Dict] = []
try:
if not transcript_path or not transcript_path.endswith('.jsonl'):
return tool_uses
subagents_dir = os.path.join(transcript_path[:-len('.jsonl')], 'subagents')
if not os.path.isdir(subagents_dir):
return tool_uses
for sub_file in glob.glob(os.path.join(subagents_dir, '*.jsonl')):
try:
with open(sub_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or ('"Skill"' not in line and '"Task"' not in line and '"Agent"' not in line):
continue
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
if entry.get('type') != 'assistant':
continue
entry_timestamp = entry.get('timestamp')
if user_prompt_timestamp and entry_timestamp and entry_timestamp <= user_prompt_timestamp:
continue
message = entry.get('message', {})
for content_item in message.get('content', []) or []:
if (isinstance(content_item, dict)
and content_item.get('type') == 'tool_use'
and content_item.get('name') in ('Skill', 'Task', 'Agent')
and content_item.get('id')): # need an id to dedup
tool_uses.append({
'type': 'PostToolUse',
'tool_name': content_item.get('name'),
'tool_input': content_item.get('input', {}),
'tool_response': {},
'timestamp': entry_timestamp,
'tool_use_id': content_item.get('id'),
'cwd': entry.get('cwd'),
'git_branch': entry.get('gitBranch'),
'is_subagent': True,
})
Comment thread
anonpran marked this conversation as resolved.
except Exception:
continue
except Exception:
pass
return tool_uses


def process_stop_event(event: Dict, api_key: str):
session_id = event.get('session_id')
transcript_path = event.get('transcript_path')
Expand Down Expand Up @@ -1012,6 +1115,7 @@ def process_stop_event(event: Dict, api_key: str):
transcript_assistant_messages = []
transcript_usage = None
transcript_model = None
transcript_tool_uses = []
if transcript_path and transcript_path != 'undefined' and user_prompt_timestamp:
transcript_data = parse_transcript_file(transcript_path, user_prompt_timestamp)
transcript_assistant_messages = [
Expand All @@ -1020,6 +1124,11 @@ def process_stop_event(event: Dict, api_key: str):
]
transcript_usage = transcript_data.get('usage')
transcript_model = transcript_data.get('model')
transcript_tool_uses = list(transcript_data.get('tool_uses', []))
# Subagents write to separate transcripts the parent Stop never reads — sweep them too.
transcript_tool_uses.extend(
collect_subagent_skill_tool_uses(transcript_path, user_prompt_timestamp)
)

# Prefer the dominant model from the transcript (covers sub-agent turns where
# the cached session model is wrong). Fall back to the audit log otherwise.
Expand All @@ -1031,6 +1140,7 @@ def process_stop_event(event: Dict, api_key: str):
transcript_assistant_messages=transcript_assistant_messages,
model=session_model,
usage=transcript_usage,
transcript_tool_uses=transcript_tool_uses,
)

if exchange:
Expand Down