feat: smart-suggest ROI script + hook tuning + guide updates (Mar 16)
- Add examples/scripts/smart-suggest-roi.py: stdlib-only analyzer correlating suggestion log with session JSONL files to measure command acceptance rate. 4 acceptance signals, tier breakdown, daily trend, --json/--since/--no-sessions CLI. - Tune Aristote smart-suggest hook: tighten 5 over-firing triggers (/tech:commit, /tech:sonarqube, /tech:dupes, /check-conventions a11y, /tech:worktree) - Guide: identity re-injection hook, context engineering maturity grid, code review workflow, 1M context window GA update, Spring Break promo, security audit patterns - Resource evaluations: Nick Tune hooks (3/5), VicKayro security audit (2/5), Karl Mazier CLAUDE.md templates, Paul Rayner ContextFlow, Siddhant agent trace, Andrew Yng context hub, JP Caparas 1M context window Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
d9cff74d71
commit
da8bc09f2d
19 changed files with 1963 additions and 6 deletions
|
|
@ -35,6 +35,11 @@ Perform security audits with isolated context, focusing on vulnerability detecti
|
|||
- [ ] Threat modeling considered
|
||||
- [ ] Security requirements defined
|
||||
- [ ] Principle of least privilege
|
||||
- [ ] Paywall/billing limits enforced server-side (not client-side)
|
||||
- [ ] Subscription status read from DB, not from a client-supplied token or claim
|
||||
- [ ] Payment webhook signatures verified (Stripe `stripe.webhooks.constructEvent`, Paddle equivalent)
|
||||
- [ ] No endpoint bypasses billing verification (e.g., admin routes that skip plan checks)
|
||||
- [ ] No race condition on session/resource creation that could allow free usage beyond limits (CWE-362)
|
||||
|
||||
### A05: Security Misconfiguration
|
||||
- [ ] Default credentials changed
|
||||
|
|
|
|||
|
|
@ -17,6 +17,22 @@ You are a senior application security engineer. Perform a 6-phase security audit
|
|||
|
||||
---
|
||||
|
||||
### Pre-Step: Establish Audit Context
|
||||
|
||||
**Before running any checks**, use `AskUserQuestion` to ask:
|
||||
|
||||
1. **Environment**: Is this code running in production, staging, or local development?
|
||||
2. **Scope**: Full audit or specific areas to prioritize?
|
||||
|
||||
This is critical for accurate findings:
|
||||
- **Local dev**: `DEBUG=True`, CORS `*`, HTTP without TLS, `.env` files — all normal. Do NOT flag as vulnerabilities. Mention in an "Before going to production" informational section instead.
|
||||
- **Staging**: Configs should mirror production. Flag deviations as MEDIUM.
|
||||
- **Production**: Any misconfiguration is a real finding with full severity.
|
||||
|
||||
If the user doesn't answer or is unsure, default to **production** (conservative).
|
||||
|
||||
---
|
||||
|
||||
### Phase 1: Configuration Security (via /security-check)
|
||||
|
||||
Execute all checks from `/security-check` (the `examples/commands/security-check.md` command). This covers:
|
||||
|
|
@ -59,6 +75,23 @@ find . -name ".env*" -not -path "*/node_modules/*" -not -path "*/.git/*" -type f
|
|||
}
|
||||
```
|
||||
|
||||
**Anti-false-positive rule — MANDATORY before reporting any secret finding:**
|
||||
|
||||
Before raising a secrets finding, run these verification commands:
|
||||
|
||||
```bash
|
||||
# 1. Verify .env is actually in .gitignore (if yes, local .env is NOT a finding)
|
||||
grep -n '\.env' .gitignore 2>/dev/null || echo ".env NOT in .gitignore"
|
||||
|
||||
# 2. Verify secrets were actually committed (empty output = no finding)
|
||||
git log --all -p -- '*.env' '*.key' '*.pem' '*.secret' 2>/dev/null | grep -E '^\+.*(password|secret|api_key|token)' | head -20
|
||||
|
||||
# 3. Check git history for provider-specific patterns
|
||||
git log --all -p 2>/dev/null | grep -E '^\+(sk-[a-zA-Z0-9]{20,}|AKIA[A-Z0-9]{16}|ghp_[a-zA-Z0-9]{36})' | head -10
|
||||
```
|
||||
|
||||
Only report a secret finding if you have **concrete proof from these commands**. A `.env` file present locally is not a finding if it's in `.gitignore`. Never report "secrets may be exposed" based on pattern matching alone.
|
||||
|
||||
**Scoring:**
|
||||
- 0 secrets found → +20 points
|
||||
- 1-3 secrets → +10 points
|
||||
|
|
|
|||
78
examples/hooks/bash/identity-reinjection.sh
Executable file
78
examples/hooks/bash/identity-reinjection.sh
Executable file
|
|
@ -0,0 +1,78 @@
|
|||
#!/bin/bash
|
||||
# .claude/hooks/identity-reinjection.sh
|
||||
# Event: UserPromptSubmit
|
||||
# Non-blocking guard: re-injects agent identity if context compaction erased it
|
||||
#
|
||||
# Problem: When Claude compacts context during a long session, agents configured
|
||||
# with a specific role (team lead, reviewer, etc.) can "forget" their identity.
|
||||
# The compacted transcript no longer contains the original system instructions.
|
||||
#
|
||||
# Solution: Store identity in a file. After each user message, check whether the
|
||||
# last assistant response includes the expected identity marker. If not, inject
|
||||
# the identity as additionalContext so the next response re-establishes the role.
|
||||
#
|
||||
# Setup:
|
||||
# 1. Create .claude/agent-identity.txt with your agent's identity instructions
|
||||
# 2. Set IDENTITY_MARKER to a short string that should appear in agent responses
|
||||
# (e.g. "LEAD:", "REVIEWER:", "DEVELOPER:")
|
||||
# 3. Wire this hook to UserPromptSubmit in settings.json
|
||||
#
|
||||
# Output format: UserPromptSubmit additionalContext injection
|
||||
# Properties:
|
||||
# - Silent no-op when identity is present (zero overhead)
|
||||
# - Silent no-op when no identity file configured
|
||||
# - Never blocks — exits 0 in all cases
|
||||
# - Compatible with agent teams (SubagentStart, SubagentStop) and solo sessions
|
||||
#
|
||||
# Based on pattern from Nick Tune: https://nick-tune.me/blog/2026-02-28-hook-driven-dev-workflows-with-claude-code/
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
INPUT=$(cat)
|
||||
|
||||
# Identity file: customize path or override via env
|
||||
IDENTITY_FILE="${CLAUDE_IDENTITY_FILE:-.claude/agent-identity.txt}"
|
||||
IDENTITY_MARKER="${CLAUDE_IDENTITY_MARKER:-}"
|
||||
|
||||
# No-op: identity file not configured
|
||||
if [[ ! -f "$IDENTITY_FILE" ]]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
IDENTITY=$(cat "$IDENTITY_FILE")
|
||||
|
||||
# No-op: empty identity
|
||||
if [[ -z "$IDENTITY" ]]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Default marker: first non-empty, non-comment line of the identity file
|
||||
if [[ -z "$IDENTITY_MARKER" ]]; then
|
||||
IDENTITY_MARKER=$(grep -m1 -v '^#' "$IDENTITY_FILE" | head -c 40 || true)
|
||||
fi
|
||||
|
||||
# Read transcript to check last assistant message
|
||||
TRANSCRIPT_PATH=$(echo "$INPUT" | jq -r '.transcript_path // empty' 2>/dev/null || true)
|
||||
|
||||
if [[ -z "$TRANSCRIPT_PATH" || ! -f "$TRANSCRIPT_PATH" ]]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Extract the last assistant message from the transcript
|
||||
LAST_ASSISTANT=$(jq -r '
|
||||
[.[] | select(.role == "assistant")] | last | .content |
|
||||
if type == "array" then map(select(.type == "text") | .text) | join("") else . end
|
||||
' "$TRANSCRIPT_PATH" 2>/dev/null || true)
|
||||
|
||||
# Identity is intact: no action needed
|
||||
if echo "$LAST_ASSISTANT" | grep -qF "$IDENTITY_MARKER" 2>/dev/null; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Identity marker missing from last response — re-inject
|
||||
# This happens after context compaction strips the original system instructions
|
||||
jq -n \
|
||||
--arg context "[Identity reminder — your role and instructions follow. Resume your role immediately.]\n\n$IDENTITY" \
|
||||
'{"additionalContext": $context}'
|
||||
|
||||
exit 0
|
||||
|
|
@ -27,6 +27,7 @@ Utility scripts for Claude Code power users.
|
|||
| `sync-claude-config.sh` | Sync Claude config files across machines |
|
||||
| `sonnetplan.sh` | Run Claude with Sonnet replacing Opus (cost optimization alias) |
|
||||
| `test-prompt-caching.ts` | Verify Anthropic prompt caching is active (no deps, fetch only) |
|
||||
| `smart-suggest-roi.py` | Analyze acceptance rate of smart-suggest hook suggestions vs session activity |
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
621
examples/scripts/smart-suggest-roi.py
Executable file
621
examples/scripts/smart-suggest-roi.py
Executable file
|
|
@ -0,0 +1,621 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
smart-suggest-roi.py — Analyze acceptance rate of smart-suggest hook suggestions.
|
||||
|
||||
Usage:
|
||||
./smart-suggest-roi.py # Full report
|
||||
./smart-suggest-roi.py --json # Machine-readable JSON
|
||||
./smart-suggest-roi.py --since 7d # Last N days
|
||||
./smart-suggest-roi.py --no-sessions # Suggestion stats only (fast)
|
||||
./smart-suggest-roi.py --log PATH # Custom log path
|
||||
|
||||
Methodology: "Followed" = the suggested command/agent was used later in the
|
||||
same session. Proxy metric — user may have used it independently of the
|
||||
suggestion, or in a different session.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import bisect
|
||||
import json
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier classification (extensible mapping)
|
||||
# ---------------------------------------------------------------------------
|
||||
TIER_MAP = {
|
||||
# Tier 0 — Enforcement (high-stakes, process gates)
|
||||
"pnpm changelog:add": 0,
|
||||
"/pr": 0,
|
||||
"/plan": 0,
|
||||
"/tech:plan": 0,
|
||||
"/tech:pr": 0,
|
||||
"/tech:commit": 0,
|
||||
# Tier 1 — Discovery (specialized workflows rarely triggered organically)
|
||||
"/test-loop": 1,
|
||||
"/retex": 1,
|
||||
"/tech:retex": 1,
|
||||
"/dupes": 1,
|
||||
"/tech:dupes": 1,
|
||||
"/loop": 1,
|
||||
"security-auditor": 1,
|
||||
"/release": 1,
|
||||
"/tech:ralph-loop": 1,
|
||||
"/tech:scaffold": 1,
|
||||
"/tech:sonarqube": 1,
|
||||
"complexity-estimator": 1,
|
||||
"/tech:diagram": 1,
|
||||
"/tech:handoff": 1,
|
||||
"/tech:daily": 1,
|
||||
"/tech:bilan-hebdo": 1,
|
||||
"/tech:worktree": 1,
|
||||
"/tech:sentry-triage": 1,
|
||||
"skill-creator": 1,
|
||||
"/tech:create-release": 1,
|
||||
"/tech:tests": 1,
|
||||
"/tech:diagnose": 1,
|
||||
# Tier 2 — Contextual (common helpers, lower novelty)
|
||||
"code-reviewer": 2,
|
||||
"debugger": 2,
|
||||
"architect-review": 2,
|
||||
"/resume": 2,
|
||||
"/tech:resume": 2,
|
||||
"ui-designer": 2,
|
||||
"requirements-analyst": 2,
|
||||
"backend-architect": 2,
|
||||
"/tech:ship": 2,
|
||||
"/critique-plan": 2,
|
||||
}
|
||||
|
||||
TIER_LABELS = {0: "Tier 0 (Enforcement)", 1: "Tier 1 (Discovery)", 2: "Tier 2 (Contextual)", -1: "Custom"}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_ts(ts_str: str) -> float:
|
||||
"""Parse ISO 8601 timestamp to Unix epoch float."""
|
||||
if not ts_str:
|
||||
return 0.0
|
||||
ts_str = ts_str.rstrip("Z")
|
||||
for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"):
|
||||
try:
|
||||
dt = datetime.strptime(ts_str, fmt).replace(tzinfo=timezone.utc)
|
||||
return dt.timestamp()
|
||||
except ValueError:
|
||||
continue
|
||||
return 0.0
|
||||
|
||||
|
||||
def first_token(cmd: str) -> str:
|
||||
"""Return first whitespace-delimited token (for commands like '/loop [interval]')."""
|
||||
return cmd.split()[0] if cmd else cmd
|
||||
|
||||
|
||||
def get_tier(cmd: str) -> int:
|
||||
"""Classify a command into a tier. Returns -1 for unknown (Custom)."""
|
||||
return TIER_MAP.get(cmd, TIER_MAP.get(first_token(cmd), -1))
|
||||
|
||||
|
||||
def parse_since(since_str: str) -> float:
|
||||
"""Parse '7d', '24h', '30m' into a Unix timestamp cutoff."""
|
||||
unit = since_str[-1]
|
||||
value = int(since_str[:-1])
|
||||
now = datetime.now(tz=timezone.utc).timestamp()
|
||||
if unit == "d":
|
||||
return now - value * 86400
|
||||
if unit == "h":
|
||||
return now - value * 3600
|
||||
if unit == "m":
|
||||
return now - value * 60
|
||||
raise ValueError(f"Unsupported time unit: {unit}. Use d/h/m (e.g. 7d, 24h).")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase 1 — Parse suggestions log
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_suggestions(log_path: Path, since_ts: float = 0.0):
|
||||
"""
|
||||
Returns list of suggestion dicts and skip count.
|
||||
Each dict: {ts, suggested, prompt_len, cmd (first token)}
|
||||
"""
|
||||
suggestions = []
|
||||
skip_count = 0
|
||||
|
||||
if not log_path.exists():
|
||||
return suggestions, skip_count
|
||||
|
||||
with log_path.open("r", encoding="utf-8", errors="replace") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
ts = parse_ts(entry.get("ts", ""))
|
||||
if ts == 0.0:
|
||||
skip_count += 1
|
||||
continue
|
||||
if ts < since_ts:
|
||||
continue
|
||||
suggested = entry.get("suggested", "")
|
||||
if not suggested:
|
||||
skip_count += 1
|
||||
continue
|
||||
suggestions.append({
|
||||
"ts": ts,
|
||||
"suggested": suggested,
|
||||
"cmd": first_token(suggested),
|
||||
"prompt_len": entry.get("prompt_len", 0),
|
||||
})
|
||||
except (json.JSONDecodeError, KeyError, TypeError):
|
||||
skip_count += 1
|
||||
|
||||
suggestions.sort(key=lambda x: x["ts"])
|
||||
return suggestions, skip_count
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase 2 — Build session index & detect acceptance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _read_first_last_ts(path: Path):
|
||||
"""Read first and last timestamp from a session JSONL file efficiently."""
|
||||
first_ts = None
|
||||
last_ts = None
|
||||
session_id = None
|
||||
|
||||
try:
|
||||
with path.open("r", encoding="utf-8", errors="replace") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
ts = parse_ts(entry.get("timestamp", ""))
|
||||
if ts == 0.0:
|
||||
continue
|
||||
if first_ts is None:
|
||||
first_ts = ts
|
||||
session_id = entry.get("sessionId", "")
|
||||
last_ts = ts
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
continue
|
||||
except (PermissionError, OSError):
|
||||
pass
|
||||
|
||||
return first_ts, last_ts, session_id
|
||||
|
||||
|
||||
def build_session_index(projects_dir: Path):
|
||||
"""
|
||||
Walk all project JSONL session files and build a sorted index for lookup.
|
||||
|
||||
Returns:
|
||||
- sessions: list of {start_ts, end_ts, session_id, path} sorted by start_ts
|
||||
- start_ts_list: just start timestamps for bisect
|
||||
"""
|
||||
sessions = []
|
||||
|
||||
if not projects_dir.exists():
|
||||
return sessions, []
|
||||
|
||||
for jsonl_file in projects_dir.glob("*/*.jsonl"):
|
||||
# Skip activity logs and smart-suggest logs (not session files)
|
||||
if "activity-" in jsonl_file.name or "smart-suggest" in jsonl_file.name:
|
||||
continue
|
||||
first_ts, last_ts, session_id = _read_first_last_ts(jsonl_file)
|
||||
if first_ts is None:
|
||||
continue
|
||||
sessions.append({
|
||||
"start_ts": first_ts,
|
||||
"end_ts": last_ts or first_ts,
|
||||
"session_id": session_id,
|
||||
"path": jsonl_file,
|
||||
})
|
||||
|
||||
sessions.sort(key=lambda x: x["start_ts"])
|
||||
start_ts_list = [s["start_ts"] for s in sessions]
|
||||
return sessions, start_ts_list
|
||||
|
||||
|
||||
def find_sessions_for_ts(ts: float, sessions: list, start_ts_list: list, window_before: float = 120.0):
|
||||
"""
|
||||
Find sessions that were active at timestamp ts.
|
||||
A session is "active" if ts is between start and end (+ small buffer).
|
||||
"""
|
||||
if not sessions:
|
||||
return []
|
||||
|
||||
# Binary search: find sessions that started before ts + window_before
|
||||
hi = bisect.bisect_right(start_ts_list, ts + window_before)
|
||||
candidates = sessions[:hi]
|
||||
|
||||
active = []
|
||||
for s in candidates:
|
||||
if s["start_ts"] <= ts + window_before and s["end_ts"] >= ts - 30:
|
||||
active.append(s)
|
||||
return active
|
||||
|
||||
|
||||
def _check_acceptance_in_session(path: Path, cmd_token: str, suggestion_ts: float, time_window: float = 600.0):
|
||||
"""
|
||||
Scan a session JSONL file for evidence the suggested command was followed.
|
||||
|
||||
Acceptance signals (in priority order):
|
||||
1. <command-name>cmd</command-name> in user message content
|
||||
2. Skill tool use with skill = cmd
|
||||
3. Agent tool use with subagent_type = cmd
|
||||
4. cmd appears in next 5 user messages within time_window seconds
|
||||
"""
|
||||
entries_after = []
|
||||
|
||||
try:
|
||||
with path.open("r", encoding="utf-8", errors="replace") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
ts = parse_ts(entry.get("timestamp", ""))
|
||||
if ts >= suggestion_ts:
|
||||
entries_after.append((ts, entry))
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
continue
|
||||
except (PermissionError, OSError):
|
||||
return None # Cannot read file
|
||||
|
||||
if not entries_after:
|
||||
return None # No entries after suggestion — cannot determine
|
||||
|
||||
user_message_count = 0
|
||||
|
||||
for ts, entry in entries_after:
|
||||
msg_type = entry.get("type", "")
|
||||
msg = entry.get("message", {})
|
||||
if not isinstance(msg, dict):
|
||||
continue
|
||||
|
||||
role = msg.get("role", "")
|
||||
content = msg.get("content", "")
|
||||
|
||||
# Signal 1: slash command invocation in user message
|
||||
if msg_type == "user" or role == "user":
|
||||
user_message_count += 1
|
||||
content_str = content if isinstance(content, str) else json.dumps(content)
|
||||
# Check for <command-name> tag
|
||||
if f"<command-name>{cmd_token}</command-name>" in content_str:
|
||||
return True
|
||||
# Check for skill invocation pattern
|
||||
if f'"skill": "{cmd_token}"' in content_str or f"'skill': '{cmd_token}'" in content_str:
|
||||
return True
|
||||
# Text mention in first 5 user messages within window
|
||||
if user_message_count <= 5 and ts - suggestion_ts <= time_window:
|
||||
if cmd_token in content_str:
|
||||
return True
|
||||
|
||||
# Signal 2 & 3: tool use in assistant messages
|
||||
if msg_type == "assistant" or role == "assistant":
|
||||
content_list = content if isinstance(content, list) else []
|
||||
for block in content_list:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
if block.get("type") != "tool_use":
|
||||
continue
|
||||
tool_name = block.get("name", "")
|
||||
tool_input = block.get("input", {}) or {}
|
||||
# Signal 2: Skill tool
|
||||
if tool_name == "Skill" and tool_input.get("skill") == cmd_token:
|
||||
return True
|
||||
# Signal 3: Agent tool
|
||||
if tool_name == "Agent" and tool_input.get("subagent_type") == cmd_token:
|
||||
return True
|
||||
|
||||
return False # No signals found
|
||||
|
||||
|
||||
def compute_acceptance(suggestions: list, sessions: list, start_ts_list: list):
|
||||
"""
|
||||
For each suggestion, find matching sessions and check acceptance.
|
||||
Mutates each suggestion dict in-place, adding 'followed' key.
|
||||
"""
|
||||
for s in suggestions:
|
||||
active = find_sessions_for_ts(s["ts"], sessions, start_ts_list)
|
||||
if not active:
|
||||
s["followed"] = None # No session context
|
||||
continue
|
||||
|
||||
# Check all active sessions — accepted if ANY matches
|
||||
result = False
|
||||
any_data = False
|
||||
for sess in active:
|
||||
check = _check_acceptance_in_session(sess["path"], s["cmd"], s["ts"])
|
||||
if check is True:
|
||||
result = True
|
||||
any_data = True
|
||||
break
|
||||
if check is False:
|
||||
any_data = True
|
||||
# check is None: no data in this file
|
||||
|
||||
if not any_data:
|
||||
s["followed"] = None
|
||||
else:
|
||||
s["followed"] = result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase 3 — Compute stats
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def compute_stats(suggestions: list):
|
||||
"""Build stats dict from annotated suggestions."""
|
||||
stats = {
|
||||
"total": len(suggestions),
|
||||
"sessions_matched": sum(1 for s in suggestions if s.get("followed") is not None),
|
||||
"followed": sum(1 for s in suggestions if s.get("followed") is True),
|
||||
"by_cmd": defaultdict(lambda: {"total": 0, "followed": 0, "unmatched": 0}),
|
||||
"by_tier": defaultdict(lambda: {"total": 0, "followed": 0}),
|
||||
"by_day": defaultdict(lambda: {"total": 0, "followed": 0}),
|
||||
}
|
||||
|
||||
for s in suggestions:
|
||||
cmd = s["cmd"]
|
||||
tier = get_tier(s["suggested"])
|
||||
day = datetime.fromtimestamp(s["ts"], tz=timezone.utc).strftime("%b %d")
|
||||
|
||||
stats["by_cmd"][cmd]["total"] += 1
|
||||
stats["by_tier"][tier]["total"] += 1
|
||||
stats["by_day"][day]["total"] += 1
|
||||
|
||||
if s.get("followed") is True:
|
||||
stats["by_cmd"][cmd]["followed"] += 1
|
||||
stats["by_tier"][tier]["followed"] += 1
|
||||
stats["by_day"][day]["followed"] += 1
|
||||
elif s.get("followed") is None:
|
||||
stats["by_cmd"][cmd]["unmatched"] += 1
|
||||
|
||||
# Compute unique commands
|
||||
stats["unique_cmds"] = len(stats["by_cmd"])
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Output helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def pct(num: int, den: int) -> str:
|
||||
if den == 0:
|
||||
return "n/a"
|
||||
return f"{round(100 * num / den)}%"
|
||||
|
||||
|
||||
def bar(count: int, max_count: int, width: int = 16) -> str:
|
||||
if max_count == 0:
|
||||
return ""
|
||||
filled = round(width * count / max_count)
|
||||
return "█" * filled + " " * (width - filled)
|
||||
|
||||
|
||||
def print_report(stats: dict, suggestions: list, skip_count: int,
|
||||
log_path: Path, projects_dir: Path, no_sessions: bool, since_str: str | None):
|
||||
sep = "═" * 51
|
||||
print(sep)
|
||||
since_label = f" ({since_str})" if since_str else f" ({_date_range(suggestions)})"
|
||||
print(f" Smart-Suggest ROI Report{since_label}")
|
||||
print(sep)
|
||||
|
||||
print()
|
||||
print("Summary")
|
||||
print(f" Suggestions emitted: {stats['total']}")
|
||||
print(f" Unique commands: {stats['unique_cmds']}")
|
||||
|
||||
if not no_sessions:
|
||||
matched = stats["sessions_matched"]
|
||||
total = stats["total"]
|
||||
followed = stats["followed"]
|
||||
print(f" Sessions matched: {matched} / {total} ({pct(matched, total)})")
|
||||
print(f" Followed: {followed} / {matched} ({pct(followed, matched)})")
|
||||
|
||||
# By tier
|
||||
if not no_sessions:
|
||||
print()
|
||||
print(f"{'By Tier':<38} {'followed / total'}")
|
||||
for tier_id in sorted(stats["by_tier"].keys()):
|
||||
t = stats["by_tier"][tier_id]
|
||||
label = TIER_LABELS.get(tier_id, "Custom")
|
||||
rate = pct(t["followed"], t["total"])
|
||||
print(f" {label + ':':34} {rate:<8} {t['followed']:>4} / {t['total']}")
|
||||
|
||||
# Top 10 most suggested
|
||||
by_cmd = stats["by_cmd"]
|
||||
sorted_by_total = sorted(by_cmd.items(), key=lambda x: x[1]["total"], reverse=True)
|
||||
print()
|
||||
print("Top 10 Most Suggested")
|
||||
for cmd, data in sorted_by_total[:10]:
|
||||
rate = f"{pct(data['followed'], data['total'])} followed" if not no_sessions else ""
|
||||
print(f" {data['total']:>4} {cmd:<34} {rate}")
|
||||
|
||||
# Top 10 most followed (only if session data available)
|
||||
if not no_sessions and stats["followed"] > 0:
|
||||
sorted_by_followed = sorted(
|
||||
[(cmd, d) for cmd, d in by_cmd.items() if d["followed"] > 0],
|
||||
key=lambda x: x[1]["followed"],
|
||||
reverse=True,
|
||||
)
|
||||
print()
|
||||
print("Top 10 Most Followed")
|
||||
for cmd, data in sorted_by_followed[:10]:
|
||||
rate = pct(data["followed"], data["total"])
|
||||
print(f" {data['followed']:>4} {cmd:<34} {rate} of {data['total']}")
|
||||
|
||||
# Never followed
|
||||
never = [(cmd, d) for cmd, d in by_cmd.items()
|
||||
if d["followed"] == 0 and d["total"] - d["unmatched"] > 0]
|
||||
if never:
|
||||
print()
|
||||
print("Never Followed (always ignored)")
|
||||
for cmd, data in sorted(never, key=lambda x: x[1]["total"], reverse=True)[:10]:
|
||||
print(f" {cmd:<36} ({data['total']} suggestions)")
|
||||
|
||||
# Daily trend
|
||||
by_day = stats["by_day"]
|
||||
if by_day:
|
||||
print()
|
||||
print("Daily Trend")
|
||||
max_day_total = max(d["total"] for d in by_day.values())
|
||||
for day in sorted(by_day.keys()):
|
||||
d = by_day[day]
|
||||
b = bar(d["total"], max_day_total)
|
||||
followed_str = f" ({d['followed']} followed)" if not no_sessions else ""
|
||||
print(f" {day} {b} {d['total']}{followed_str}")
|
||||
|
||||
print()
|
||||
if not no_sessions:
|
||||
print("Note: \"Followed\" means the suggested command/agent was used later in the")
|
||||
print("same session. Proxy metric — the user may have used it independently of")
|
||||
print("the suggestion, or followed it in a different session.")
|
||||
print()
|
||||
|
||||
if skip_count > 0:
|
||||
print(f" [{skip_count} malformed lines skipped]")
|
||||
|
||||
print(sep)
|
||||
print(f" Log: {log_path}")
|
||||
if not no_sessions:
|
||||
from pathlib import Path as _P
|
||||
project_count = sum(1 for _ in projects_dir.glob("*/"))
|
||||
print(f" Sessions: {projects_dir} ({project_count} projects)")
|
||||
print(sep)
|
||||
|
||||
|
||||
def _date_range(suggestions: list) -> str:
|
||||
if not suggestions:
|
||||
return "no data"
|
||||
first = datetime.fromtimestamp(suggestions[0]["ts"], tz=timezone.utc)
|
||||
last = datetime.fromtimestamp(suggestions[-1]["ts"], tz=timezone.utc)
|
||||
delta = last - first
|
||||
days = max(1, delta.days + 1)
|
||||
return f"{days} days"
|
||||
|
||||
|
||||
def print_json(stats: dict, suggestions: list, skip_count: int):
|
||||
output = {
|
||||
"summary": {
|
||||
"total": stats["total"],
|
||||
"unique_cmds": stats["unique_cmds"],
|
||||
"sessions_matched": stats["sessions_matched"],
|
||||
"followed": stats["followed"],
|
||||
"follow_rate": round(stats["followed"] / stats["sessions_matched"], 3)
|
||||
if stats["sessions_matched"] > 0 else None,
|
||||
},
|
||||
"by_cmd": {
|
||||
cmd: {
|
||||
"total": d["total"],
|
||||
"followed": d["followed"],
|
||||
"unmatched": d["unmatched"],
|
||||
"follow_rate": round(d["followed"] / (d["total"] - d["unmatched"]), 3)
|
||||
if (d["total"] - d["unmatched"]) > 0 else None,
|
||||
}
|
||||
for cmd, d in stats["by_cmd"].items()
|
||||
},
|
||||
"by_tier": {
|
||||
TIER_LABELS.get(t, "Custom"): {
|
||||
"total": d["total"],
|
||||
"followed": d["followed"],
|
||||
"follow_rate": round(d["followed"] / d["total"], 3) if d["total"] > 0 else None,
|
||||
}
|
||||
for t, d in stats["by_tier"].items()
|
||||
},
|
||||
"by_day": dict(stats["by_day"]),
|
||||
"skip_count": skip_count,
|
||||
}
|
||||
print(json.dumps(output, indent=2))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Analyze smart-suggest hook ROI from suggestion and session logs."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--log",
|
||||
type=Path,
|
||||
default=Path.home() / ".claude" / "logs" / "smart-suggest.jsonl",
|
||||
help="Path to smart-suggest.jsonl log (default: ~/.claude/logs/smart-suggest.jsonl)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--projects-dir",
|
||||
type=Path,
|
||||
default=Path.home() / ".claude" / "projects",
|
||||
help="Path to Claude projects directory (default: ~/.claude/projects)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--since",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Filter to last N days/hours/minutes (e.g. 7d, 24h, 30m)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-sessions",
|
||||
action="store_true",
|
||||
help="Skip session scanning — show suggestion stats only (fast mode)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--json",
|
||||
action="store_true",
|
||||
help="Output machine-readable JSON",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
# Resolve since cutoff
|
||||
since_ts = 0.0
|
||||
if args.since:
|
||||
try:
|
||||
since_ts = parse_since(args.since)
|
||||
except ValueError as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Phase 1: parse suggestions
|
||||
suggestions, skip_count = parse_suggestions(args.log, since_ts)
|
||||
|
||||
if not suggestions:
|
||||
print(f"No suggestions found in {args.log}", file=sys.stderr)
|
||||
if since_ts > 0:
|
||||
print(f"(filtered to last {args.since})", file=sys.stderr)
|
||||
sys.exit(0)
|
||||
|
||||
# Phase 2: session index + acceptance (unless --no-sessions)
|
||||
if not args.no_sessions:
|
||||
sessions, start_ts_list = build_session_index(args.projects_dir)
|
||||
compute_acceptance(suggestions, sessions, start_ts_list)
|
||||
else:
|
||||
# Mark all as unmatched so stats are computed correctly
|
||||
for s in suggestions:
|
||||
s["followed"] = None
|
||||
|
||||
# Phase 3: stats
|
||||
stats = compute_stats(suggestions)
|
||||
|
||||
# Output
|
||||
if args.json:
|
||||
print_json(stats, suggestions, skip_count)
|
||||
else:
|
||||
print_report(
|
||||
stats, suggestions, skip_count,
|
||||
args.log, args.projects_dir, args.no_sessions, args.since
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue