insights-clean.py — provenance-filtered, config-aware rebuild of Claude Code's /insights (only analyzes sessions a human actually drove; gap analysis against your CLAUDE.md/skills/automation config)

insights-clean.py

#!/usr/bin/env python3
"""insights-clean.py — config-aware rebuild of Claude Code /insights.

Fixes vs the built-in:
  1. Provenance filtering: only sessions with real typed user input
     (promptSource=="typed") are analyzed. Observer sessions, Cue/headless
     runs, and SDK-driven sessions are counted but excluded from analysis.
  2. Config-aware suggestions: the report prompt receives CLAUDE.md files,
     the skill inventory, and cue.yaml, and must not suggest what already
     exists.
  3. Honest stats: interactive vs headless vs observer split is explicit.

Usage:
  insights-clean.py --dry-run                # classification stats only, no LLM
  insights-clean.py                          # full run (facets + report)
  insights-clean.py --days 30 --max-new-facets 50
  insights-clean.py --report-only            # reuse cached facets, just regenerate report
"""

import argparse
import json
import os
import re
import subprocess
import sys
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path

PROJECTS_DIR = Path.home() / ".claude" / "projects"
DATA_DIR = Path.home() / ".claude" / "usage-data" / "insights-clean"
META_DIR = DATA_DIR / "meta"
FACETS_DIR = DATA_DIR / "facets"
WORK_DIR = Path("/tmp/insights-clean-work")  # cwd for claude -p calls, excluded from scans

# Project dirs never analyzed (observer/synthetic session homes)
EXCLUDED_DIR_SUBSTRINGS = [
    "claude-mem-observer-sessions",
    "tmp-insights-clean-work",
]

FACET_PROMPT = """Analyze this Claude Code session and extract structured facets.

RESPOND WITH ONLY A VALID JSON OBJECT — no prose, no markdown fences — with exactly these keys:
  "underlying_goal": string — what the USER was trying to achieve
  "goal_categories": object mapping category -> count, categories: coding, debugging, research, writing, automation, configuration, other, warmup_minimal
  "outcome": "achieved" | "partially_achieved" | "not_achieved"
  "user_satisfaction": "happy" | "satisfied" | "likely_satisfied" | "dissatisfied" | "frustrated" | "unknown"
  "claude_helpfulness": "essential" | "very_helpful" | "moderately_helpful" | "slightly_helpful" | "unhelpful"
  "session_type": "single_task" | "multi_task" | "iterative_refinement" | "exploration"
  "friction_counts": object mapping friction type -> count, types: misunderstood_request, wrong_approach, buggy_code, user_rejected_action, excessive_changes, environment_issue
  "friction_detail": string — one sentence on the worst friction, or ""
  "brief_summary": string — 1-2 sentences on what happened
  "user_instructions_to_claude": array of strings — explicit process corrections/preferences the user stated (max 3)

CRITICAL GUIDELINES:
1. goal_categories: count ONLY what the USER explicitly asked for, not Claude's autonomous work.
2. user_satisfaction: base ONLY on explicit user signals in their messages.
3. friction_counts: be specific; empty object if none.
4. If very short or just warmup, use goal_categories {"warmup_minimal": 1}.

SESSION:
"""

def log(msg):
    print(msg, file=sys.stderr, flush=True)

# ---------- session scanning ----------

def list_session_files():
    files = []
    for d in PROJECTS_DIR.iterdir():
        if not d.is_dir():
            continue
        if any(s in d.name for s in EXCLUDED_DIR_SUBSTRINGS):
            continue
        for f in d.glob("*.jsonl"):
            files.append(f)
    return files

def scan_session(path):
    """Light single pass: provenance + counts. Returns metadata dict."""
    typed = 0
    user_msgs = 0
    interruptions = 0
    ts_first = None
    ts_last = None
    cwd = None
    entrypoints = Counter()
    with open(path, "r", errors="replace") as fh:
        for line in fh:
            if '"timestamp"' in line:
                m = re.search(r'"timestamp":"([^"]+)"', line)
                if m:
                    if ts_first is None:
                        ts_first = m.group(1)
                    ts_last = m.group(1)
            if '"type":"user"' not in line:
                continue
            if '"isMeta":true' in line:
                continue
            if '"tool_result"' in line or '"tool_use_id"' in line:
                continue
            user_msgs += 1
            if '"promptSource":"typed"' in line:
                typed += 1
            m = re.search(r'"entrypoint":"([^"]+)"', line)
            if m:
                entrypoints[m.group(1)] += 1
            if cwd is None:
                m = re.search(r'"cwd":"([^"]+)"', line)
                if m:
                    cwd = m.group(1)
            if "[Request interrupted by user" in line:
                interruptions += 1
    dur = 0.0
    if ts_first and ts_last:
        try:
            t0 = datetime.fromisoformat(ts_first.replace("Z", "+00:00"))
            t1 = datetime.fromisoformat(ts_last.replace("Z", "+00:00"))
            dur = (t1 - t0).total_seconds() / 60
        except ValueError:
            pass
    return {
        "session_id": path.stem,
        "project_dir": path.parent.name,
        "cwd": cwd or "",
        "start_time": ts_first or "",
        "duration_minutes": round(dur, 1),
        "user_message_count": user_msgs,
        "typed_message_count": typed,
        "interruptions": interruptions,
        "entrypoints": dict(entrypoints),
        "transcript_mtime": path.stat().st_mtime,
    }

def get_meta(path):
    """Cached scan_session."""
    cache = META_DIR / f"{path.stem}.json"
    mtime = path.stat().st_mtime
    if cache.exists():
        try:
            d = json.load(open(cache))
            if d.get("transcript_mtime") == mtime:
                return d
        except (json.JSONDecodeError, OSError):
            pass
    d = scan_session(path)
    META_DIR.mkdir(parents=True, exist_ok=True)
    json.dump(d, open(cache, "w"), indent=1)
    return d

def classify(meta):
    if "claude-mem" in meta["cwd"]:
        return "observer"
    if meta["typed_message_count"] > 0:
        return "interactive"
    if meta["user_message_count"] >= 2:
        # SDK-driven but multi-prompt: resumed/steered (e.g. via Discord relay)
        return "steered"
    return "headless"

# ---------- transcript condensation for facet extraction ----------

def condense_transcript(path, max_chars=40000):
    parts = []
    with open(path, "r", errors="replace") as fh:
        for line in fh:
            try:
                d = json.loads(line)
            except json.JSONDecodeError:
                continue
            if d.get("isMeta"):
                continue
            t = d.get("type")
            msg = d.get("message")
            if not isinstance(msg, dict):
                continue
            content = msg.get("content")
            if t == "user":
                if isinstance(content, str):
                    parts.append(f"USER: {content.strip()[:2000]}")
                elif isinstance(content, list):
                    for c in content:
                        if isinstance(c, dict) and c.get("type") == "text":
                            parts.append(f"USER: {c.get('text','').strip()[:2000]}")
            elif t == "assistant" and isinstance(content, list):
                for c in content:
                    if not isinstance(c, dict):
                        continue
                    if c.get("type") == "text" and c.get("text", "").strip():
                        parts.append(f"ASSISTANT: {c['text'].strip()[:1500]}")
                    elif c.get("type") == "tool_use":
                        parts.append(f"ASSISTANT uses tool: {c.get('name','?')}")
    text = "\n".join(parts)
    if len(text) > max_chars:
        head, tail = int(max_chars * 0.6), int(max_chars * 0.35)
        text = text[:head] + "\n[... middle truncated ...]\n" + text[-tail:]
    return text

# ---------- LLM calls via headless claude ----------

def claude_query(prompt, model, timeout=300):
    WORK_DIR.mkdir(parents=True, exist_ok=True)
    r = subprocess.run(
        ["claude", "-p", "--model", model],
        input=prompt, capture_output=True, text=True,
        timeout=timeout, cwd=WORK_DIR,
    )
    if r.returncode != 0:
        raise RuntimeError(f"claude -p failed: {r.stderr[:300]}")
    return r.stdout.strip()

def extract_facet(path, meta, model):
    transcript = condense_transcript(path)
    if not transcript.strip():
        return None
    out = claude_query(FACET_PROMPT + transcript, model)
    m = re.search(r"\{[\s\S]*\}", out)
    if not m:
        return None
    try:
        facet = json.loads(m.group(0))
    except json.JSONDecodeError:
        return None
    facet["session_id"] = meta["session_id"]
    facet["project_dir"] = meta["project_dir"]
    facet["start_time"] = meta["start_time"]
    return facet

# ---------- config context ----------

def read_trunc(p, limit=8000):
    try:
        return Path(p).expanduser().read_text(errors="replace")[:limit]
    except OSError:
        return ""

def skill_inventory(project_cwds):
    lines = []
    dirs = [Path.home() / ".claude" / "skills"]
    dirs += [Path(c) / ".claude" / "skills" for c in project_cwds]
    seen = set()
    for d in dirs:
        if not d.is_dir():
            continue
        for sk in sorted(d.iterdir()):
            f = sk / "SKILL.md"
            key = sk.name
            if key in seen or not f.exists():
                continue
            seen.add(key)
            desc = ""
            for line in f.read_text(errors="replace").splitlines()[:15]:
                if line.startswith("description:"):
                    desc = line[len("description:"):].strip()[:200]
                    break
            lines.append(f"- {sk.name}: {desc}")
    return "\n".join(lines)

def gather_config(top_cwds):
    sections = []
    g = read_trunc("~/.claude/CLAUDE.md")
    if g:
        sections.append("## Global CLAUDE.md\n" + g)
    for c in top_cwds[:3]:
        p = Path(c) / "CLAUDE.md"
        if p.exists():
            sections.append(f"## Project CLAUDE.md ({c})\n" + read_trunc(p))
        cue = Path(c) / ".maestro" / "cue.yaml"
        if cue.exists():
            sections.append(f"## Cue automation ({c}/.maestro/cue.yaml)\n" + read_trunc(cue, 4000))
    inv = skill_inventory(top_cwds)
    if inv:
        sections.append("## Installed skills\n" + inv)
    return "\n\n".join(sections)

# ---------- report ----------

REPORT_PROMPT = """You are generating a Claude Code usage insights report for a senior developer.

You are given:
A) STATS — deterministic numbers computed from session transcripts (trust these exactly).
B) FACETS — per-session LLM-extracted summaries of INTERACTIVE sessions only (sessions where the user actually typed). Headless automation and observer sessions were deliberately excluded from analysis.
C) USER CONFIG — the user's actual CLAUDE.md files, installed skills, and cue.yaml automation config.

Write a markdown report with these sections:
# Insights (clean)
## At a glance — 3-4 sentences, honest about sample size.
## Project areas — group the facets into real work threads; give per-thread session counts.
## What worked — concrete workflows that went well, cite specific sessions.
## Friction — recurring problems, only if they appear in 2+ distinct sessions; one-offs go in a single line at the end.
## Gap analysis — THE KEY SECTION. Compare observed behavior against USER CONFIG:
   - Suggest ONLY things NOT already covered by existing CLAUDE.md rules, skills, hooks, or cue.yaml subscriptions.
   - If a session shows the user manually doing something an existing skill/cue already automates, flag that as "you have this automated but did it by hand".
   - If existing config was violated or ignored in sessions, flag it.
   - Never recommend a tool, script, or workflow that does not exist without marking it clearly as "would need to be built".
   - If there are no genuine gaps, say so. An empty gap analysis is a valid result.

Rules: be blunt and concise, no filler praise, no invented statistics, cite session ids (8 chars) where useful. Do not exceed 150 lines.

=== A) STATS ===
{stats}

=== B) FACETS ===
{facets}

=== C) USER CONFIG ===
{config}
"""

def build_report(stats, facets, config, model):
    facet_lines = []
    for f in facets:
        facet_lines.append(json.dumps({k: f.get(k) for k in (
            "session_id", "project_dir", "start_time", "underlying_goal", "outcome",
            "user_satisfaction", "session_type", "friction_counts", "friction_detail",
            "brief_summary", "user_instructions_to_claude")}, ensure_ascii=False))
    prompt = REPORT_PROMPT.format(
        stats=json.dumps(stats, indent=1),
        facets="\n".join(facet_lines),
        config=config,
    )
    return claude_query(prompt, model, timeout=600)

# ---------- main ----------

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--days", type=int, default=30)
    ap.add_argument("--max-new-facets", type=int, default=50)
    ap.add_argument("--dry-run", action="store_true", help="classification stats only, no LLM calls")
    ap.add_argument("--report-only", action="store_true", help="skip facet extraction, use cache")
    ap.add_argument("--parallel", type=int, default=5)
    ap.add_argument("--model-facets", default="claude-haiku-4-5")
    ap.add_argument("--model-report", default="claude-sonnet-5")
    args = ap.parse_args()

    cutoff = datetime.now(timezone.utc) - timedelta(days=args.days)
    files = list_session_files()
    log(f"Scanning {len(files)} session files ...")

    metas = {}
    for i, f in enumerate(files):
        if f.stat().st_mtime < cutoff.timestamp():
            continue
        m = get_meta(f)
        m["_path"] = str(f)
        metas[m["session_id"]] = m
        if (i + 1) % 200 == 0:
            log(f"  scanned {i+1}/{len(files)}")

    classes = Counter()
    interactive = []
    for m in metas.values():
        c = classify(m)
        classes[c] += 1
        if c in ("interactive", "steered") and m["user_message_count"] >= 2 and m["duration_minutes"] >= 1:
            m["class"] = c
            interactive.append(m)
        elif c == "interactive":
            classes["interactive_too_short"] += 1

    interactive.sort(key=lambda m: m["start_time"], reverse=True)
    proj_counts = Counter(m["cwd"] for m in interactive)

    stats = {
        "window_days": args.days,
        "sessions_in_window": len(metas),
        "classification": dict(classes),
        "analyzed_interactive": len(interactive),
        "top_project_cwds": proj_counts.most_common(10),
        "total_interruptions": sum(m["interruptions"] for m in interactive),
        "total_typed_messages": sum(m["typed_message_count"] for m in interactive),
        "total_hours_interactive": round(sum(m["duration_minutes"] for m in interactive) / 60, 1),
        "date_range": [
            min((m["start_time"] for m in interactive), default=""),
            max((m["start_time"] for m in interactive), default=""),
        ],
    }
    log(json.dumps(stats, indent=1))
    if args.dry_run:
        print(json.dumps(stats, indent=1))
        return

    # facet extraction
    FACETS_DIR.mkdir(parents=True, exist_ok=True)
    facets = []
    todo = []
    for m in interactive:
        cache = FACETS_DIR / f"{m['session_id']}.json"
        if cache.exists():
            try:
                facets.append(json.load(open(cache)))
                continue
            except (json.JSONDecodeError, OSError):
                pass
        todo.append(m)

    if not args.report_only:
        todo = todo[: args.max_new_facets]
        log(f"Extracting facets for {len(todo)} sessions ({args.model_facets}, {args.parallel} parallel) ...")
        from concurrent.futures import ThreadPoolExecutor, as_completed
        done = 0
        with ThreadPoolExecutor(max_workers=args.parallel) as ex:
            futs = {ex.submit(extract_facet, Path(m["_path"]), m, args.model_facets): m for m in todo}
            for fut in as_completed(futs):
                m = futs[fut]
                done += 1
                try:
                    f = fut.result()
                except (RuntimeError, subprocess.TimeoutExpired) as e:
                    log(f"  facet failed {m['session_id'][:8]}: {e}")
                    continue
                if f:
                    json.dump(f, open(FACETS_DIR / f"{m['session_id']}.json", "w"), indent=1)
                    facets.append(f)
                log(f"  [{done}/{len(todo)}] {m['session_id'][:8]} {'ok' if f else 'SKIP'}")

    # drop warmups
    facets = [f for f in facets if list(f.get("goal_categories", {}).keys()) != ["warmup_minimal"]]
    stats["facets_used"] = len(facets)

    top_cwds = [c for c, _ in proj_counts.most_common(5) if c]
    config = gather_config(top_cwds)
    log(f"Config context: {len(config)} chars. Generating report ({args.model_report}) ...")
    report = build_report(stats, facets, config, args.model_report)

    ts = datetime.now().strftime("%Y-%m-%d-%H%M%S")
    out = DATA_DIR / f"report-{ts}.md"
    out.write_text(report + "\n\n---\n## Appendix: deterministic stats\n```json\n"
                   + json.dumps(stats, indent=1) + "\n```\n")
    (DATA_DIR / "report.md").write_text(out.read_text())
    print(out)

if __name__ == "__main__":
    main()